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

GH-3957: Add JmsInboundGateway.replyToExpression (#8560)

Browse Source 
* GH-3957: Add JmsInboundGateway.replyToExpression

Fixes https://github.com/spring-projects/spring-integration/issues/3957

Sometimes we cannot use a standard `JmsReplyTo` property for sending replies from the server.
A `DestinationResolver` API does not have access to the request message.

* Introduce a `ChannelPublishingJmsMessageListener.replyToExpression` property to evaluate
a reply destination against request JMS `Message`
* Use this expression only of no `JmsReplyTo` property
* Expose this property on Java DSL level
* To simplify end-user experience with lambda configuration for this property, introduce a `CheckedFunction`
which essentially re-throws exception "sneaky" way

* Fix Javadoc for `CheckedFunction`

* * Fix language in docs
* Fix Javadocs lines length
* Regular `catch` and re-throw in the `CheckedFunction`
This commit is contained in:
Artem Bilan
2023-02-22 15:23:13 -05:00
committed by GitHub
parent 3f99424d93
commit acd8a03d4d
7 changed files with 198 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2023 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.
@@ -30,7 +30,10 @@ import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.log.LogAccessor;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
@@ -42,6 +45,7 @@ import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.jms.support.converter.SimpleMessageConverter;
import org.springframework.jms.support.destination.DestinationResolver;
import org.springframework.jms.support.destination.DynamicDestinationResolver;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessagingException;
@@ -90,12 +94,16 @@ public class ChannelPublishingJmsMessageListener
private DestinationResolver destinationResolver = new DynamicDestinationResolver();
private Expression replyToExpression;
private JmsHeaderMapper headerMapper = new DefaultJmsHeaderMapper();
private BeanFactory beanFactory;
private MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private StandardEvaluationContext evaluationContext;
/**
* Specify whether a JMS reply Message is expected.
* @param expectReply true if a reply is expected.
@@ -261,6 +269,17 @@ public class ChannelPublishingJmsMessageListener
this.destinationResolver = destinationResolver;
}
/**
* Set a SpEL expression to resolve a 'replyTo' destination from a request
* {@link jakarta.jms.Message} as a root evaluation object
* if {@link jakarta.jms.Message#getJMSReplyTo()} is null.
* @param replyToExpression the SpEL expression for 'replyTo' destination.
* @since 6.1
*/
public void setReplyToExpression(Expression replyToExpression) {
this.replyToExpression = replyToExpression;
}
/**
* Provide a {@link MessageConverter} implementation to use when
* converting between JMS Messages and Spring Integration Messages.
@@ -382,6 +401,7 @@ public class ChannelPublishingJmsMessageListener
}
this.gatewayDelegate.afterPropertiesSet();
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(this.beanFactory);
}
protected void start() {
@@ -417,21 +437,22 @@ public class ChannelPublishingJmsMessageListener
/**
* Determine a reply destination for the given message.
* <p> This implementation first checks the boolean 'error' flag which signifies that the reply is an error message.
* If reply is not an error it will first check the JMS Reply-To {@link Destination} of the supplied request message;
* if that is not <code>null</code> it is returned; if it is <code>null</code>, then the configured
* {@link #resolveDefaultReplyDestination default reply destination} is returned; if this too is <code>null</code>,
* It will first check the JMS Reply-To {@link Destination}
* of the supplied request message;
* if that is null, then the configured {@link #replyToExpression} is evaluated
* (if any), then a{@link #resolveDefaultReplyDestination default reply destination}
* is returned; if this too is null,
* then an {@link InvalidDestinationException} is thrown.
* @param request the original incoming JMS message
* @param session the JMS Session to operate on
* @return the reply destination (never <code>null</code>)
* @return the reply destination (never null)
* @throws JMSException if thrown by JMS API methods
* @throws InvalidDestinationException if no {@link Destination} can be determined
* @see #setDefaultReplyDestination
* @see jakarta.jms.Message#getJMSReplyTo()
*/
private Destination getReplyDestination(jakarta.jms.Message request, Session session) throws JMSException {
Destination replyTo = request.getJMSReplyTo();
Destination replyTo = resolveReplyTo(request, session);
if (replyTo == null) {
replyTo = resolveDefaultReplyDestination(session);
if (replyTo == null) {
@@ -440,6 +461,24 @@ public class ChannelPublishingJmsMessageListener
}
}
return replyTo;
}
@Nullable
private Destination resolveReplyTo(jakarta.jms.Message request, Session session) throws JMSException {
Destination replyTo = request.getJMSReplyTo();
if (replyTo == null) {
if (this.replyToExpression != null) {
Object replyToValue = this.replyToExpression.getValue(this.evaluationContext, request);
if (replyToValue instanceof Destination destination) {
return destination;
}
else if (replyToValue instanceof String destinationName) {
return this.destinationResolver.resolveDestinationName(session, destinationName, false);
}
}
}
return replyTo;
}
/**

4
spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java
View File

@@ -103,9 +103,9 @@ public class JmsInboundGateway extends MessagingGatewaySupport implements Orderl
}
/**
* Set to false to prevent listener container shutdown when the endpoint is stopped.
* Set to {@code false} to prevent listener container shutdown when the endpoint is stopped.
* Then, if so configured, any cached consumer(s) in the container will remain.
* Otherwise the shared connection and will be closed and the listener invokers shut
* Otherwise, the shared connection and will be closed and the listener invokers shut
* down; this behavior is new starting with version 5.1. Default: true.
* @param shutdownContainerOnStop false to not shutdown.
* @since 5.1

44
spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsInboundGatewaySpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2023 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.
@@ -19,11 +19,15 @@ package org.springframework.integration.jms.dsl;
import java.util.function.Consumer;
import jakarta.jms.Destination;
import jakarta.jms.Message;
import org.springframework.expression.Expression;
import org.springframework.integration.dsl.MessagingGatewaySpec;
import org.springframework.integration.expression.FunctionExpression;
import org.springframework.integration.jms.ChannelPublishingJmsMessageListener;
import org.springframework.integration.jms.JmsHeaderMapper;
import org.springframework.integration.jms.JmsInboundGateway;
import org.springframework.integration.util.CheckedFunction;
import org.springframework.jms.listener.AbstractMessageListenerContainer;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.jms.support.destination.DestinationResolver;
@@ -137,6 +141,44 @@ public class JmsInboundGatewaySpec<S extends JmsInboundGatewaySpec<S>>
return _this();
}
/**
* Set a SpEL expression to resolve a 'replyTo' destination from a request {@link jakarta.jms.Message}
* as a root evaluation object if {@link jakarta.jms.Message#getJMSReplyTo()} is null.
* @param replyToExpression the SpEL expression for 'replyTo' destination.
* @return the spec.
* @since 6.1
* @see ChannelPublishingJmsMessageListener#setReplyToExpression(Expression)
*/
public S replyToExpression(String replyToExpression) {
return replyToExpression(PARSER.parseExpression(replyToExpression));
}
/**
* Set a function to resolve a 'replyTo' destination from a request {@link jakarta.jms.Message}
* as a root evaluation object if {@link jakarta.jms.Message#getJMSReplyTo()} is null.
* @param replyToFunction the function for 'replyTo' destination.
* @return the spec.
* @since 6.1
* @see ChannelPublishingJmsMessageListener#setReplyToExpression(Expression)
*/
public S replyToFunction(CheckedFunction<Message, ?> replyToFunction) {
return replyToExpression(new FunctionExpression<>(replyToFunction.unchecked()));
}
/**
* Set a SpEL expression to resolve a 'replyTo' destination from a request {@link jakarta.jms.Message}
* as a root evaluation object if {@link jakarta.jms.Message#getJMSReplyTo()} is null.
* @param replyToExpression the SpEL expression for 'replyTo' destination.
* @return the spec.
* @since 6.1
* @see ChannelPublishingJmsMessageListener#setReplyToExpression(Expression)
*/
public S replyToExpression(Expression replyToExpression) {
this.target.getListener().setReplyToExpression(replyToExpression);
return _this();
}
/**
* @param messageConverter the messageConverter.
* @return the spec.

25
spring-integration-jms/src/test/java/org/springframework/integration/jms/dsl/JmsTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2023 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.
@@ -22,6 +22,8 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import jakarta.jms.JMSException;
import jakarta.jms.TextMessage;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.ListableBeanFactory;
@@ -137,6 +139,9 @@ public class JmsTests extends ActiveMQMultiContextTests {
@Autowired
private CountDownLatch redeliveryLatch;
@Autowired
JmsTemplate jmsTemplate;
@Test
public void testPollingFlow() {
this.controlBus.send("@'integerMessageSource.inboundChannelAdapter'.start()");
@@ -264,6 +269,19 @@ public class JmsTests extends ActiveMQMultiContextTests {
this.jmsMessageDrivenRedeliveryFlowContainer.stop();
}
@Test
public void customReplyToHeader() throws JMSException {
this.jmsTemplate.send("jmsPipelineTest", session -> {
TextMessage message = session.createTextMessage("test data");
message.setStringProperty("myReplyTo", "replyToQueue");
return message;
});
jakarta.jms.Message replyMessage = this.jmsTemplate.receive("replyToQueue");
assertThat(replyMessage).isNotNull();
assertThat(replyMessage.getBody(String.class)).isEqualTo("TEST DATA");
}
@MessagingGateway(defaultRequestChannel = "controlBus.input")
private interface ControlBusGateway {
@@ -278,7 +296,9 @@ public class JmsTests extends ActiveMQMultiContextTests {
@Bean
public JmsTemplate jmsTemplate() {
return new JmsTemplate(connectionFactory);
JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
jmsTemplate.setReceiveTimeout(10_000);
return jmsTemplate;
}
@Bean(name = PollerMetadata.DEFAULT_POLLER)
@@ -422,6 +442,7 @@ public class JmsTests extends ActiveMQMultiContextTests {
}))
.requestDestination("jmsPipelineTest")
.replyToFunction(message -> message.getStringProperty("myReplyTo"))
.configureListenerContainer(c ->
c.transactionManager(mock(PlatformTransactionManager.class))))
.filter(payload -> !"junk".equals(payload))