From 03db5430d9bd6f644dea12c91bf07ae8d1b18587 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 13 Feb 2017 17:39:57 -0500 Subject: [PATCH] Fix RK Expression for Partitioned Streams Fixes #43 The final routing key expression was incorrect when the stream is partitioned and the user supplied a base for the routing key expression. Make RK Check Method Name More Expressive --- .../rabbit/RabbitMessageChannelBinder.java | 21 ++++++++----- .../binder/rabbit/RabbitBinderTests.java | 31 ++++++++++++++----- 2 files changed, 37 insertions(+), 15 deletions(-) diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index 5d067482d..8ef7a8c69 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -18,9 +18,6 @@ package org.springframework.cloud.stream.binder.rabbit; import java.util.Arrays; -import com.rabbitmq.client.AMQP; -import com.rabbitmq.client.Envelope; - import org.springframework.amqp.core.MessagePostProcessor; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder; @@ -40,6 +37,7 @@ import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPos import org.springframework.amqp.support.postprocessor.GZipPostProcessor; import org.springframework.boot.autoconfigure.amqp.RabbitProperties; import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder; +import org.springframework.cloud.stream.binder.BinderHeaders; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; @@ -64,6 +62,9 @@ import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; import org.springframework.util.StringUtils; +import com.rabbitmq.client.AMQP; +import com.rabbitmq.client.Envelope; + /** * A {@link org.springframework.cloud.stream.binder.Binder} implementation backed by RabbitMQ. * @author Mark Fisher @@ -93,6 +94,8 @@ public class RabbitMessageChannelBinder private final RabbitProperties rabbitProperties; + private final RabbitExchangeQueueProvisioner provisioningProvider; + private ConnectionFactory connectionFactory; private MessagePostProcessor decompressingPostProcessor = new DelegatingDecompressingPostProcessor(); @@ -107,8 +110,6 @@ public class RabbitMessageChannelBinder private RabbitExtendedBindingProperties extendedBindingProperties = new RabbitExtendedBindingProperties(); - RabbitExchangeQueueProvisioner provisioningProvider; - public RabbitMessageChannelBinder(ConnectionFactory connectionFactory, RabbitProperties rabbitProperties, RabbitExchangeQueueProvisioner provisioningProvider) { super(true, new String[0], provisioningProvider); @@ -198,10 +199,10 @@ public class RabbitMessageChannelBinder } else { if (routingKeyExpression == null) { - endpoint.setRoutingKeyExpressionString(buildPartitionRoutingExpression(destination)); + endpoint.setRoutingKeyExpressionString(buildPartitionRoutingExpression(destination, false)); } else { - endpoint.setRoutingKeyExpressionString(buildPartitionRoutingExpression(routingKeyExpression)); + endpoint.setRoutingKeyExpressionString(buildPartitionRoutingExpression(routingKeyExpression, true)); } } if (extendedProperties.getDelayExpression() != null) { @@ -217,6 +218,12 @@ public class RabbitMessageChannelBinder return endpoint; } + private String buildPartitionRoutingExpression(String expressionRoot, boolean rootIsExpression) { + return rootIsExpression + ? expressionRoot + " + '-' + headers['" + BinderHeaders.PARTITION_HEADER + "']" + : "'" + expressionRoot + "-' + headers['" + BinderHeaders.PARTITION_HEADER + "']"; + } + @Override protected MessageProducer createConsumerEndpoint(ConsumerDestination consumerDestination, String group, ExtendedConsumerProperties properties) { diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java index d1c766086..86fda69de 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java @@ -16,6 +16,11 @@ package org.springframework.cloud.stream.binder.rabbit; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + import java.util.HashMap; import java.util.List; import java.util.Map; @@ -23,11 +28,11 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.zip.Deflater; -import com.rabbitmq.http.client.domain.QueueInfo; import org.aopalliance.aop.Advice; import org.apache.commons.logging.Log; import org.junit.Rule; import org.junit.Test; +import org.junit.rules.TestName; import org.mockito.ArgumentCaptor; import org.springframework.amqp.core.AcknowledgeMode; @@ -73,10 +78,7 @@ import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.MessagingException; import org.springframework.messaging.support.GenericMessage; -import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.Mockito.spy; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; +import com.rabbitmq.http.client.domain.QueueInfo; /** * @author Mark Fisher @@ -93,6 +95,9 @@ public class RabbitBinderTests extends @Rule public RabbitTestSupport rabbitAvailableRule = new RabbitTestSupport(true); + @Rule + public TestName testName = new TestName(); + @Override protected RabbitTestBinder getBinder() { if (testBinder == null) { @@ -108,7 +113,12 @@ public class RabbitBinderTests extends @Override protected ExtendedProducerProperties createProducerProperties() { - return new ExtendedProducerProperties<>(new RabbitProducerProperties()); + ExtendedProducerProperties props = new ExtendedProducerProperties<>( + new RabbitProducerProperties()); + if (testName.getMethodName().equals("testPartitionedModuleSpEL")) { + props.getExtension().setRoutingKeyExpression("'part.0'"); + } + return props; } @Override @@ -386,8 +396,7 @@ public class RabbitBinderTests extends producerBinding = binder.bindProducer("props.0", channel, producerProperties); endpoint = extractEndpoint(producerBinding); - assertThat(TestUtils.getPropertyValue(endpoint, "routingKeyExpression", SpelExpression.class) - .getExpressionString()).isEqualTo("'props.0-' + headers['partition']"); + assertThat(getEndpointRouting(endpoint)).isEqualTo("'props.0-' + headers['partition']"); assertThat(TestUtils.getPropertyValue(endpoint, "delayExpression", SpelExpression.class) .getExpressionString()).isEqualTo("42"); mode = TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode", MessageDeliveryMode.class); @@ -1060,6 +1069,12 @@ public class RabbitBinderTests extends return CLASS_UNDER_TEST_NAME; } + @Override + protected void checkRkExpressionForPartitionedModuleSpEL(Object endpoint) { + assertThat(getEndpointRouting(endpoint)) + .contains(getExpectedRoutingBaseDestination("'part.0'", "test") + " + '-' + headers['partition']"); + } + @Override public Spy spyOn(final String queue) { final RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());