From 72f7c723920d0e10f5d6ac38f4231b61eabe52ec Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 12 Nov 2019 15:08:01 -0500 Subject: [PATCH] GH-3107: Add errorOnTimeout for TcpInboundGateway Fixes https://github.com/spring-projects/spring-integration/issues/3107 The `MessagingGatewaySupport` has an `errorOnTimeout` option to throw a `MessageTimeoutException` when downstream reply doesn't come back in time for configured reply timeout * Expose an `errorOnTimeout` option as a `TcpInboundGateway` ctor property * Add new factory methods into a `Tcp` factory for Java DSL * Ensure a property works as expected in the `IpIntegrationTests` * Document a new option * Add a setter for MessagingGatewaySupport.errorOnTimeout option * Expose an `errorOnTimeout` option on the DSL's `MessagingGatewaySpec` making all the out-of-the-box inbound gateways possible to react to the `MessageTimeoutException` when no reply during reply timeout * Propagate properly `errorOnTimeout` in the `JmsInboundGateway` * Modify docs respectively * Improve docs about `errorOnTimeout` --- .../integration/dsl/MessagingGatewaySpec.java | 16 +++++ .../gateway/MessagingGatewaySupport.java | 14 +++- .../integration/ip/dsl/Tcp.java | 6 +- .../integration/ip/tcp/TcpInboundGateway.java | 55 +++++++++------- .../ip/dsl/IpIntegrationTests.java | 64 ++++++++++++------- .../ChannelPublishingJmsMessageListener.java | 4 ++ .../integration/jms/JmsInboundGateway.java | 8 +++ .../integration/jms/dsl/JmsTests.java | 46 +++++++++---- src/reference/asciidoc/endpoint-summary.adoc | 4 ++ src/reference/asciidoc/ip.adoc | 2 +- 10 files changed, 160 insertions(+), 59 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/MessagingGatewaySpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/MessagingGatewaySpec.java index f572e5601d..90b9996346 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/MessagingGatewaySpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/MessagingGatewaySpec.java @@ -154,6 +154,22 @@ public abstract class MessagingGatewaySpec, return _this(); } + /** + * If errorOnTimeout is true, construct an instance that will send an + * {@link org.springframework.messaging.support.ErrorMessage} with a + * {@link org.springframework.integration.MessageTimeoutException} payload to the error channel + * if a reply is expected but none is received. If no error channel is configured, + * the {@link org.springframework.integration.MessageTimeoutException} will be thrown. + * @param errorOnTimeout true to create the error message on reply timeout. + * @return the spec + * @since 5.2.2 + * @see MessagingGatewaySupport#setErrorOnTimeout + */ + public S errorOnTimeout(boolean errorOnTimeout) { + this.target.setErrorOnTimeout(errorOnTimeout); + return _this(); + } + /** * An {@link InboundMessageMapper} to use. * @param requestMapper the requestMapper. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java index 0b87b39218..43d70a4570 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java @@ -88,7 +88,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint private final Object replyMessageCorrelatorMonitor = new Object(); - private final boolean errorOnTimeout; + private boolean errorOnTimeout; private final AtomicLong messageCount = new AtomicLong(); @@ -139,6 +139,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint * configured, the {@link MessageTimeoutException} will be thrown. * @param errorOnTimeout true to create the error message. * @since 4.2 + * @see #setErrorOnTimeout */ public MessagingGatewaySupport(boolean errorOnTimeout) { MessagingTemplate template = new MessagingTemplate(); @@ -149,6 +150,17 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint this.errorOnTimeout = errorOnTimeout; } + /** + * If errorOnTimeout is true, construct an instance that will send an + * {@link ErrorMessage} with a {@link MessageTimeoutException} payload to the error + * channel if a reply is expected but none is received. If no error channel is + * configured, the {@link MessageTimeoutException} will be thrown. + * @param errorOnTimeout true to create the error message on reply timeout. + * @since 5.2.2 + */ + public void setErrorOnTimeout(boolean errorOnTimeout) { + this.errorOnTimeout = errorOnTimeout; + } /** * Set the request channel. diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/Tcp.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/Tcp.java index 1992c66093..861aadb164 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/Tcp.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/Tcp.java @@ -24,6 +24,8 @@ import org.springframework.integration.ip.tcp.connection.AbstractConnectionFacto * * @author Gary Russell * @author Tim Ysewyn + * @author Artem Bilan + * * @since 5.0 * */ @@ -31,7 +33,6 @@ public final class Tcp { /** * Boolean indicating the connection factory should use NIO. - * * @deprecated This isn't used anymore within the framework and will be removed in a future release. */ @Deprecated @@ -40,7 +41,6 @@ public final class Tcp { /** * Boolean indicating the connection factory should not use NIO * (default). - * * @deprecated This isn't used anymore within the framework and will be removed in a future release. */ @Deprecated @@ -124,6 +124,7 @@ public final class Tcp { */ public static TcpInboundChannelAdapterSpec inboundAdapter( AbstractConnectionFactorySpec connectionFactorySpec) { + return new TcpInboundChannelAdapterSpec(connectionFactorySpec); } @@ -163,6 +164,7 @@ public final class Tcp { */ public static TcpOutboundChannelAdapterSpec outboundAdapter( AbstractConnectionFactorySpec connectionFactorySpec) { + return new TcpOutboundChannelAdapterSpec(connectionFactorySpec); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java index dc8f59b45e..466729aae8 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java @@ -49,32 +49,39 @@ import org.springframework.util.Assert; * inbound / outbound channel adapters should be used. * * @author Gary Russell + * @author Artem Bilan + * * @since 2.0 */ public class TcpInboundGateway extends MessagingGatewaySupport implements TcpListener, TcpSender, ClientModeCapable, OrderlyShutdownCapable { - private volatile AbstractServerConnectionFactory serverConnectionFactory; + /** + * A default retry interval in milliseconds - {@value #DEFAULT_RETRY_INTERVAL}. + */ + public static final long DEFAULT_RETRY_INTERVAL = 60000; - private volatile AbstractClientConnectionFactory clientConnectionFactory; + private final Map connections = new ConcurrentHashMap<>(); - private final Map connections = new ConcurrentHashMap(); + private final AtomicInteger activeCount = new AtomicInteger(); - private volatile boolean isClientMode; + private AbstractServerConnectionFactory serverConnectionFactory; - private volatile boolean isSingleUse; + private AbstractClientConnectionFactory clientConnectionFactory; - private volatile long retryInterval = 60000; + private boolean isClientMode; - private volatile ScheduledFuture scheduledFuture; + private boolean isSingleUse; - private volatile ClientModeConnectionManager clientModeConnectionManager; + private long retryInterval = DEFAULT_RETRY_INTERVAL; private volatile boolean active; - private volatile boolean shuttingDown; + private volatile ClientModeConnectionManager clientModeConnectionManager; - private final AtomicInteger activeCount = new AtomicInteger(); + private volatile ScheduledFuture scheduledFuture; + + private volatile boolean shuttingDown; @Override public boolean onMessage(Message message) { @@ -117,7 +124,7 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements } private boolean doOnMessage(Message message) { - Message reply = this.sendAndReceiveMessage(message); + Message reply = sendAndReceiveMessage(message); if (reply == null) { if (logger.isDebugEnabled()) { logger.debug("null reply received for " + message + " nothing to send"); @@ -144,13 +151,15 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements } private void publishNoConnectionEvent(Message message, String connectionId) { - AbstractConnectionFactory cf = this.serverConnectionFactory != null ? this.serverConnectionFactory - : this.clientConnectionFactory; + AbstractConnectionFactory cf = + this.serverConnectionFactory != null + ? this.serverConnectionFactory + : this.clientConnectionFactory; ApplicationEventPublisher applicationEventPublisher = cf.getApplicationEventPublisher(); if (applicationEventPublisher != null) { applicationEventPublisher.publishEvent( - new TcpConnectionFailedCorrelationEvent(this, connectionId, - new MessagingException(message, "Connection not found to process reply."))); + new TcpConnectionFailedCorrelationEvent(this, connectionId, + new MessagingException(message, "Connection not found to process reply."))); } } @@ -163,7 +172,6 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements /** * Must be {@link AbstractClientConnectionFactory} or {@link AbstractServerConnectionFactory}. - * * @param connectionFactory the Connection Factory */ public void setConnectionFactory(AbstractConnectionFactory connectionFactory) { @@ -192,6 +200,7 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements public void removeDeadConnection(TcpConnection connection) { this.connections.remove(connection.getConnectionId()); } + @Override public String getComponentType() { return "ip:tcp-inbound-gateway"; @@ -221,11 +230,11 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements this.clientConnectionFactory.start(); } if (this.isClientMode) { - ClientModeConnectionManager manager = new ClientModeConnectionManager( - this.clientConnectionFactory); + ClientModeConnectionManager manager = + new ClientModeConnectionManager(this.clientConnectionFactory); this.clientModeConnectionManager = manager; - Assert.state(this.getTaskScheduler() != null, "Client mode requires a task scheduler"); - this.scheduledFuture = this.getTaskScheduler().scheduleAtFixedRate(manager, this.retryInterval); + Assert.state(getTaskScheduler() != null, "Client mode requires a task scheduler"); + this.scheduledFuture = getTaskScheduler().scheduleAtFixedRate(manager, this.retryInterval); } } } @@ -272,8 +281,9 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements } /** - * @param retryInterval - * the retryInterval to set + * Configure a retry interval. + * Defaults to {@link #DEFAULT_RETRY_INTERVAL}. + * @param retryInterval the retryInterval to set */ public void setRetryInterval(long retryInterval) { this.retryInterval = retryInterval; @@ -307,4 +317,5 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements this.stop(); return this.activeCount.get(); } + } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java index 4b9dc2da53..7a9dbf9d83 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java @@ -24,8 +24,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import org.aopalliance.intercept.MethodInterceptor; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; @@ -34,6 +33,7 @@ import org.springframework.context.ApplicationListener; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.integration.MessageTimeoutException; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.core.MessagingTemplate; @@ -60,7 +60,8 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; +import org.springframework.util.ReflectionUtils; /** * @author Gary Russell @@ -69,7 +70,7 @@ import org.springframework.test.context.junit4.SpringRunner; * @since 5.0 * */ -@RunWith(SpringRunner.class) +@SpringJUnitConfig @DirtiesContext public class IpIntegrationTests { @@ -106,7 +107,7 @@ public class IpIntegrationTests { private AtomicBoolean adviceCalled; @Test - public void testTcpAdapters() { + void testTcpAdapters() { ApplicationEventPublisher publisher = e -> { }; AbstractServerConnectionFactory server = Tcp.netServer(0).backlog(2).soTimeout(5000).id("server").get(); assertThat(server.getComponentName()).isEqualTo("server"); @@ -133,7 +134,7 @@ public class IpIntegrationTests { } @Test - public void testTcpGateways() { + void testTcpGateways() { TestingUtilities.waitListening(this.server1, null); this.client1.stop(); this.client1.setPort(this.server1.getPort()); @@ -142,12 +143,12 @@ public class IpIntegrationTests { MessagingTemplate messagingTemplate = new MessagingTemplate(this.clientTcpFlowInput); assertThat(messagingTemplate.convertSendAndReceive("foo", String.class)).isEqualTo("FOO"); - + assertThat(messagingTemplate.convertSendAndReceive("junk", String.class)).isEqualTo("error:non-convertible"); assertThat(this.adviceCalled.get()).isTrue(); } @Test - public void testUdp() throws Exception { + void testUdp() throws Exception { assertThat(this.config.listeningLatch.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(this.config.serverPort).isEqualTo(this.udpInbound.getPort()); Message outMessage = MessageBuilder.withPayload("foo") @@ -160,7 +161,7 @@ public class IpIntegrationTests { } @Test - public void testUdpInheritance() { + void testUdpInheritance() { UdpMulticastOutboundChannelAdapterSpec udpMulticastOutboundChannelAdapterSpec = Udp.outboundMulticastAdapter("headers['udp_dest']"); @@ -174,7 +175,7 @@ public class IpIntegrationTests { } @Test - public void testCloseStream() throws InterruptedException { + void testCloseStream() throws InterruptedException { IntegrationFlow server = IntegrationFlows.from(Tcp.inboundGateway(Tcp.netServer(0) .deserializer(new ByteArrayRawSerializer()))) .transform(p -> "reply:" + new String(p).toUpperCase()) @@ -192,19 +193,19 @@ public class IpIntegrationTests { } this.applicationContext.addApplicationListener(new Listener()); this.flowContext.registration(server) - .id("streamCloseServer") - .register(); + .id("streamCloseServer") + .register(); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); IntegrationFlow client = IntegrationFlows.from(MessageChannels.direct()) - .handle(Tcp.outboundGateway(Tcp.netClient("localhost", port.get()) - .singleUseConnections(true) - .serializer(new ByteArrayRawSerializer())) - .closeStreamAfterSend(true)) - .transform(Transformers.objectToString()) - .get(); + .handle(Tcp.outboundGateway(Tcp.netClient("localhost", port.get()) + .singleUseConnections(true) + .serializer(new ByteArrayRawSerializer())) + .closeStreamAfterSend(true)) + .transform(Transformers.objectToString()) + .get(); IntegrationFlowRegistration clientRegistration = this.flowContext.registration(client) - .id("streamCloseClient") - .register(); + .id("streamCloseClient") + .register(); assertThat(clientRegistration.getMessagingTemplate() .convertSendAndReceive("foo", String.class)).isEqualTo("reply:FOO"); } @@ -227,12 +228,31 @@ public class IpIntegrationTests { @Bean public IntegrationFlow inTcpGateway() { - return IntegrationFlows.from(Tcp.inboundGateway(server1())) + return IntegrationFlows.from( + Tcp.inboundGateway(server1()) + .replyTimeout(1) + .errorOnTimeout(true) + .errorChannel("inTcpGatewayErrorFlow.input")) .transform(Transformers.objectToString()) + .filter((payload) -> !"junk".equals(payload)) .transform(String::toUpperCase) .get(); } + @Bean + public IntegrationFlow inTcpGatewayErrorFlow() { + return (flow) -> flow + .handle((payload, headers) -> { + if (payload instanceof MessageTimeoutException) { + return "error:non-convertible"; + } + else { + ReflectionUtils.rethrowRuntimeException(payload); + return null; + } + }); + } + @Bean public IntegrationFlow inUdpAdapter() { return IntegrationFlows.from(Udp.inboundAdapter(0)) @@ -252,7 +272,7 @@ public class IpIntegrationTests { @Bean public ApplicationListener events() { - return (ApplicationListener) event -> { + return event -> { this.serverPort = event.getPort(); this.listeningLatch.countDown(); }; diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java index 764db23686..1a687c7fe6 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java @@ -142,6 +142,10 @@ public class ChannelPublishingJmsMessageListener this.gatewayDelegate.setReplyTimeout(replyTimeout); } + public void setErrorOnTimeout(boolean errorOnTimeout) { + this.gatewayDelegate.setErrorOnTimeout(errorOnTimeout); + } + @Override public void setShouldTrack(boolean shouldTrack) { this.gatewayDelegate.setShouldTrack(shouldTrack); diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java index 3c1ae35371..1117550e7b 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java @@ -90,6 +90,12 @@ public class JmsInboundGateway extends MessagingGatewaySupport implements Orderl this.endpoint.getListener().setReplyTimeout(replyTimeout); } + @Override + public void setErrorOnTimeout(boolean errorOnTimeout) { + super.setErrorOnTimeout(errorOnTimeout); + this.endpoint.getListener().setErrorOnTimeout(errorOnTimeout); + } + @Override public void setShouldTrack(boolean shouldTrack) { super.setShouldTrack(shouldTrack); @@ -108,6 +114,8 @@ public class JmsInboundGateway extends MessagingGatewaySupport implements Orderl this.endpoint.setShutdownContainerOnStop(shutdownContainerOnStop); } + + @Override public String getComponentType() { return this.endpoint.getComponentType(); diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/dsl/JmsTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/dsl/JmsTests.java index b8919b66c2..c088cc6d83 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/dsl/JmsTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/dsl/JmsTests.java @@ -28,10 +28,7 @@ import java.util.concurrent.atomic.AtomicInteger; import javax.jms.ConnectionFactory; import org.apache.activemq.ActiveMQConnectionFactory; -import org.apache.logging.log4j.Level; -import org.junit.Rule; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.Test; import org.springframework.beans.factory.ListableBeanFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -39,6 +36,7 @@ import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.Configuration; +import org.springframework.integration.MessageTimeoutException; import org.springframework.integration.annotation.InboundChannelAdapter; import org.springframework.integration.annotation.IntegrationComponentScan; import org.springframework.integration.annotation.MessagingGateway; @@ -56,11 +54,12 @@ import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.dsl.MessageChannels; import org.springframework.integration.dsl.Pollers; import org.springframework.integration.endpoint.MethodInvokingMessageSource; +import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.jms.ActiveMQMultiContextTests; import org.springframework.integration.jms.JmsDestinationPollingSource; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.integration.support.MessageBuilder; -import org.springframework.integration.test.rule.Log4j2LevelAdjuster; +import org.springframework.integration.test.condition.LogLevels; import org.springframework.integration.test.util.TestUtils; import org.springframework.jms.connection.CachingConnectionFactory; import org.springframework.jms.core.JmsTemplate; @@ -76,7 +75,7 @@ import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.messaging.support.InterceptableChannel; import org.springframework.stereotype.Component; import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; import org.springframework.transaction.PlatformTransactionManager; /** @@ -86,7 +85,9 @@ import org.springframework.transaction.PlatformTransactionManager; * * @since 5.0 */ -@RunWith(SpringRunner.class) +@SpringJUnitConfig +@LogLevels(level = "debug", + categories = { "org.springframework", "org.springframework.integration", "org.apache" }) @DirtiesContext public class JmsTests extends ActiveMQMultiContextTests { @@ -145,10 +146,6 @@ public class JmsTests extends ActiveMQMultiContextTests { @Autowired private CountDownLatch redeliveryLatch; - @Rule - public final Log4j2LevelAdjuster adjuster = Log4j2LevelAdjuster.forLevel(Level.DEBUG) - .categories("org.springframework", "org.springframework.integration", "org.apache"); - @Test public void testPollingFlow() { this.controlBus.send("@'jmsTests.ContextConfiguration.integerMessageSource.inboundChannelAdapter'.start()"); @@ -230,6 +227,20 @@ public class JmsTests extends ActiveMQMultiContextTests { .isEqualTo("HELLO THROUGH THE JMS PIPELINE"); assertThat(this.jmsInboundGatewayChannelCalled.get()).isTrue(); + + message = MessageBuilder.withPayload("junk") + .setReplyChannel(replyChannel) + .setHeader("destination", "jmsPipelineTest") + .build(); + + this.jmsOutboundGatewayChannel.send(message); + + receive = replyChannel.receive(5000); + + assertThat(receive) + .isNotNull() + .extracting(Message::getPayload) + .isEqualTo("error: junk is not convertible"); } @Test @@ -406,9 +417,22 @@ public class JmsTests extends ActiveMQMultiContextTests { return IntegrationFlows.from( Jms.inboundGateway(jmsConnectionFactory()) .requestChannel(jmsInboundGatewayInputChannel()) + .replyTimeout(1) + .errorOnTimeout(true) + .errorChannel(new FixedSubscriberChannel(new AbstractReplyProducingMessageHandler() { + + @Override + protected Object handleRequestMessage(Message requestMessage) { + return "error: " + + ((MessageTimeoutException) requestMessage.getPayload()) + .getFailedMessage().getPayload() + " is not convertible"; + } + + })) .destination("jmsPipelineTest") .configureListenerContainer(c -> c.transactionManager(mock(PlatformTransactionManager.class)))) + .filter(payload -> !"junk".equals(payload)) .transform(String::toUpperCase) .get(); } diff --git a/src/reference/asciidoc/endpoint-summary.adoc b/src/reference/asciidoc/endpoint-summary.adoc index 25ebc36825..021bc6ef65 100644 --- a/src/reference/asciidoc/endpoint-summary.adoc +++ b/src/reference/asciidoc/endpoint-summary.adoc @@ -223,3 +223,7 @@ The `` element lets you send data to a `void` meth As discussed in <<./gateway.adoc#gateway,Messaging Gateways>>, the `` element lets any Java program invoke a messaging flow. Each of these works without requiring any source-level dependencies on Spring Integration. The equivalent of an outbound gateway in this context is using a service activator (see <<./service-activator.adoc#service-activator,Service Activator>>) to invoke a method that returns an `Object` of some kind. + +Starting with version `5.2.2`, all the inbound gateways can be configured with an `errorOnTimeout` boolean flag to throw a `MessageTimeoutException` when the downstream flow doesn't return a reply during the reply timeout. +The timer is not started until the thread returns control to the gateway, so usually it is only useful when the downstream flow is asynchronous or it stops because of a `null` return from some handler, e.g. <<./filter.adoc#filter,filter>>. +Such an exception can be handled on the `errorChannel` flow, e.g. producing a compensation reply for requesting client. diff --git a/src/reference/asciidoc/ip.adoc b/src/reference/asciidoc/ip.adoc index faa93bef08..b7e4b02ba9 100644 --- a/src/reference/asciidoc/ip.adoc +++ b/src/reference/asciidoc/ip.adoc @@ -778,7 +778,7 @@ Two additional attributes support this mechanism. `retry-interval` specifies (in milliseconds) how often the framework tries to reconnect after a connection failure. `scheduler` supplies a `TaskScheduler` to schedule the connection attempts and to test that the connection is still active. -If the gateway is started, you may force the gateway to establish a connection by sending a command: `@adapter_id.retryConnection()` and examine the current state with `@adapter_id.isClientModeConnected()`. +If the gateway is started, you may force the gateway to establish a connection by sending a `` command: `@adapter_id.retryConnection()` and examine the current state with `@adapter_id.isClientModeConnected()`. The outbound gateway, after sending a message over the connection, waits for a response, constructs a response message, and puts it on the reply channel. Communications over the connections are single-threaded.