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

Ft multiple partition support (#272)

Browse Source 
* Added support for partitioned multiplex

* removed debug line

* added tests regarding the multiplex feature for multiple instances
This commit is contained in:
MichMich
2019-11-07 18:23:09 +02:00
committed by Gary Russell
parent 3bdf7ae8df
commit c4d3e47201
2 changed files with 65 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -16,7 +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;
@@ -39,6 +41,7 @@ import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.DeclarationExceptionEvent;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
@@ -63,6 +66,7 @@ import org.springframework.util.StringUtils;
* @author Soby Chacko
* @author Gary Russell
* @author Oleg Zhurakousky
* @author Michael Michailidis
*/
// @checkstyle:off
public class RabbitExchangeQueueProvisioner
@@ -170,8 +174,26 @@ public class RabbitExchangeQueueProvisioner
else {
String[] provisionedDestinations = Stream
.of(StringUtils.tokenizeToStringArray(name, ",", true, true))
.map(destination -> doProvisionConsumerDestination(destination, group,
properties).getName())
.flatMap(destination -> {
if (properties.isPartitioned() && !ObjectUtils.isEmpty(properties.getInstanceIndexList())) {
List<String> consumerDestinationNames = new ArrayList<>();
for (Integer index : properties.getInstanceIndexList()) {
ExtendedConsumerProperties<RabbitConsumerProperties> temporaryProperties =
new ExtendedConsumerProperties<>(properties.getExtension());
BeanUtils.copyProperties(properties, temporaryProperties);
temporaryProperties.setInstanceIndex(index);
consumerDestinationNames.add(doProvisionConsumerDestination(destination, group,
temporaryProperties).getName());
}
return consumerDestinationNames.stream();
}
else {
return Stream.of(doProvisionConsumerDestination(destination, group,
properties).getName());
}
})
.toArray(String[]::new);
consumerDestination = new RabbitConsumerDestination(
StringUtils.arrayToCommaDelimitedString(provisionedDestinations),

41
spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
View File

@@ -88,6 +88,7 @@ import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerP
import org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchangeQueueProvisioner;
import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSupport;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.cloud.stream.provisioning.ConsumerDestination;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationListener;
import org.springframework.context.ConfigurableApplicationContext;
@@ -417,6 +418,46 @@ public class RabbitBinderTests extends
assertThat(endpoint.isRunning()).isFalse();
}
@Test
public void testMultiplexOnPartitionedConsumer() throws Exception {
final ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties = createConsumerProperties();
RabbitTestSupport.RabbitProxy proxy = new RabbitTestSupport.RabbitProxy();
CachingConnectionFactory cf = new CachingConnectionFactory("localhost",
proxy.getPort());
final RabbitExchangeQueueProvisioner rabbitExchangeQueueProvisioner = new RabbitExchangeQueueProvisioner(cf);
consumerProperties.setMultiplex(true);
consumerProperties.setPartitioned(true);
consumerProperties.setInstanceIndexList(Arrays.asList(1, 2, 3));
final ConsumerDestination consumerDestination = rabbitExchangeQueueProvisioner.provisionConsumerDestination("foo", "boo", consumerProperties);
final String name = consumerDestination.getName();
assertThat(name).isEqualTo("foo.boo-1,foo.boo-2,foo.boo-3");
}
@Test
public void testMultiplexOnPartitionedConsumerWithMultipleDestinations() throws Exception {
final ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties = createConsumerProperties();
RabbitTestSupport.RabbitProxy proxy = new RabbitTestSupport.RabbitProxy();
CachingConnectionFactory cf = new CachingConnectionFactory("localhost",
proxy.getPort());
final RabbitExchangeQueueProvisioner rabbitExchangeQueueProvisioner = new RabbitExchangeQueueProvisioner(cf);
consumerProperties.setMultiplex(true);
consumerProperties.setPartitioned(true);
consumerProperties.setInstanceIndexList(Arrays.asList(1, 2, 3));
final ConsumerDestination consumerDestination = rabbitExchangeQueueProvisioner.provisionConsumerDestination("foo,qaa", "boo", consumerProperties);
final String name = consumerDestination.getName();
assertThat(name).isEqualTo("foo.boo-1,foo.boo-2,foo.boo-3,qaa.boo-1,qaa.boo-2,qaa.boo-3");
}
@Test
public void testConsumerPropertiesWithUserInfrastructureNoBind() throws Exception {
RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource());