diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java index 9e6b784ce6..57a615dcc4 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -120,6 +120,8 @@ public abstract class IpAdapterParserUtils { public static final String NIO_CONNECTION_SUPPORT = "nio-connection-support"; + public static final String NET_CONNECTION_SUPPORT = "net-connection-support"; + public static final String SOCKET_FACTORY_SUPPORT = "socket-factory-support"; public static final String BACKLOG = "backlog"; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java index 235f837cba..8e9c699cb7 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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,6 +30,7 @@ import org.springframework.core.serializer.Serializer; import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; +import org.springframework.integration.ip.tcp.connection.DefaultTcpNetConnectionSupport; import org.springframework.integration.ip.tcp.connection.DefaultTcpNetSSLSocketFactorySupport; import org.springframework.integration.ip.tcp.connection.DefaultTcpNetSocketFactorySupport; import org.springframework.integration.ip.tcp.connection.DefaultTcpNioConnectionSupport; @@ -38,6 +39,7 @@ import org.springframework.integration.ip.tcp.connection.DefaultTcpSocketSupport import org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactoryChain; import org.springframework.integration.ip.tcp.connection.TcpMessageMapper; import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpNetConnectionSupport; import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpNioConnectionSupport; @@ -118,6 +120,8 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean A reference to a TcpNioConnectionSupport strategy implementation. - When 'using-nio' is true, this is used to create connections. + When 'using-nio' is true, this is used to create connection objects. Two default implementations are provided 'DefaultTcpNioConnectionSupport' and 'DefaultTcpNioSSLConnectionSupport' depending on whether SSL is in use of not. @@ -754,6 +754,20 @@ + + + + A reference to a TcpNetConnectionSupport strategy implementation. + When 'using-nio' is false, this is used to create connection objects. + + + + + + + + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml index d83a3b277e..e5478d7838 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml @@ -92,8 +92,12 @@ apply-sequence="true" ssl-context-support="sslContextSupport" socket-support="socketSupport" + net-connection-support="netConnectionSupport" socket-factory-support="socketFactorySupport" /> + + message = new GenericMessage<>("foo", + Collections.singletonMap(MessageHeaders.REPLY_CHANNEL, replies)); + channel1.send(message); + channel2.send(message); + assertThat(replies.getQueueSize(), equalTo(2)); + Message replyA = replies.receive(0); + Message replyB = replies.receive(0); + assertThat((String) replyA.getPayload(), containsString("ip_connectionId:pushback:")); + assertThat(replyB.getPayload(), not(equalTo(replyA.getPayload()))); + CompositeDeserializer deserializer = server.getBean(CompositeDeserializer.class); + assertTrue(deserializer.receivedCrLf); + assertTrue(deserializer.receivedStxEtx); + System.getProperties().remove(PORT); + client.close(); + server.close(); + } + + @Test + public void testPushbackNio() throws Exception { + AnnotationConfigApplicationContext server = new AnnotationConfigApplicationContext(ServerNio.class); + TcpNioServerConnectionFactory serverCF = server.getBean(TcpNioServerConnectionFactory.class); + int port = waitForPort(serverCF); + System.setProperty(PORT, String.valueOf(port)); + AnnotationConfigApplicationContext client = new AnnotationConfigApplicationContext(ClientNio.class); + MessageChannel channel1 = client.getBean("out1", MessageChannel.class); // crlf + MessageChannel channel2 = client.getBean("out2", MessageChannel.class); // stxetx + QueueChannel replies = new QueueChannel(); + Message message = new GenericMessage<>("foo", + Collections.singletonMap(MessageHeaders.REPLY_CHANNEL, replies)); + channel1.send(message); + channel2.send(message); + assertThat(replies.getQueueSize(), equalTo(2)); + Message replyA = replies.receive(0); + Message replyB = replies.receive(0); + assertThat((String) replyA.getPayload(), containsString("ip_connectionId:pushback:")); + assertThat(replyB.getPayload(), not(equalTo(replyA.getPayload()))); + CompositeDeserializer deserializer = server.getBean(CompositeDeserializer.class); + assertTrue(deserializer.receivedCrLf); + assertTrue(deserializer.receivedStxEtx); + System.getProperties().remove(PORT); + client.close(); + server.close(); + } + + @Test + public void testPushbackNioSSL() throws Exception { + AnnotationConfigApplicationContext server = new AnnotationConfigApplicationContext(ServerNioSSL.class, + SSLConfig.class); + TcpNioServerConnectionFactory serverCF = server.getBean(TcpNioServerConnectionFactory.class); + int port = waitForPort(serverCF); + System.setProperty(PORT, String.valueOf(port)); + AnnotationConfigApplicationContext client = new AnnotationConfigApplicationContext(ClientNioSSL.class, + SSLConfig.class); + MessageChannel channel1 = client.getBean("out1", MessageChannel.class); // crlf + MessageChannel channel2 = client.getBean("out2", MessageChannel.class); // stxetx + QueueChannel replies = new QueueChannel(); + Message message = new GenericMessage<>("foo", + Collections.singletonMap(MessageHeaders.REPLY_CHANNEL, replies)); + channel1.send(message); + channel2.send(message); + assertThat(replies.getQueueSize(), equalTo(2)); + Message replyA = replies.receive(0); + Message replyB = replies.receive(0); + assertThat((String) replyA.getPayload(), containsString("ip_connectionId:pushback:")); + assertThat(replyB.getPayload(), not(equalTo(replyA.getPayload()))); + CompositeDeserializer deserializer = server.getBean(CompositeDeserializer.class); + assertTrue(deserializer.receivedCrLf); + assertTrue(deserializer.receivedStxEtx); + System.getProperties().remove(PORT); + client.close(); + server.close(); + } + + private int waitForPort(AbstractServerConnectionFactory serverCF) throws InterruptedException { + int port = serverCF.getPort(); + int n = 0; + while (n++ < 200 && port == 0) { + Thread.sleep(100); + port = serverCF.getPort(); + } + assertTrue(n < 200); + return port; + } + + @Configuration + @EnableIntegration + public static class ServerNet { + + @Bean + public TcpNetServerConnectionFactory sf() { + TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(0); + server.setDeserializer(deserializer()); + DefaultTcpNetConnectionSupport connectionSupport = new DefaultTcpNetConnectionSupport(); + connectionSupport.setPushbackCapable(true); + server.setTcpNetConnectionSupport(connectionSupport); + return server; + } + + @Bean + public CompositeDeserializer deserializer() { + return new CompositeDeserializer(); + } + + @Bean + public TcpInboundGateway inGate() { + TcpInboundGateway inGate = new TcpInboundGateway(); + inGate.setConnectionFactory(sf()); + inGate.setRequestChannelName("in"); + return inGate; + } + + @ServiceActivator(inputChannel = "in") + public String handle(Message message) { + return IpHeaders.CONNECTION_ID + ":" + (String) message.getHeaders().get(IpHeaders.CONNECTION_ID); + } + + } + + @Configuration + @EnableIntegration + public static class ClientNet { + + @Bean + public TcpNetClientConnectionFactory cf1() { + TcpNetClientConnectionFactory cf = new TcpNetClientConnectionFactory("localhost", + Integer.parseInt(System.getProperty(PORT))); + cf.setSingleUse(true); + return cf; + } + + @Bean + @ServiceActivator(inputChannel = "out1") + public TcpOutboundGateway outGate1() { + TcpOutboundGateway outGate = new TcpOutboundGateway(); + outGate.setConnectionFactory(cf1()); + outGate.setReplyChannelName("toString"); + return outGate; + } + + @Bean + public TcpNetClientConnectionFactory cf2() { + TcpNetClientConnectionFactory cf = new TcpNetClientConnectionFactory("localhost", + Integer.parseInt(System.getProperty(PORT))); + cf.setSerializer(new ByteArrayStxEtxSerializer()); + return cf; + } + + @Bean + @ServiceActivator(inputChannel = "out2") + public TcpOutboundGateway outGate2() { + TcpOutboundGateway outGate = new TcpOutboundGateway(); + outGate.setConnectionFactory(cf2()); + outGate.setReplyChannelName("toString"); + return outGate; + } + + @Bean + @Transformer(inputChannel = "toString") + public ObjectToStringTransformer otst() { + return new ObjectToStringTransformer(); + } + + } + + @Configuration + @EnableIntegration + public static class ServerNio { + + @Bean + public TcpNioServerConnectionFactory sf() { + TcpNioServerConnectionFactory server = new TcpNioServerConnectionFactory(0); + server.setDeserializer(deserializer()); + DefaultTcpNioConnectionSupport connectionSupport = new DefaultTcpNioConnectionSupport(); + connectionSupport.setPushbackCapable(true); + server.setTcpNioConnectionSupport(connectionSupport); + return server; + } + + @Bean + public CompositeDeserializer deserializer() { + return new CompositeDeserializer(); + } + + @Bean + public TcpInboundGateway inGate() { + TcpInboundGateway inGate = new TcpInboundGateway(); + inGate.setConnectionFactory(sf()); + inGate.setRequestChannelName("in"); + return inGate; + } + + @ServiceActivator(inputChannel = "in") + public String handle(Message message) { + return IpHeaders.CONNECTION_ID + ":" + (String) message.getHeaders().get(IpHeaders.CONNECTION_ID); + } + + } + + @Configuration + @EnableIntegration + public static class ClientNio { + + @Bean + public TcpNioClientConnectionFactory cf1() { + TcpNioClientConnectionFactory cf = new TcpNioClientConnectionFactory("localhost", + Integer.parseInt(System.getProperty(PORT))); + cf.setSingleUse(true); + return cf; + } + + @Bean + @ServiceActivator(inputChannel = "out1") + public TcpOutboundGateway outGate1() { + TcpOutboundGateway outGate = new TcpOutboundGateway(); + outGate.setConnectionFactory(cf1()); + outGate.setRemoteTimeout(50000); + outGate.setReplyChannelName("toString"); + return outGate; + } + + @Bean + public TcpNioClientConnectionFactory cf2() { + TcpNioClientConnectionFactory cf = new TcpNioClientConnectionFactory("localhost", + Integer.parseInt(System.getProperty(PORT))); + cf.setSerializer(new ByteArrayStxEtxSerializer()); + return cf; + } + + @Bean + @ServiceActivator(inputChannel = "out2") + public TcpOutboundGateway outGate2() { + TcpOutboundGateway outGate = new TcpOutboundGateway(); + outGate.setConnectionFactory(cf2()); + outGate.setReplyChannelName("toString"); + return outGate; + } + + @Bean + @Transformer(inputChannel = "toString") + public ObjectToStringTransformer otst() { + return new ObjectToStringTransformer(); + } + + } + + @Configuration + @EnableIntegration + public static class ServerNioSSL { + + @Bean + public TcpNioServerConnectionFactory sf(DefaultTcpNioSSLConnectionSupport sslNioSupport) { + TcpNioServerConnectionFactory server = new TcpNioServerConnectionFactory(0); + server.setDeserializer(deserializer()); + server.setTcpNioConnectionSupport(sslNioSupport); + return server; + } + + @Bean + public CompositeDeserializer deserializer() { + return new CompositeDeserializer(); + } + + @Bean + public TcpInboundGateway inGate(TcpNioServerConnectionFactory sf) { + TcpInboundGateway inGate = new TcpInboundGateway(); + inGate.setConnectionFactory(sf); + inGate.setRequestChannelName("in"); + return inGate; + } + + @ServiceActivator(inputChannel = "in") + public String handle(Message message) { + return IpHeaders.CONNECTION_ID + ":" + (String) message.getHeaders().get(IpHeaders.CONNECTION_ID); + } + + } + + @Configuration + @EnableIntegration + public static class ClientNioSSL { + + @Bean + public TcpNioClientConnectionFactory cf1(DefaultTcpNioSSLConnectionSupport sslNioSupport) { + TcpNioClientConnectionFactory cf = new TcpNioClientConnectionFactory("localhost", + Integer.parseInt(System.getProperty(PORT))); + cf.setTcpNioConnectionSupport(sslNioSupport); + return cf; + } + + @Bean + @ServiceActivator(inputChannel = "out1") + public TcpOutboundGateway outGate1(DefaultTcpNioSSLConnectionSupport sslNioSupport) { + TcpOutboundGateway outGate = new TcpOutboundGateway(); + outGate.setConnectionFactory(cf1(sslNioSupport)); + outGate.setReplyChannelName("toString"); + return outGate; + } + + @Bean + public TcpNioClientConnectionFactory cf2(DefaultTcpNioSSLConnectionSupport sslNioSupport) { + TcpNioClientConnectionFactory cf = new TcpNioClientConnectionFactory("localhost", + Integer.parseInt(System.getProperty(PORT))); + cf.setSerializer(new ByteArrayStxEtxSerializer()); + cf.setTcpNioConnectionSupport(sslNioSupport); + return cf; + } + + @Bean + @ServiceActivator(inputChannel = "out2") + public TcpOutboundGateway outGate2(DefaultTcpNioSSLConnectionSupport sslNioSupport) { + TcpOutboundGateway outGate = new TcpOutboundGateway(); + outGate.setConnectionFactory(cf2(sslNioSupport)); + outGate.setReplyChannelName("toString"); + return outGate; + } + + @Bean + @Transformer(inputChannel = "toString") + public ObjectToStringTransformer otst() { + return new ObjectToStringTransformer(); + } + + } + + @Configuration + public static class SSLConfig { + + @Bean + public DefaultTcpNioSSLConnectionSupport connectionSupport() { + DefaultTcpSSLContextSupport sslContextSupport = new DefaultTcpSSLContextSupport("test.ks", + "test.truststore.ks", "secret", "secret"); + sslContextSupport.setProtocol("SSL"); + DefaultTcpNioSSLConnectionSupport tcpNioConnectionSupport = + new DefaultTcpNioSSLConnectionSupport(sslContextSupport); + tcpNioConnectionSupport.setPushbackCapable(true); + return tcpNioConnectionSupport; + } + + } + + private static class CompositeDeserializer implements Deserializer { + + private final ByteArrayStxEtxSerializer stxEtx = new ByteArrayStxEtxSerializer(); + + private final ByteArrayCrLfSerializer crlf = new ByteArrayCrLfSerializer(); + + private volatile boolean receivedStxEtx; + + private volatile boolean receivedCrLf; + + CompositeDeserializer() { + super(); + } + + @Override + public byte[] deserialize(InputStream inputStream) throws IOException { + PushbackInputStream pbis = (PushbackInputStream) inputStream; + int first = pbis.read(); + if (first < 0) { + throw new SoftEndOfStreamException(); + } + pbis.unread(first); + if (first == ByteArrayStxEtxSerializer.STX) { + this.receivedStxEtx = true; + return this.stxEtx.deserialize(pbis); + } + else { + this.receivedCrLf = true; + return this.crlf.deserialize(pbis); + } + } + + } + +} diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionTests.java index ce4e912df6..6dd865ac54 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2017 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. @@ -36,8 +36,6 @@ import org.junit.Test; import org.mockito.Mockito; import org.springframework.beans.DirectFieldAccessor; -import org.springframework.context.ApplicationEvent; -import org.springframework.context.ApplicationEventPublisher; import org.springframework.integration.ip.tcp.connection.TcpNioConnection.ChannelInputStream; import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer; import org.springframework.integration.ip.tcp.serializer.MapJsonSerializer; @@ -49,31 +47,20 @@ import org.springframework.messaging.support.ErrorMessage; /** * @author Gary Russell + * @author Artem Bilan + * * @since 2.2.2 * */ public class TcpNetConnectionTests { - private final ApplicationEventPublisher nullPublisher = new ApplicationEventPublisher() { - - @Override - public void publishEvent(ApplicationEvent event) { - } - - @Override - public void publishEvent(Object event) { - - } - - }; - @Test public void testErrorLog() throws Exception { Socket socket = mock(Socket.class); InputStream stream = mock(InputStream.class); when(socket.getInputStream()).thenReturn(stream); when(stream.read()).thenReturn((int) 'x'); - TcpNetConnection connection = new TcpNetConnection(socket, true, false, nullPublisher, null); + TcpNetConnection connection = new TcpNetConnection(socket, true, false, e -> { }, null); connection.setDeserializer(new ByteArrayStxEtxSerializer()); final AtomicReference log = new AtomicReference(); Log logger = mock(Log.class); @@ -88,8 +75,8 @@ public class TcpNetConnectionTests { connection.run(); assertNotNull(log.get()); assertEquals("Read exception " + - connection.getConnectionId() + - " MessageMappingException:Expected STX to begin message", + connection.getConnectionId() + + " MessageMappingException:Expected STX to begin message", log.get()); } @@ -98,7 +85,7 @@ public class TcpNetConnectionTests { SocketChannel socketChannel = mock(SocketChannel.class); Socket socket = mock(Socket.class); when(socketChannel.socket()).thenReturn(socket); - TcpNioConnection connection = new TcpNioConnection(socketChannel, true, false, nullPublisher, null); + TcpNioConnection connection = new TcpNioConnection(socketChannel, true, false, e -> { }, null); ChannelInputStream inputStream = TestUtils.getPropertyValue(connection, "channelInputStream", ChannelInputStream.class); inputStream.write(ByteBuffer.wrap(new byte[] { (byte) 0x80 })); @@ -111,7 +98,7 @@ public class TcpNetConnectionTests { PipedInputStream pipe = new PipedInputStream(); when(inSocket.getInputStream()).thenReturn(pipe); - TcpConnectionSupport inboundConnection = new TcpNetConnection(inSocket, true, false, nullPublisher, null); + TcpConnectionSupport inboundConnection = new TcpNetConnection(inSocket, true, false, e -> { }, null); inboundConnection.setDeserializer(new MapJsonSerializer()); MapMessageConverter inConverter = new MapMessageConverter(); MessageConvertingTcpMessageMapper inMapper = new MessageConvertingTcpMessageMapper(inConverter); @@ -119,7 +106,7 @@ public class TcpNetConnectionTests { ByteArrayOutputStream baos = new ByteArrayOutputStream(); Socket outSocket = mock(Socket.class); - TcpNetConnection outboundConnection = new TcpNetConnection(outSocket, true, false, nullPublisher, null); + TcpNetConnection outboundConnection = new TcpNetConnection(outSocket, true, false, e -> { }, null); when(outSocket.getOutputStream()).thenReturn(baos); MapMessageConverter outConverter = new MapMessageConverter(); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java index 26e8d8c7b0..6cd0d0a893 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java @@ -33,6 +33,7 @@ import java.util.concurrent.atomic.AtomicReference; import javax.net.SocketFactory; +import org.junit.Rule; import org.junit.Test; import org.springframework.integration.ip.tcp.serializer.AbstractByteArraySerializer; @@ -41,6 +42,7 @@ import org.springframework.integration.ip.tcp.serializer.ByteArrayLengthHeaderSe import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer; import org.springframework.integration.ip.util.SocketTestUtils; import org.springframework.integration.ip.util.TestingUtilities; +import org.springframework.integration.test.support.LongRunningIntegrationTest; import org.springframework.messaging.Message; import org.springframework.messaging.support.ErrorMessage; @@ -52,6 +54,9 @@ import org.springframework.messaging.support.ErrorMessage; */ public class TcpNioConnectionReadTests { + @Rule + public LongRunningIntegrationTest longRunningIntegrationTest = new LongRunningIntegrationTest(); + private final CountDownLatch latch = new CountDownLatch(1); private AbstractServerConnectionFactory getConnectionFactory( @@ -111,7 +116,7 @@ public class TcpNioConnectionReadTests { AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> { responses.add(message); try { - Thread.sleep(1000); + Thread.sleep(10); } catch (InterruptedException e) { Thread.currentThread().interrupt(); diff --git a/src/reference/asciidoc/ip.adoc b/src/reference/asciidoc/ip.adoc index c572938515..0509a26184 100644 --- a/src/reference/asciidoc/ip.adoc +++ b/src/reference/asciidoc/ip.adoc @@ -931,7 +931,7 @@ The keystore file names (first two constructor arguments) use the Spring `Resour Starting with _version 4.3.6_, when using NIO, you can specify an `ssl-handshake-timeout` (seconds) on the connection factory. This timeout (default 30) is used during SSL handshake when waiting for data; if the timeout is exceeded, the process is aborted and the socket closed. -[[advanced-techniques]] +[[tcp-advanced-techniques]] === Advanced Techniques ==== Strategy Interfaces @@ -942,8 +942,10 @@ However, a number of strategy interfaces are provided to allow customization and * `TcpSSLContextSupport` * `TcpSocketFactorySupport` * `TcpSocketSupport` +* `TcpNetConnectionSupport` * `TcpNioConnectionSupport` +.TcpSSLContextSupport [source,java] ---- public interface TcpSSLContextSupport { @@ -957,6 +959,7 @@ Implementations of this interface are responsible for creating an SSLContext. The implementation provided by the framework is the `DefaultTcpSSLContextSupport` described above. If you require different behavior, implement this interface and provide the connection factory with a reference to a bean of your class' implementation. +.TcpSocketFactorySupport [source,java] ---- public interface TcpSocketFactorySupport { @@ -975,6 +978,7 @@ The second implementation is `DefaultTcpNetSSLSocketFactorySupport`; this is use NOTE: This interface only applies if `using-nio` is "false"; socket factories are not used by NIO. +.TcpSocketSupport [source,java] ---- public interface TcpSocketSupport { @@ -993,6 +997,56 @@ The sole implementation provided by the framework is the `DefaultTcpSocketSuppor To supply your own implementation of `TcpSocketFactorySupport` or `TcpSocketSupport`, provide the connection factory with references to beans of your custom type using the `socket-factory-support` and `socket-support` attributes, respectively. +.TcpNetConnectionSupport +[source, java] +---- +public interface TcpNetConnectionSupport { + + TcpNetConnection createNewConnection(Socket socket, + boolean server, boolean lookupHost, + ApplicationEventPublisher applicationEventPublisher, + String connectionFactoryName) throws Exception; + +} +---- + +This interface is invoked to create `TcpNetConnection` objects (or objects from subclasses). +The framework provides a single implementation `DefatulTcpNetConnectionSupport` which creates simple `TcpNetConnection` objects by default. +It has two properties `pushbackCapable` and `pushbackBufferSize`; when push back is enabled, the implementation returns a subclass that wraps the connection's `InputStream` in a `PushbackInputStream`. +Aligned with the `PushbackInputStream` default, the buffer size defaults to 1. +This enables deserializers to "unread" (push back) bytes into the stream. +The following is a trivial example of how it might be used in a delegating deserializer which "peeks" at the first byte to determine which deserializer to invoke: + +[source, java] +---- +public class CompositeDeserializer implements Deserializer { + + private final ByteArrayStxEtxSerializer stxEtx = new ByteArrayStxEtxSerializer(); + + private final ByteArrayCrLfSerializer crlf = new ByteArrayCrLfSerializer(); + + @Override + public byte[] deserialize(InputStream inputStream) throws IOException { + PushbackInputStream pbis = (PushbackInputStream) inputStream; + int first = pbis.read(); + if (first < 0) { + throw new SoftEndOfStreamException(); + } + pbis.unread(first); + if (first == ByteArrayStxEtxSerializer.STX) { + this.receivedStxEtx = true; + return this.stxEtx.deserialize(pbis); + } + else { + this.receivedCrLf = true; + return this.crlf.deserialize(pbis); + } + } + +} +---- + +.TcpNioConnectionSupport [source, java] ---- public interface TcpNioConnectionSupport { @@ -1005,9 +1059,10 @@ public interface TcpNioConnectionSupport { } ---- -This interface is invoked to create `TcpNioConnection` objects (or subclasses). +This interface is invoked to create `TcpNioConnection` objects (or objects from subclasses). Two implementations are provided `DefaultTcpNioSSLConnectionSupport` and `DefaultTcpNioConnectionSupport` which are used depending on whether SSL is in use or not. A common use case would be to subclass `DefaultTcpNioSSLConnectionSupport` and override `postProcessSSLEngine`; see the example below. +As with the `DefatulTcpNetConnectionSupport`, these implementations also support push back. ==== Example: Enabling SSL Client Authentication @@ -1221,7 +1276,7 @@ Defaults to true. | Y | Y | -| See <> +| See <> | read-delay | Y | Y diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 3120986d61..1d80affba6 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -215,5 +215,9 @@ See <> for more information. ==== TCP Changes + A new `ThreadAffinityClientConnectionFactory` is provided that binds TCP connections to threads. See <> for more information. + +You can now configure the TCP connection factories to support `PushbackInputStream` s, allowing deserializers to "unread" (push back) bytes after "reading ahead". +See <> for more information.