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

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<RabbitConsumerProperties> properties) {

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<RabbitProducerProperties> createProducerProperties() {
-		return new ExtendedProducerProperties<>(new RabbitProducerProperties());
+		ExtendedProducerProperties<RabbitProducerProperties> 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());