Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4131: Add delayExpression to AMQP Outbounds

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4131

Specify an expression on the outbound endpoints to set the `x-delay` header when
using the RabbitMQ Delayed Message Exchange plugin.

Polishing

- PR Comments
- Add `setDelay`
- Port `FunctionExpression` from DSL
- Add `SupplierExpression`

Javadoc Fixes

Use ValueExpression for delay

* Simple polishing for `SupplierExpression`
* Mention plain `delay` property in the `amqp.adoc`
This commit is contained in:
Gary Russell
2016-10-06 13:39:28 -04:00
committed by Artem Bilan
parent c31a96d4cb
commit 9673a02c7d
16 changed files with 544 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParser.java
View File

@@ -81,6 +81,8 @@ public class AmqpOutboundChannelAdapterParser extends AbstractOutboundChannelAda
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "confirm-ack-channel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "confirm-nack-channel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "return-channel");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "delay-expression",
"delayExpressionString");
return builder.getBeanDefinition();
}

2
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParser.java
View File

@@ -100,6 +100,8 @@ public class AmqpOutboundGatewayParser extends AbstractConsumerEndpointParser {
mapperBuilder.setFactoryMethod("outboundMapper");
IntegrationNamespaceUtils.configureHeaderMapper(element, builder, parserContext, mapperBuilder,
null);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "delay-expression",
"delayExpressionString");
return builder;
}

56
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java
View File

@@ -31,6 +31,7 @@ import org.springframework.expression.Expression;
import org.springframework.integration.amqp.support.AmqpHeaderMapper;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.expression.ValueExpression;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
@@ -77,6 +78,10 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin
private volatile ConnectionFactory connectionFactory;
private volatile Expression delayExpression;
private volatile ExpressionEvaluatingMessageProcessor<Integer> delayGenerator;
private volatile boolean running;
public void setHeaderMapper(AmqpHeaderMapper headerMapper) {
@@ -188,6 +193,44 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin
this.lazyConnect = lazyConnect;
}
/**
* Set the value to set in the {@code x-delay} header when using the
* RabbitMQ delayed message exchange plugin. By default, the {@link AmqpHeaders#DELAY}
* header (if present) is mapped; setting the delay here overrides that value.
* @param delay the delay.
* @since 5.0
*/
public void setDelay(int delay) {
this.delayExpression = new ValueExpression<>(delay);
}
/**
* Set the SpEL expression to calculate the {@code x-delay} header when using the
* RabbitMQ delayed message exchange plugin. By default, the {@link AmqpHeaders#DELAY}
* header (if present) is mapped; setting the expression here overrides that value.
* @param delayExpression the expression.
* @since 5.0
*/
public void setDelayExpression(Expression delayExpression) {
this.delayExpression = delayExpression;
}
/**
* Set the SpEL expression to calculate the {@code x-delay} header when using the
* RabbitMQ delayed message exchange plugin. By default, the {@link AmqpHeaders#DELAY}
* header (if present) is mapped; setting the expression here overrides that value.
* @param delayExpression the expression.
* @since 5.0
*/
public void setDelayExpressionString(String delayExpression) {
if (delayExpression == null) {
this.delayExpression = null;
}
else {
this.delayExpression = EXPRESSION_PARSER.parseExpression(delayExpression);
}
}
protected final void setConnectionFactory(ConnectionFactory connectionFactory) {
this.connectionFactory = connectionFactory;
}
@@ -284,6 +327,13 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin
Assert.state(this.confirmNackChannel == null || nullChannel != null,
"A 'confirmCorrelationExpression' is required when specifying a 'confirmNackChannel'");
}
if (this.delayExpression != null) {
this.delayGenerator = new ExpressionEvaluatingMessageProcessor<Integer>(this.delayExpression,
Integer.class);
if (beanFactory != null) {
this.delayGenerator.setBeanFactory(beanFactory);
}
}
endpointInit();
}
@@ -365,6 +415,12 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin
return routingKey;
}
protected void addDelayProperty(Message<?> message, org.springframework.amqp.core.Message amqpMessage) {
if (this.delayGenerator != null) {
amqpMessage.getMessageProperties().setDelay(this.delayGenerator.processMessage(message));
}
}
protected Message<?> buildReplyMessage(MessageConverter converter,
org.springframework.amqp.core.Message amqpReplyMessage) {
Object replyObject = converter.fromMessage(amqpReplyMessage);

2
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java
View File

@@ -101,6 +101,7 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint
MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter();
org.springframework.amqp.core.Message amqpMessage = MappingUtils.mapMessage(requestMessage, converter,
getHeaderMapper(), getDefaultDeliveryMode());
addDelayProperty(requestMessage, amqpMessage);
((RabbitTemplate) this.amqpTemplate).send(exchangeName, routingKey, amqpMessage, correlationData);
}
else {
@@ -120,6 +121,7 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint
MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter();
org.springframework.amqp.core.Message amqpMessage = MappingUtils.mapMessage(requestMessage, converter,
getHeaderMapper(), getDefaultDeliveryMode());
addDelayProperty(requestMessage, amqpMessage);
org.springframework.amqp.core.Message amqpReplyMessage =
((RabbitTemplate) this.amqpTemplate).sendAndReceive(exchangeName, routingKey, amqpMessage,
correlationData);

7
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java
View File

@@ -60,10 +60,11 @@ public class AsyncAmqpOutboundGateway extends AbstractAmqpOutboundEndpoint {
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
org.springframework.amqp.core.Message amqpMessage = MappingUtils.mapMessage(requestMessage,
this.messageConverter, getHeaderMapper(), getDefaultDeliveryMode());
addDelayProperty(requestMessage, amqpMessage);
RabbitMessageFuture future = this.template.sendAndReceive(generateExchangeName(requestMessage),
generateRoutingKey(requestMessage),
MappingUtils.mapMessage(requestMessage, this.messageConverter, getHeaderMapper(),
getDefaultDeliveryMode()));
generateRoutingKey(requestMessage), amqpMessage);
future.addCallback(new FutureCallback(requestMessage));
CorrelationData correlationData = generateCorrelationData(requestMessage);
if (correlationData != null && future.getConfirm() != null) {

14
spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-5.0.xsd
View File

@@ -542,6 +542,20 @@ property set to TRUE.
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="delay-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
A SpEL expression that evaluates to the value that will be set on the 'x-delay' header
when using the RabbitMQ Delayed Message Exchange plugin. Takes precedence over any
'AmqpHeaders.DELAY' set on the outbound message.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.messaging.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="integration:smartLifeCycleAttributeGroup"/>
</xsd:complexType>

1
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml
View File

@@ -25,6 +25,7 @@
exchange-name="outboundchanneladapter.test.1"
default-delivery-mode="NON_PERSISTENT"
lazy-connect="false"
delay-expression="42"
mapped-request-headers="foo*"/>
<bean id="customHeaderMapper" class="org.mockito.Mockito" factory-method="mock">

3
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java
View File

@@ -124,6 +124,9 @@ public class AmqpOutboundChannelAdapterParserTests {
AmqpOutboundEndpoint.class);
assertNotNull(TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode"));
assertFalse(TestUtils.getPropertyValue(endpoint, "lazyConnect", Boolean.class));
assertEquals("42",
TestUtils.getPropertyValue(endpoint, "delayExpression", org.springframework.expression.Expression.class)
.getExpressionString());
Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate");
amqpTemplateField.setAccessible(true);

2
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml
View File

@@ -15,6 +15,7 @@
exchange-name="si.test.exchange"
routing-key="si.test.binding"
amqp-template="amqpTemplate"
delay-expression="42"
auto-startup="false"
order="5"
return-channel="returnChannel">
@@ -92,6 +93,7 @@
exchange-name="si.test.exchange"
routing-key="si.test.binding"
async-template="asyncTemplate"
delay-expression="42"
auto-startup="false"
order="5"
return-channel="returnChannel">

3
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java
View File

@@ -92,6 +92,9 @@ public class AmqpOutboundGatewayParserTests {
assertEquals(Long.valueOf(777), sendTimeout);
assertTrue(TestUtils.getPropertyValue(gateway, "lazyConnect", Boolean.class));
assertEquals("42",
TestUtils.getPropertyValue(gateway, "delayExpression", org.springframework.expression.Expression.class)
.getExpressionString());
}
@SuppressWarnings("rawtypes")

77
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java
View File

@@ -16,27 +16,39 @@
package org.springframework.integration.amqp.outbound;
import static org.hamcrest.Matchers.equalTo;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import static org.mockito.BDDMockito.willAnswer;
import static org.mockito.BDDMockito.willDoNothing;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.AsyncRabbitTemplate;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.scheduling.TaskScheduler;
/**
* @author Gary Russell
@@ -45,6 +57,61 @@ import org.springframework.messaging.MessageHeaders;
*/
public class OutboundEndpointTests {
@Test
public void testDelayExpression() {
ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
RabbitTemplate amqpTemplate = spy(new RabbitTemplate(connectionFactory));
AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(amqpTemplate);
willDoNothing()
.given(amqpTemplate).send(anyString(), anyString(), any(Message.class), any(CorrelationData.class));
willAnswer(invocation -> invocation.getArgumentAt(2, Message.class))
.given(amqpTemplate)
.sendAndReceive(anyString(), anyString(), any(Message.class), any(CorrelationData.class));
endpoint.setExchangeName("foo");
endpoint.setRoutingKey("bar");
endpoint.setDelayExpressionString("42");
endpoint.setBeanFactory(mock(BeanFactory.class));
endpoint.afterPropertiesSet();
endpoint.handleMessage(new GenericMessage<>("foo"));
ArgumentCaptor<Message> captor = ArgumentCaptor.forClass(Message.class);
verify(amqpTemplate).send(eq("foo"), eq("bar"), captor.capture(), any(CorrelationData.class));
assertThat(captor.getValue().getMessageProperties().getDelay(), equalTo(42));
endpoint.setExpectReply(true);
endpoint.setOutputChannel(new NullChannel());
endpoint.handleMessage(new GenericMessage<>("foo"));
verify(amqpTemplate).sendAndReceive(eq("foo"), eq("bar"), captor.capture(), any(CorrelationData.class));
assertThat(captor.getValue().getMessageProperties().getDelay(), equalTo(42));
endpoint.setDelay(23);
endpoint.setRoutingKey("baz");
endpoint.afterPropertiesSet();
endpoint.handleMessage(new GenericMessage<>("foo"));
verify(amqpTemplate).sendAndReceive(eq("foo"), eq("baz"), captor.capture(), any(CorrelationData.class));
assertThat(captor.getValue().getMessageProperties().getDelay(), equalTo(23));
}
@Test
public void testAsyncDelayExpression() {
ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
AsyncRabbitTemplate amqpTemplate = spy(new AsyncRabbitTemplate(new RabbitTemplate(connectionFactory),
new SimpleMessageListenerContainer(connectionFactory), "replyTo"));
amqpTemplate.setTaskScheduler(mock(TaskScheduler.class));
AsyncAmqpOutboundGateway gateway = new AsyncAmqpOutboundGateway(amqpTemplate);
willAnswer(
invocation -> amqpTemplate.new RabbitMessageFuture("foo", invocation.getArgumentAt(2, Message.class)))
.given(amqpTemplate).sendAndReceive(anyString(), anyString(), any(Message.class));
gateway.setExchangeName("foo");
gateway.setRoutingKey("bar");
gateway.setDelayExpressionString("42");
gateway.setBeanFactory(mock(BeanFactory.class));
gateway.setOutputChannel(new NullChannel());
gateway.afterPropertiesSet();
ArgumentCaptor<Message> captor = ArgumentCaptor.forClass(Message.class);
gateway.handleMessage(new GenericMessage<>("foo"));
verify(amqpTemplate).sendAndReceive(eq("foo"), eq("bar"), captor.capture());
assertThat(captor.getValue().getMessageProperties().getDelay(), equalTo(42));
}
@Test
public void testHeaderMapperWinsAdapter() {
ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
@@ -52,10 +119,10 @@ public class OutboundEndpointTests {
AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(amqpTemplate);
final AtomicReference<Message> amqpMessage =
new AtomicReference<Message>();
doAnswer(invocation -> {
willAnswer(invocation -> {
amqpMessage.set((Message) invocation.getArguments()[2]);
return null;
}).when(amqpTemplate).send(anyString(), anyString(), any(Message.class),
}).given(amqpTemplate).send(anyString(), anyString(), any(Message.class),
any(CorrelationData.class));
org.springframework.messaging.Message<?> message = MessageBuilder.withPayload("foo")
.setHeader(MessageHeaders.CONTENT_TYPE, "bar")
@@ -76,10 +143,10 @@ public class OutboundEndpointTests {
endpoint.setHeaderMapper(mapper);
final AtomicReference<Message> amqpMessage =
new AtomicReference<Message>();
doAnswer(invocation -> {
willAnswer(invocation -> {
amqpMessage.set((Message) invocation.getArguments()[2]);
return null;
}).when(amqpTemplate)
}).given(amqpTemplate)
.doSendAndReceiveWithTemporary(anyString(), anyString(), any(Message.class), any(CorrelationData.class));
org.springframework.messaging.Message<?> message = MessageBuilder.withPayload("foo")
.setHeader(MessageHeaders.CONTENT_TYPE, "bar")

185
spring-integration-core/src/main/java/org/springframework/integration/expression/FunctionExpression.java Normal file
View File

@@ -0,0 +1,185 @@
/*
* Copyright 2014-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.expression;
import java.util.function.Function;
import org.springframework.core.convert.TypeDescriptor;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.EvaluationException;
import org.springframework.expression.Expression;
import org.springframework.expression.TypedValue;
import org.springframework.expression.common.ExpressionUtils;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.util.Assert;
/**
* An {@link Expression} that simply invokes {@link Function#apply(Object)} on its
* provided {@link Function}.
* <p>
* This is a powerful alternative to the SpEL, when Java 8 and its Lambda support is in use.
* <p>
* If the target component has support for an {@link Expression} property,
* a {@link FunctionExpression} can be specified instead of a
* {@link org.springframework.expression.spel.standard.SpelExpression}
* as an alternative to evaluate the value from the Lambda, rather than runtime SpEL resolution.
* <p>
* The {@link FunctionExpression} is 'read-only', hence only {@link #getValue} operations
* are allowed.
* Any {@link #setValue} operations and {@link #getValueType} related operations
* throw {@link EvaluationException}.
*
* @param <S> The evaluation context root object type.
*
* @author Artem Bilan
* @author Gary Russell
* @since 5.0
*/
public class FunctionExpression<S> implements Expression {
private final Function<S, ?> function;
private final EvaluationContext defaultContext = new StandardEvaluationContext();
private final EvaluationException readOnlyException;
public FunctionExpression(Function<S, ?> function) {
Assert.notNull(function, "'function' must not be null.");
this.function = function;
this.readOnlyException = new EvaluationException(getExpressionString(),
"FunctionExpression is a 'read only' Expression implementation");
}
@Override
public Object getValue() throws EvaluationException {
return this.function.apply(null);
}
@Override
@SuppressWarnings("unchecked")
public Object getValue(Object rootObject) throws EvaluationException {
return this.function.apply((S) rootObject);
}
@Override
public <T> T getValue(Class<T> desiredResultType) throws EvaluationException {
return getValue(this.defaultContext, desiredResultType);
}
@Override
public <T> T getValue(Object rootObject, Class<T> desiredResultType) throws EvaluationException {
return getValue(this.defaultContext, rootObject, desiredResultType);
}
@Override
public Object getValue(EvaluationContext context) throws EvaluationException {
return getValue(context.getRootObject().getValue());
}
@Override
public Object getValue(EvaluationContext context, Object rootObject) throws EvaluationException {
return getValue(rootObject);
}
@Override
public <T> T getValue(EvaluationContext context, Class<T> desiredResultType) throws EvaluationException {
return ExpressionUtils.convertTypedValue(context, new TypedValue(getValue(context)), desiredResultType);
}
@Override
public <T> T getValue(EvaluationContext context, Object rootObject, Class<T> desiredResultType)
throws EvaluationException {
return ExpressionUtils.convertTypedValue(context, new TypedValue(getValue(rootObject)), desiredResultType);
}
@Override
public Class<?> getValueType() throws EvaluationException {
throw this.readOnlyException;
}
@Override
public Class<?> getValueType(Object rootObject) throws EvaluationException {
throw this.readOnlyException;
}
@Override
public Class<?> getValueType(EvaluationContext context) throws EvaluationException {
throw this.readOnlyException;
}
@Override
public Class<?> getValueType(EvaluationContext context, Object rootObject) throws EvaluationException {
throw this.readOnlyException;
}
@Override
public TypeDescriptor getValueTypeDescriptor() throws EvaluationException {
throw this.readOnlyException;
}
@Override
public TypeDescriptor getValueTypeDescriptor(Object rootObject) throws EvaluationException {
throw this.readOnlyException;
}
@Override
public TypeDescriptor getValueTypeDescriptor(EvaluationContext context) throws EvaluationException {
throw this.readOnlyException;
}
@Override
public TypeDescriptor getValueTypeDescriptor(EvaluationContext context, Object rootObject)
throws EvaluationException {
throw this.readOnlyException;
}
@Override
public void setValue(EvaluationContext context, Object value) throws EvaluationException {
throw this.readOnlyException;
}
@Override
public void setValue(Object rootObject, Object value) throws EvaluationException {
throw this.readOnlyException;
}
@Override
public void setValue(EvaluationContext context, Object rootObject, Object value) throws EvaluationException {
throw this.readOnlyException;
}
@Override
public boolean isWritable(EvaluationContext context) throws EvaluationException {
return false;
}
@Override
public boolean isWritable(EvaluationContext context, Object rootObject) throws EvaluationException {
return false;
}
@Override
public boolean isWritable(Object rootObject) throws EvaluationException {
return false;
}
@Override
public String getExpressionString() {
return this.function.toString();
}
}

184
spring-integration-core/src/main/java/org/springframework/integration/expression/SupplierExpression.java Normal file
View File

@@ -0,0 +1,184 @@
/*
* Copyright 2014-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.expression;
import org.boon.core.Supplier;
import org.springframework.core.convert.TypeDescriptor;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.EvaluationException;
import org.springframework.expression.Expression;
import org.springframework.expression.TypedValue;
import org.springframework.expression.common.ExpressionUtils;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.util.Assert;
/**
* An {@link Expression} that simply invokes {@link Supplier#get()} on its
* provided {@link Supplier}.
* <p>
* This is a powerful alternative to the SpEL, when Java 8 and its Lambda support is in use.
* <p>
* If the target component has support for an {@link Expression} property,
* a {@link SupplierExpression} can be specified instead of a
* {@link org.springframework.expression.spel.standard.SpelExpression}
* as an alternative to evaluate the value from the Lambda, rather than runtime SpEL resolution.
* <p>
* The {@link SupplierExpression} is 'read-only', hence only {@link #getValue} operations
* are allowed.
* Any {@link #setValue} operations and {@link #getValueType} related operations
* throw {@link EvaluationException}.
*
* @param <T> The type the Supplier will return.
*
* @author Artem Bilan
* @author Gary Russell
* @since 5.0
*/
public class SupplierExpression<T> implements Expression {
private final Supplier<T> supplier;
private final EvaluationContext defaultContext = new StandardEvaluationContext();
private final EvaluationException readOnlyException;
public SupplierExpression(Supplier<T> supplier) {
Assert.notNull(supplier, "'function' must not be null.");
this.supplier = supplier;
this.readOnlyException = new EvaluationException(getExpressionString(),
"SupplierExpression is a 'read only' Expression implementation");
}
@Override
public Object getValue() throws EvaluationException {
return this.supplier.get();
}
@Override
public Object getValue(Object rootObject) throws EvaluationException {
return getValue();
}
@Override
public <C> C getValue(Class<C> desiredResultType) throws EvaluationException {
return getValue(this.defaultContext, desiredResultType);
}
@Override
public <C> C getValue(Object rootObject, Class<C> desiredResultType) throws EvaluationException {
return getValue(this.defaultContext, rootObject, desiredResultType);
}
@Override
public Object getValue(EvaluationContext context) throws EvaluationException {
return getValue(context.getRootObject().getValue());
}
@Override
public Object getValue(EvaluationContext context, Object rootObject) throws EvaluationException {
return getValue(rootObject);
}
@Override
public <C> C getValue(EvaluationContext context, Class<C> desiredResultType) throws EvaluationException {
return ExpressionUtils.convertTypedValue(context, new TypedValue(getValue(context)), desiredResultType);
}
@Override
public <C> C getValue(EvaluationContext context, Object rootObject, Class<C> desiredResultType)
throws EvaluationException {
return ExpressionUtils.convertTypedValue(context, new TypedValue(getValue(rootObject)), desiredResultType);
}
@Override
public Class<?> getValueType() throws EvaluationException {
throw this.readOnlyException;
}
@Override
public Class<?> getValueType(Object rootObject) throws EvaluationException {
throw this.readOnlyException;
}
@Override
public Class<?> getValueType(EvaluationContext context) throws EvaluationException {
throw this.readOnlyException;
}
@Override
public Class<?> getValueType(EvaluationContext context, Object rootObject) throws EvaluationException {
throw this.readOnlyException;
}
@Override
public TypeDescriptor getValueTypeDescriptor() throws EvaluationException {
throw this.readOnlyException;
}
@Override
public TypeDescriptor getValueTypeDescriptor(Object rootObject) throws EvaluationException {
throw this.readOnlyException;
}
@Override
public TypeDescriptor getValueTypeDescriptor(EvaluationContext context) throws EvaluationException {
throw this.readOnlyException;
}
@Override
public TypeDescriptor getValueTypeDescriptor(EvaluationContext context, Object rootObject)
throws EvaluationException {
throw this.readOnlyException;
}
@Override
public void setValue(EvaluationContext context, Object value) throws EvaluationException {
throw this.readOnlyException;
}
@Override
public void setValue(Object rootObject, Object value) throws EvaluationException {
throw this.readOnlyException;
}
@Override
public void setValue(EvaluationContext context, Object rootObject, Object value) throws EvaluationException {
throw this.readOnlyException;
}
@Override
public boolean isWritable(EvaluationContext context) throws EvaluationException {
return false;
}
@Override
public boolean isWritable(EvaluationContext context, Object rootObject) throws EvaluationException {
return false;
}
@Override
public boolean isWritable(Object rootObject) throws EvaluationException {
return false;
}
@Override
public String getExpressionString() {
return this.supplier.toString();
}
}

4
spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionEvaluatingMessageProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -69,7 +69,7 @@ public class ExpressionEvaluatingMessageProcessor<T> extends AbstractMessageProc
*/
@Override
public T processMessage(Message<?> message) {
return this.evaluateExpression(this.expression, message, this.expectedType);
return evaluateExpression(this.expression, message, this.expectedType);
}
@Override

7
src/reference/asciidoc/amqp.adoc
View File

@@ -1149,7 +1149,14 @@ To configure a default user id for outbound messages, configure it on a `RabbitT
Similarly, to set the user id property on replies, inject an appropriately configured template into the inbound gateway.
See the http://docs.spring.io/spring-amqp/reference/html/_reference.html#template-user-id[Spring AMQP documentation] for more information.
[[amqp-delay]]
=== Delayed Message Exchange
Spring AMQP supports the http://docs.spring.io/spring-amqp/reference/html/_reference.html#delayed-message-exchange[RabbitMQ Delayed Message Exchange Plugin].
For inbound messages, the `x-delay` header is mapped to the `AmqpHeaders.RECEIVED_DELAY` header.
Setting the `AMQPHeaders.DELAY` header will cause the corresponding `x-delay` header to be set in outbound messages.
You can also specify the `delay` and `delayExpression` properties on outbound endpoints (`delay-expression` when using XML configuration).
This takes precedence over the `AmqpHeaders.DELAY` header.
[[amqp-channels]]
=== AMQP Backed Message Channels

5
src/reference/asciidoc/whats-new.adoc
View File

@@ -55,3 +55,8 @@ See <<stream-reading>> for more information.
The `BarrierMessageHandler` now supports a discard channel to which late-arriving trigger messages are sent.
See <<barrier>> for more information.
==== AMQP Changes
The AMQP outbound endpoints now support setting a delay expression for when using the RabbitMQ Delayed Message Exchange plugin.
See <<amqp-delay>> for more information.