GH-162 made SPEL properties consistent

Initial change in RabbitProducerProperties to use Expression as type of SPEL properties rather then String

Resolves #162
Resolves #163

spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java

@@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binder.rabbit.properties;
 import javax.validation.constraints.Min;
 
 import org.springframework.amqp.core.MessageDeliveryMode;
+import org.springframework.expression.Expression;
 
 /**
  * @author Marius Bogoevici
@@ -69,12 +70,12 @@ public class RabbitProducerProperties extends RabbitCommonProperties {
 	/**
 	 * when using a delayed message exchange, a SpEL expression to determine the delay to apply to messages
 	 */
-	private String delayExpression;
+	private Expression delayExpression;
 
 	/**
 	 * a custom routing key when publishing messages; default is the destination name; suffixed by "-partition" when partitioned
 	 */
-	private String routingKeyExpression;
+	private Expression routingKeyExpression;
 
 	/**
 	 * the channel name to which to send publisher confirms (acks) if the connection
@@ -167,19 +168,19 @@ public class RabbitProducerProperties extends RabbitCommonProperties {
 		this.transacted = transacted;
 	}
 
-	public String getDelayExpression() {
+	public Expression getDelayExpression() {
 		return this.delayExpression;
 	}
 
-	public void setDelayExpression(String delayExpression) {
+	public void setDelayExpression(Expression delayExpression) {
 		this.delayExpression = delayExpression;
 	}
 
-	public String getRoutingKeyExpression() {
+	public Expression getRoutingKeyExpression() {
 		return this.routingKeyExpression;
 	}
 
-	public void setRoutingKeyExpression(String routingKeyExpression) {
+	public void setRoutingKeyExpression(Expression routingKeyExpression) {
 		this.routingKeyExpression = routingKeyExpression;
 	}
 

spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitExpressionEvaluatingInterceptor.java

@@ -30,6 +30,7 @@ import org.springframework.util.Assert;
  * Interceptor to evaluate expressions for outbound messages before serialization.
  *
  * @author Gary Russell
+ * @author Oleg Zhurakousky
  * @since 2.0
  *
  */
@@ -54,19 +55,19 @@ public class RabbitExpressionEvaluatingInterceptor implements ChannelInterceptor
 	 * @param delayExpression the delay expression.
 	 * @param evaluationContext the evaluation context.
 	 */
-	public RabbitExpressionEvaluatingInterceptor(String routingKeyExpression, String delayExpression,
+	public RabbitExpressionEvaluatingInterceptor(Expression routingKeyExpression, Expression delayExpression,
 			EvaluationContext evaluationContext) {
 		Assert.isTrue(routingKeyExpression != null || delayExpression != null,
 				"At least one expression is required");
 		Assert.notNull(evaluationContext, "the 'evaluationContext' cannot be null");
 		if (routingKeyExpression != null) {
-			this.routingKeyExpression = PARSER.parseExpression(routingKeyExpression);
+			this.routingKeyExpression = routingKeyExpression;
 		}
 		else {
 			this.routingKeyExpression = null;
 		}
 		if (delayExpression != null) {
-			this.delayExpression = PARSER.parseExpression(delayExpression);
+			this.delayExpression = delayExpression;
 		}
 		else {
 			this.delayExpression = null;

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

@@ -73,6 +73,8 @@ import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.integration.StaticMessageHeaderAccessor;
 import org.springframework.integration.acks.AcknowledgmentCallback;
 import org.springframework.integration.acks.AcknowledgmentCallback.Status;
+import org.springframework.expression.Expression;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
 import org.springframework.integration.amqp.inbound.AmqpMessageSource;
 import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
@@ -112,6 +114,7 @@ import com.rabbitmq.client.Envelope;
  * @author Marius Bogoevici
  * @author Artem Bilan
  * @author Soby Chacko
+ * @author Oleg Zhurakousky
  */
 public class RabbitMessageChannelBinder
 		extends AbstractMessageChannelBinder<ExtendedConsumerProperties<RabbitConsumerProperties>,
@@ -265,7 +268,7 @@ public class RabbitMessageChannelBinder
 		endpoint.setExchangeName(producerDestination.getName());
 		RabbitProducerProperties extendedProperties = producerProperties.getExtension();
 		boolean expressionInterceptorNeeded = expressionInterceptorNeeded(extendedProperties);
-		String routingKeyExpression = extendedProperties.getRoutingKeyExpression();
+		Expression routingKeyExpression = extendedProperties.getRoutingKeyExpression();
 		if (!producerProperties.isPartitioned()) {
 			if (routingKeyExpression == null) {
 				endpoint.setRoutingKey(destination);
@@ -276,21 +279,21 @@ public class RabbitMessageChannelBinder
 							+ RabbitExpressionEvaluatingInterceptor.ROUTING_KEY_HEADER + "']");
 				}
 				else {
-					endpoint.setRoutingKeyExpressionString(routingKeyExpression);
+					endpoint.setRoutingKeyExpression(routingKeyExpression);
 				}
 			}
 		}
 		else {
 			if (routingKeyExpression == null) {
-				endpoint.setRoutingKeyExpressionString(buildPartitionRoutingExpression(destination, false));
+				endpoint.setRoutingKeyExpression(buildPartitionRoutingExpression(destination, false));
 			}
 			else {
 				if (expressionInterceptorNeeded) {
-					endpoint.setRoutingKeyExpressionString(buildPartitionRoutingExpression("headers['"
+					endpoint.setRoutingKeyExpression(buildPartitionRoutingExpression("headers['"
 							+ RabbitExpressionEvaluatingInterceptor.ROUTING_KEY_HEADER + "']", true));
 				}
 				else {
-					endpoint.setRoutingKeyExpressionString(buildPartitionRoutingExpression(routingKeyExpression,
+					endpoint.setRoutingKeyExpression(buildPartitionRoutingExpression(routingKeyExpression.getExpressionString(),
 							true));
 				}
 			}
@@ -301,7 +304,7 @@ public class RabbitMessageChannelBinder
 						+ RabbitExpressionEvaluatingInterceptor.DELAY_HEADER + "']");
 			}
 			else {
-				endpoint.setDelayExpressionString(extendedProperties.getDelayExpression());
+				endpoint.setDelayExpression(extendedProperties.getDelayExpression());
 			}
 		}
 		DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.outboundMapper();
@@ -346,9 +349,9 @@ public class RabbitMessageChannelBinder
 
 	private boolean expressionInterceptorNeeded(RabbitProducerProperties extendedProperties) {
 		return extendedProperties.getRoutingKeyExpression() != null
-					&& extendedProperties.getRoutingKeyExpression().contains("payload")
+					&& extendedProperties.getRoutingKeyExpression().getExpressionString().contains("payload")
 				|| (extendedProperties.getDelayExpression() != null
-					&& extendedProperties.getDelayExpression().contains("payload"));
+					&& extendedProperties.getDelayExpression().getExpressionString().contains("payload"));
 	}
 
 	private void checkConnectionFactoryIsErrorCapable() {
@@ -373,10 +376,11 @@ public class RabbitMessageChannelBinder
 		}
 	}
 
-	private String buildPartitionRoutingExpression(String expressionRoot, boolean rootIsExpression) {
-		return rootIsExpression
+	private Expression buildPartitionRoutingExpression(String expressionRoot, boolean rootIsExpression) {
+		String partitionRoutingExpression = rootIsExpression
 				? expressionRoot + " + '-' + headers['" + BinderHeaders.PARTITION_HEADER + "']"
 				: "'" + expressionRoot + "-' + headers['" + BinderHeaders.PARTITION_HEADER + "']";
+		return new SpelExpressionParser().parseExpression(partitionRoutingExpression);
 	}
 
 	@Override

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

@@ -162,7 +162,7 @@ public class RabbitBinderTests extends
 		ExtendedProducerProperties<RabbitProducerProperties> props = new ExtendedProducerProperties<>(
 				new RabbitProducerProperties());
 		if (testName.getMethodName().equals("testPartitionedModuleSpEL")) {
-			props.getExtension().setRoutingKeyExpression("'part.0'");
+			props.getExtension().setRoutingKeyExpression(spelExpressionParser.parseExpression("'part.0'"));
 		}
 		return props;
 	}
@@ -579,7 +579,7 @@ public class RabbitBinderTests extends
 		producerProperties.setPartitionSelectorClass(TestPartitionSelectorClass.class);
 		producerProperties.setPartitionCount(1);
 		producerProperties.getExtension().setTransacted(true);
-		producerProperties.getExtension().setDelayExpression("42");
+		producerProperties.getExtension().setDelayExpression(spelExpressionParser.parseExpression("42"));
 		producerProperties.setRequiredGroups("prodPropsRequired");
 
 		BindingProperties producerBindingProperties = createProducerBindingProperties(producerProperties);
@@ -1344,7 +1344,7 @@ public class RabbitBinderTests extends
 	public void testRoutingKeyExpression() throws Exception {
 		RabbitTestBinder binder = getBinder();
 		ExtendedProducerProperties<RabbitProducerProperties> producerProperties = createProducerProperties();
-		producerProperties.getExtension().setRoutingKeyExpression("payload.field");
+		producerProperties.getExtension().setRoutingKeyExpression(spelExpressionParser.parseExpression("payload.field"));
 
 		DirectChannel output = createBindableChannel("output", createProducerBindingProperties(producerProperties));
 		output.setBeanName("rkeProducer");
@@ -1381,10 +1381,10 @@ public class RabbitBinderTests extends
 	public void testRoutingKeyExpressionPartitionedAndDelay() throws Exception {
 		RabbitTestBinder binder = getBinder();
 		ExtendedProducerProperties<RabbitProducerProperties> producerProperties = createProducerProperties();
-		producerProperties.getExtension().setRoutingKeyExpression("payload.field");
+		producerProperties.getExtension().setRoutingKeyExpression(spelExpressionParser.parseExpression("payload.field"));
 		// requires delayed message exchange plugin; tested locally
 //		producerProperties.getExtension().setDelayedExchange(true);
-		producerProperties.getExtension().setDelayExpression("1000");
+		producerProperties.getExtension().setDelayExpression(spelExpressionParser.parseExpression("1000"));
 		producerProperties.setPartitionKeyExpression(new ValueExpression<>(0));
 
 		DirectChannel output = createBindableChannel("output", createProducerBindingProperties(producerProperties));

spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java

@@ -299,7 +299,7 @@ public class RabbitBinderModuleTests {
 		RabbitProducerProperties rabbitProducerProperties =
 				(RabbitProducerProperties)((ExtendedPropertiesBinder) rabbitBinder).getExtendedProducerProperties("output");
 
-		assertThat(rabbitProducerProperties.getRoutingKeyExpression()).isEqualTo("fooRoutingKey");
+		assertThat(rabbitProducerProperties.getRoutingKeyExpression().getExpressionString()).isEqualTo("fooRoutingKey");
 		assertThat(rabbitProducerProperties.getBatchSize()).isEqualTo(512);
 
 		RabbitConsumerProperties rabbitConsumerProperties =