From 2feca1b8744de124121720ab999da7ca2c398706 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Mon, 9 Jul 2018 22:29:32 +0200 Subject: [PATCH] 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 --- .../properties/RabbitProducerProperties.java | 13 +++++----- ...RabbitExpressionEvaluatingInterceptor.java | 7 +++--- .../rabbit/RabbitMessageChannelBinder.java | 24 +++++++++++-------- .../binder/rabbit/RabbitBinderTests.java | 10 ++++---- .../integration/RabbitBinderModuleTests.java | 2 +- 5 files changed, 31 insertions(+), 25 deletions(-) diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java index d6297c019..bfa848797 100644 --- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java +++ b/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; } diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitExpressionEvaluatingInterceptor.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitExpressionEvaluatingInterceptor.java index e3067be72..2d8efe2ee 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitExpressionEvaluatingInterceptor.java +++ b/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; 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 cf6e16416..fb310e2b9 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 @@ -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, @@ -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 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 952fac654..8c8c7cba4 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 @@ -162,7 +162,7 @@ public class RabbitBinderTests extends ExtendedProducerProperties 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 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 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)); diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java index 3fe4bf5a5..5dfc01d50 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java +++ b/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 =