SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-149 Code restructuring/refactoring

Browse Source 
Resolves #150
This commit is contained in:
Oleg Zhurakousky
2018-05-21 12:04:52 +02:00
parent 4a4c88f228
commit a8bf0e8a09
2 changed files with 13 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java
View File

@@ -16,10 +16,9 @@
package org.springframework.cloud.stream.binder.rabbit.provisioning;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Stream;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -58,6 +57,7 @@ import org.springframework.util.StringUtils;
*
* @author Soby Chacko
* @author Gary Russell
* @author Oleg Zhurakousky
*/
public class RabbitExchangeQueueProvisioner implements ApplicationListener<DeclarationExceptionEvent>,
ProvisioningProvider<ExtendedConsumerProperties<RabbitConsumerProperties>,
@@ -131,19 +131,17 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
@Override
public ConsumerDestination provisionConsumerDestination(String name, String group,
ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
ConsumerDestination consumerDestination;
if (!properties.isMultiplex()) {
return doProvisionConsumerDestination(name, group, properties);
consumerDestination = doProvisionConsumerDestination(name, group, properties);
}
else {
String[] destinations = StringUtils.commaDelimitedListToStringArray(name);
List<String> queues = new ArrayList<>();
for (String destination : destinations) {
ConsumerDestination dest = doProvisionConsumerDestination(destination.trim(), group, properties);
queues.add(dest.getName());
}
return new RabbitConsumerDestination(
StringUtils.arrayToCommaDelimitedString(queues.toArray(new String[queues.size()])), null);
String[] provisionedDestinations = Stream.of(StringUtils.tokenizeToStringArray(name, ",", true, true))
.map(destination -> doProvisionConsumerDestination(destination, group, properties).getName())
.toArray(String[]::new);
consumerDestination = new RabbitConsumerDestination(StringUtils.arrayToCommaDelimitedString(provisionedDestinations), null);
}
return consumerDestination;
}
private ConsumerDestination doProvisionConsumerDestination(String name, String group,
@@ -498,20 +496,14 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
public void cleanAutoDeclareContext(ConsumerDestination destination,
ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties) {
synchronized (this.autoDeclareContext) {
String[] names = new String[] { destination.getName() };
if (consumerProperties.isMultiplex()) {
names = StringUtils.commaDelimitedListToStringArray(destination.getName());
}
for (int i = 0; i < names.length; i++) {
names[i] = names[i].trim();
}
for (String name : names) {
Stream.of(StringUtils.tokenizeToStringArray(destination.getName(), ",", true, true)).forEach(name -> {
name = name.trim();
removeSingleton(name + ".binding");
removeSingleton(name);
String dlq = name + ".dlq";
removeSingleton(dlq + ".binding");
removeSingleton(dlq);
}
});
}
}

11
spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java
View File

@@ -361,16 +361,7 @@ public class RabbitMessageChannelBinder
listenerContainer.setRecoveryInterval(properties.getExtension().getRecoveryInterval());
listenerContainer.setTxSize(properties.getExtension().getTxSize());
listenerContainer.setTaskExecutor(new SimpleAsyncTaskExecutor(consumerDestination.getName() + "-"));
String[] queues;
if (properties.isMultiplex()) {
queues = StringUtils.commaDelimitedListToStringArray(destination);
}
else {
queues = new String[] { destination };
}
for (int i = 0; i < queues.length; i++) {
queues[i] = queues[i].trim();
}
String[] queues = StringUtils.tokenizeToStringArray(destination, ",", true, true);
listenerContainer.setQueueNames(queues);
listenerContainer.setAfterReceivePostProcessors(this.decompressingPostProcessor);
listenerContainer.setMessagePropertiesConverter(