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

GH-3107: Add errorOnTimeout for TcpInboundGateway

Browse Source 
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`
This commit is contained in:
Artem Bilan
2019-11-12 15:08:01 -05:00
committed by Gary Russell
parent 0bbdd3a5f7
commit 72f7c72392
10 changed files with 160 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/Tcp.java
View File

@@ -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);
}

55
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java
View File

@@ -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<String, TcpConnection> connections = new ConcurrentHashMap<>();
private final Map<String, TcpConnection> connections = new ConcurrentHashMap<String, TcpConnection>();
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();
}
}

64
spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java
View File

@@ -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<String> 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())))
.<byte[], String>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())
.<String>filter((payload) -> !"junk".equals(payload))
.<String, String>transform(String::toUpperCase)
.get();
}
@Bean
public IntegrationFlow inTcpGatewayErrorFlow() {
return (flow) -> flow
.<Exception>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<UdpServerListeningEvent> events() {
return (ApplicationListener<UdpServerListeningEvent>) event -> {
return event -> {
this.serverPort = event.getPort();
this.listeningLatch.countDown();
};