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 afaf66a55a..c129d556d5 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 @@ -103,6 +103,8 @@ public abstract class IpAdapterParserUtils { static final String TCP_CONNECTION_FACTORY = "connection-factory"; + public static final String INTERCEPTOR_FACTORY_CHAIN = "interceptor-factory-chain"; + /** * Adds a constructor-arg to the provided bean definition builder * with the value of the attribute whose name is provided if that diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionParser.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionParser.java index b4f7a7e42b..4a5fe20db9 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionParser.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionParser.java @@ -92,6 +92,8 @@ public class TcpConnectionParser extends AbstractBeanDefinitionParser { IpAdapterParserUtils.OUTPUT_CONVERTER); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IpAdapterParserUtils.SINGLE_USE); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, + IpAdapterParserUtils.INTERCEPTOR_FACTORY_CHAIN); return builder.getBeanDefinition(); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java index e3df7aeb75..85a1e5ca92 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java @@ -103,6 +103,7 @@ public class TcpSendingMessageHandler implements MessageHandler, TcpSender { } catch (MessageMappingException e) { // retry - socket may have closed if (e.getCause() instanceof IOException) { + logger.debug("Fail on first write attempt", e); doWrite(message); } else { throw e; @@ -120,13 +121,20 @@ public class TcpSendingMessageHandler implements MessageHandler, TcpSender { if (connection == null) { throw new MessageMappingException(message, "Failed to create connection"); } + if (logger.isDebugEnabled()) { + logger.debug("Got Connection " + connection.getConnectionId()); + } connection.send(message); } catch (Exception e) { + String connectionId = null; + if (this.connection != null) { + connectionId = this.connection.getConnectionId(); + } this.connection = null; if (e instanceof MessageMappingException) { throw (MessageMappingException) e; } - throw new MessageMappingException(message, "Failed to map message", e); + throw new MessageMappingException(message, "Failed to map message using " + connectionId, e); } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java index 5e715fa931..b5d4766f51 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java @@ -31,6 +31,8 @@ import org.springframework.util.Assert; */ public abstract class AbstractClientConnectionFactory extends AbstractConnectionFactory { + protected TcpConnection theConnection; + /** * Constructs a factory that will established connections to the host and port. * @param host The host. diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java index 7f68c7534d..4d3ba93764 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java @@ -79,6 +79,8 @@ public abstract class AbstractConnectionFactory protected boolean active; + protected TcpConnectionInterceptorFactoryChain interceptorFactoryChain; + /** * Sets socket attributes on the socket. * @param socket The socket. @@ -270,6 +272,13 @@ public abstract class AbstractConnectionFactory this.mapper = mapper; } + /** + * @return the singleUse + */ + public boolean isSingleUse() { + return singleUse; + } + /** * If true, sockets created by this factory will be used once. * @param singleUse @@ -278,10 +287,15 @@ public abstract class AbstractConnectionFactory this.singleUse = singleUse; } + public void setPoolSize(int poolSize) { this.poolSize = poolSize; } + public void setInterceptorFactoryChain(TcpConnectionInterceptorFactoryChain interceptorFactoryChain) { + this.interceptorFactoryChain = interceptorFactoryChain; + } + /** * Closes the server. */ @@ -307,6 +321,23 @@ public abstract class AbstractConnectionFactory this.close(); } + protected TcpConnection wrapConnection(TcpConnection connection) throws Exception { + if (this.interceptorFactoryChain == null) { + return connection; + } + for (TcpConnectionInterceptorFactory factory : + this.interceptorFactoryChain.getInterceptorFactories()) { + TcpConnectionInterceptor wrapper = factory.getInterceptor(); + wrapper.setTheConnection(connection); + // if no ultimate listener, register each wrapper in turn + if (this.listener == null) { + connection.registerListener(wrapper); + } + connection = wrapper; + } + return connection; + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java index 3cff726112..af8c6be7c4 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java @@ -65,7 +65,9 @@ public abstract class AbstractServerConnectionFactory extends AbstractConnection * @param socket The new socket. */ protected void initializeConnection(TcpConnection connection, Socket socket) { - connection.registerListener(this.listener); + if (this.listener != null) { + connection.registerListener(this.listener); + } connection.registerSender(this.sender); connection.setMapper(this.mapper); connection.setInputConverter(this.inputConverter); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java index 4344b98dec..56fe8c3e0f 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java @@ -36,10 +36,10 @@ public abstract class AbstractTcpConnection implements TcpConnection { protected Log logger = LogFactory.getLog(this.getClass()); - @SuppressWarnings("unchecked") + @SuppressWarnings("rawtypes") protected InputStreamingConverter inputConverter; - @SuppressWarnings("unchecked") + @SuppressWarnings("rawtypes") protected OutputStreamingConverter outputConverter; protected TcpMessageMapper mapper; @@ -49,7 +49,15 @@ public abstract class AbstractTcpConnection implements TcpConnection { protected TcpSender sender; protected boolean singleUse; + + protected final boolean server; + + protected String connectionId; + public AbstractTcpConnection(boolean server) { + this.server = server; + } + /** * Closes this connection. */ @@ -78,6 +86,14 @@ public abstract class AbstractTcpConnection implements TcpConnection { } } + /** + * + * @return the input converter + */ + public InputStreamingConverter getInputConverter() { + return inputConverter; + } + /** * @param inputConverter the input converter to set */ @@ -85,6 +101,14 @@ public abstract class AbstractTcpConnection implements TcpConnection { this.inputConverter = inputConverter; } + /** + * + * @return the output converter + */ + public OutputStreamingConverter getOutputConverter() { + return outputConverter; + } + /** * @param outputConverter the output converter to set */ @@ -135,4 +159,8 @@ public abstract class AbstractTcpConnection implements TcpConnection { return this.singleUse; } + public boolean isServer() { + return server; + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java new file mode 100644 index 0000000000..116e81af19 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java @@ -0,0 +1,159 @@ +/* + * Copyright 2002-2010 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.ip.tcp.connection; + + +import org.springframework.commons.serializer.InputStreamingConverter; +import org.springframework.commons.serializer.OutputStreamingConverter; +import org.springframework.integration.Message; + +/** + * Base class for TcpConnectionIntercepters; passes all method calls through + * to the underlying {@link TcpConnection}. + * + * @author Gary Russell + * @since 2.0 + * + */ +public abstract class AbstractTcpConnectionInterceptor implements TcpConnectionInterceptor { + + private TcpConnection theConnection; + + private TcpListener tcpListener; + + private TcpSender tcpSender; + + public void close() { + this.theConnection.close(); + } + + public boolean isOpen() { + return this.theConnection.isOpen(); + } + + public Object getPayload() throws Exception { + return this.theConnection.getPayload(); + } + + public String getHostName() { + return this.theConnection.getHostName(); + } + + public String getHostAddress() { + return this.theConnection.getHostAddress(); + } + + public int getPort() { + return this.theConnection.getPort(); + } + + public void registerListener(TcpListener listener) { + this.theConnection.registerListener(this); + this.tcpListener = listener; + } + + public void registerSender(TcpSender sender) { + this.tcpSender = sender; + this.theConnection.registerSender(this); + } + + public String getConnectionId() { + return this.theConnection.getConnectionId(); + } + + public boolean isSingleUse() { + return this.theConnection.isSingleUse(); + } + + public void run() { + this.theConnection.run(); + } + + public void setSingleUse(boolean singleUse) { + this.theConnection.setSingleUse(singleUse); + } + + public void setMapper(TcpMessageMapper mapper) { + this.theConnection.setMapper(mapper); + } + + public InputStreamingConverter getInputConverter() { + return this.theConnection.getInputConverter(); + } + + public void setInputConverter(InputStreamingConverter inputConverter) { + this.theConnection.setInputConverter(inputConverter); + } + + public OutputStreamingConverter getOutputConverter() { + return this.theConnection.getOutputConverter(); + } + + public void setOutputConverter(OutputStreamingConverter outputConverter) { + this.theConnection.setOutputConverter(outputConverter); + } + + public boolean isServer() { + return this.theConnection.isServer(); + } + + public void onMessage(Message message) { + if (this.tcpListener == null) { + throw new NoListenerException("No listener registered for message reception"); + } + this.tcpListener.onMessage(message); + } + + public void send(Message message) throws Exception { + this.theConnection.send(message); + } + + /** + * Returns the underlying connection (or next interceptor) + * @return the connection + */ + public TcpConnection getTheConnection() { + return this.theConnection; + } + + /** + * Sets the underlying connection (or next interceptor) + * @param theConnection the connection + */ + public void setTheConnection(TcpConnection theConnection) { + this.theConnection = theConnection; + } + + /** + * @return the listener + */ + public TcpListener getListener() { + return tcpListener; + } + + public void addNewConnection(TcpConnection connection) { + if (this.tcpSender != null) { + this.tcpSender.addNewConnection(this); + } + } + + public void removeDeadConnection(TcpConnection connection) { + if (this.tcpSender != null) { + this.tcpSender.removeDeadConnection(this); + } + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/NoListenerException.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/NoListenerException.java new file mode 100644 index 0000000000..e31cce6820 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/NoListenerException.java @@ -0,0 +1,37 @@ +/* + * Copyright 2002-2010 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.ip.tcp.connection; + +import org.springframework.core.NestedRuntimeException; + +/** + * @author Gary Russell + * @since 2.0 + * + */ +public class NoListenerException extends NestedRuntimeException { + + private static final long serialVersionUID = -5644042657316429223L; + + public NoListenerException(String msg, Throwable cause) { + super(msg, cause); + } + + public NoListenerException(String msg) { + super(msg); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java index 994ce31542..2c2ae3e7f6 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java @@ -106,19 +106,42 @@ public interface TcpConnection extends Runnable { */ public boolean isSingleUse(); + /** + * + * @return True if connection is used once. + */ + public boolean isServer(); + /** * @param mapper the mapper */ public void setMapper(TcpMessageMapper mapper); + /** + * + * @return the input converter + */ + public InputStreamingConverter getInputConverter(); + /** * @param inputConverter the inputConverter to set */ public void setInputConverter(InputStreamingConverter inputConverter); + /** + * + * @return the output converter + */ + public OutputStreamingConverter getOutputConverter(); + /** * @param outputConverter the outputConverter to set */ public void setOutputConverter(OutputStreamingConverter outputConverter); + /** + * @return this connection's listener + */ + public TcpListener getListener(); + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptor.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptor.java new file mode 100644 index 0000000000..3a7e9caeab --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptor.java @@ -0,0 +1,27 @@ +/* + * Copyright 2002-2010 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.ip.tcp.connection; + +/** + * @author Gary Russell + * @since 2.0 + * + */ +public interface TcpConnectionInterceptor extends TcpConnection, TcpListener, TcpSender { + + public void setTheConnection(TcpConnection connection); + +} \ No newline at end of file diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorFactory.java new file mode 100644 index 0000000000..c580d73f83 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorFactory.java @@ -0,0 +1,31 @@ +/* + * Copyright 2002-2010 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.ip.tcp.connection; + + +/** + * Base class for TcpConnectionInterceptorFactories. Subclasses create prototype beans by + * default. + * + * @author Gary Russell + * @since 2.0 + * + */ +public abstract class TcpConnectionInterceptorFactory { + + public abstract TcpConnectionInterceptor getInterceptor(); +} + diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorFactoryChain.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorFactoryChain.java new file mode 100644 index 0000000000..9615c7e7c9 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorFactoryChain.java @@ -0,0 +1,35 @@ +/* + * Copyright 2002-2010 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.ip.tcp.connection; + +/** + * @author Gary Russell + * @since 2.0 + * + */ +public class TcpConnectionInterceptorFactoryChain { + + private TcpConnectionInterceptorFactory[] interceptorFactories; + + public TcpConnectionInterceptorFactory[] getInterceptorFactories() { + return interceptorFactories; + } + + public void setInterceptors(TcpConnectionInterceptorFactory[] interceptorFactories) { + this.interceptorFactories = interceptorFactories; + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpListener.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpListener.java index dce17577c9..c27cfbf040 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpListener.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpListener.java @@ -34,11 +34,4 @@ public interface TcpListener { */ public abstract void onMessage(Message message); - /** - * Return true if the connection factory is a server - * and it is listening. - * @return true if listening - */ - public boolean isListening(); - } \ No newline at end of file diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java index b304e7fd31..b79d8e1a76 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java @@ -29,8 +29,6 @@ import javax.net.SocketFactory; public class TcpNetClientConnectionFactory extends AbstractClientConnectionFactory { - protected TcpNetConnection theConnection; - /** * Creates a TcpNetClientConnectionFactory for connections to the host and port. * @param host the host @@ -45,14 +43,15 @@ public class TcpNetClientConnectionFactory extends * true, a new connection is returned; otherwise a single connection is * reused for all requests while the connection remains open. */ - public TcpNetConnection getConnection() throws Exception { + public TcpConnection getConnection() throws Exception { if (this.theConnection != null && this.theConnection.isOpen()) { return this.theConnection; } logger.debug("Opening new socket connection to " + this.host + ":" + this.port); Socket socket = SocketFactory.getDefault().createSocket(this.host, this.port); setSocketAttributes(socket); - TcpNetConnection connection = new TcpNetConnection(socket, false); + TcpConnection connection = new TcpNetConnection(socket, false); + connection = wrapConnection(connection); initializeConnection(connection, socket); this.taskExecutor.execute(connection); if (!this.singleUse) { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java index 685e6e7f52..abbe8e7fe1 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java @@ -35,8 +35,6 @@ public class TcpNetConnection extends AbstractTcpConnection { private final Socket socket; - private final boolean server; - /** * Constructs a TcpNetConnection for the socket. * @param socket the socket @@ -44,8 +42,9 @@ public class TcpNetConnection extends AbstractTcpConnection { * a result of an incoming request. */ public TcpNetConnection(Socket socket, boolean server) { + super(server); this.socket = socket; - this.server = server; + getConnectionId(); } /** @@ -121,15 +120,6 @@ public class TcpNetConnection extends AbstractTcpConnection { } break; } - /* - * For single use sockets, we close after receipt if we are on the client - * side, or the server side has no outbound adapter registered - */ - if (this.singleUse && (!this.server || this.sender == null)) { - logger.debug("Closing single use socket after inbound message"); - this.close(); - okToRun = false; - } if (logger.isDebugEnabled()) logger.debug("Message received " + message); try { @@ -139,13 +129,35 @@ public class TcpNetConnection extends AbstractTcpConnection { } listener.onMessage(message); } catch (Exception e) { - logger.error("Exception sending meeeage: " + message, e); + if (e instanceof NoListenerException) { + if (this.singleUse) { + logger.debug("Closing single use socket after inbound message " + this.connectionId); + this.close(); + okToRun = false; + } else { + logger.warn("Unexpected message - no inbound adapter registered with connection " + message); + } + } else { + logger.error("Exception sending meeeage: " + message, e); + } + } + /* + * For single use sockets, we close after receipt if we are on the client + * side, or the server side has no outbound adapter registered + */ + if (this.singleUse && this.server && this.sender == null) { + logger.debug("Closing single use socket after inbound message " + this.connectionId); + this.close(); + okToRun = false; } } } public String getConnectionId() { - return SocketIoUtils.getSocketId(this.socket); + if (this.connectionId == null) { + this.connectionId = SocketIoUtils.getSocketId(this.socket); + } + return this.connectionId; } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java index 001edfb9ca..df253f828d 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java @@ -69,11 +69,12 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto final Socket socket = serverSocket.accept(); logger.debug("Accepted connection from " + socket.getInetAddress().getHostAddress()); setSocketAttributes(socket); - TcpNetConnection connection = new TcpNetConnection(socket, true); + TcpConnection connection = new TcpNetConnection(socket, true); + connection = wrapConnection(connection); this.initializeConnection(connection, socket); this.taskExecutor.execute(connection); } - } catch (IOException e) { + } catch (Exception e) { this.listening = false; if (this.active) { logger.error("Error on ServerSocket", e); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java index 38c1c11269..b118f7f622 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java @@ -39,8 +39,6 @@ import java.util.concurrent.LinkedBlockingQueue; public class TcpNioClientConnectionFactory extends AbstractClientConnectionFactory { - protected TcpNioConnection theConnection; - protected boolean usingDirectBuffers; private Selector selector; @@ -63,7 +61,7 @@ public class TcpNioClientConnectionFactory extends * true, a new connection is returned; otherwise a single connection is * reused for all requests while the connection remains open. */ - public TcpNioConnection getConnection() throws Exception { + public TcpConnection getConnection() throws Exception { int n = 0; while (this.selector == null) { Thread.sleep(100); @@ -78,13 +76,14 @@ public class TcpNioClientConnectionFactory extends SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress(this.host, this.port)); setSocketAttributes(socketChannel.socket()); TcpNioConnection connection = new TcpNioConnection(socketChannel, false); + connection.setUsingDirectBuffers(this.usingDirectBuffers); if (this.taskExecutor == null) { connection.setTaskExecutor(Executors.newSingleThreadExecutor()); } else { connection.setTaskExecutor(this.taskExecutor); } - initializeConnection(connection, socketChannel.socket()); - connection.setUsingDirectBuffers(this.usingDirectBuffers); + TcpConnection wrappedConnection = wrapConnection(connection); + initializeConnection(wrappedConnection, socketChannel.socket()); socketChannel.configureBlocking(false); if (this.soTimeout > 0) { connection.setLastRead(System.currentTimeMillis()); @@ -93,9 +92,9 @@ public class TcpNioClientConnectionFactory extends newChannels.add(socketChannel); selector.wakeup(); if (!this.singleUse) { - this.theConnection = connection; + this.theConnection = wrappedConnection; } - return connection; + return wrappedConnection; } /** diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java index cc4947bffd..3e7a4e9fce 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java @@ -43,8 +43,6 @@ public class TcpNioConnection extends AbstractTcpConnection { private final SocketChannel socketChannel; - private final boolean server; - private OutputStream channelOutputStream; private PipedOutputStream pipedOutputStream; @@ -62,7 +60,7 @@ public class TcpNioConnection extends AbstractTcpConnection { private boolean active = true; private long lastRead; - + /** * Constructs a TcpNetConnection for the SocketChannel. * @param socketChannel the socketChannel @@ -70,11 +68,15 @@ public class TcpNioConnection extends AbstractTcpConnection { * a result of an incoming request. */ public TcpNioConnection(SocketChannel socketChannel, boolean server) throws Exception { + super(server); this.socketChannel = socketChannel; - this.server = server; this.pipedInputStream = new PipedInputStream(); this.pipedOutputStream = new PipedOutputStream(this.pipedInputStream); this.channelOutputStream = new ChannelOutputStream(); + getConnectionId(); + if (this.connectionId == null) { + throw new Exception("Null id"); + } } public void close() { @@ -166,7 +168,6 @@ public class TcpNioConnection extends AbstractTcpConnection { private synchronized void convertAndSend() throws IOException { if (this.pipedInputStream.available() <= 0) { - System.err.println("NO WORK"); return; } Message message = null; @@ -175,7 +176,9 @@ public class TcpNioConnection extends AbstractTcpConnection { } catch (Exception e) { this.close(); if (e instanceof SocketTimeoutException && this.singleUse) { - logger.debug("Closing single use socket after timeout"); + if (logger.isDebugEnabled()) { + logger.debug("Closing single use socket after timeout " + this.connectionId); + } } else { if (!(e instanceof SoftEndOfStreamException)) { logger.error("Read exception " + @@ -187,20 +190,30 @@ public class TcpNioConnection extends AbstractTcpConnection { } return; } - /* - * For single use sockets, we close after receipt if we are on the client - * side, or the server side has no outbound adapter registered - */ - if (this.singleUse && (!this.server || this.sender == null)) { - logger.debug("Closing single use socket after inbound message"); - this.close(); - } + try { if (message != null) { listener.onMessage(message); } } catch (Exception e) { - logger.error("Exception sending meeeage: " + message, e); + if (e instanceof NoListenerException) { + if (this.singleUse) { + if (logger.isDebugEnabled()) { + logger.debug("Closing single use channel after inbound message " + this.connectionId); + } + this.close(); + } + } else { + logger.error("Exception sending meeeage: " + message, e); + } + } + /* + * For single use sockets, we close after receipt if we are on the client + * side, or the server side has no outbound adapter registered + */ + if (this.singleUse && this.server && this.sender == null) { + logger.debug("Closing single use cbannel after inbound message " + this.connectionId); + this.close(); } } @@ -268,7 +281,10 @@ public class TcpNioConnection extends AbstractTcpConnection { } public String getConnectionId() { - return SocketIoUtils.getSocketId(this.socketChannel.socket()); + if (this.connectionId == null) { + this.connectionId = SocketIoUtils.getSocketId(this.socketChannel.socket()); + } + return this.connectionId; } /** @@ -326,7 +342,7 @@ public class TcpNioConnection extends AbstractTcpConnection { doWrite(buffer); } - private void doWrite(ByteBuffer buffer) throws IOException { + private synchronized void doWrite(ByteBuffer buffer) throws IOException { socketChannel.write(buffer); int remaining = buffer.remaining(); if (remaining == 0) { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java index 3d4c0525d6..2b2ad48fd4 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java @@ -187,8 +187,9 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto private TcpNioConnection createTcpNioConnection(SocketChannel socketChannel) { try { TcpNioConnection connection = new TcpNioConnection(socketChannel, true); - this.initializeConnection(connection, socketChannel.socket()); connection.setUsingDirectBuffers(this.usingDirectBuffers); + TcpConnection wrappedConnection = wrapConnection(connection); + this.initializeConnection(wrappedConnection, socketChannel.socket()); return connection; } catch (Exception e) { logger.error("Failed to establish new incoming connection", e); diff --git a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd index c17fb79fd1..44ea131895 100644 --- a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd +++ b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd @@ -372,6 +372,15 @@ its configuration specifies the number of threads. + + + + + + + + + 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 3c2a57a602..4c8cb3c1ff 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 @@ -297,6 +297,7 @@ task-executor="externalTE" pool-size="321" using-direct-buffers="true" + interceptor-factory-chain="interceptors" /> + + + + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java index d66c203002..2a82a25f59 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java @@ -16,6 +16,7 @@ package org.springframework.integration.ip.config; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNotSame; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; @@ -436,6 +437,7 @@ public class ParserUnitTests { assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor")); assertEquals(321, dfa.getPropertyValue("poolSize")); assertEquals(true, dfa.getPropertyValue("usingDirectBuffers")); + assertNotNull(dfa.getPropertyValue("interceptorFactoryChain")); } @Test @@ -454,7 +456,8 @@ public class ParserUnitTests { assertEquals(true, dfa.getPropertyValue("singleUse")); assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor")); assertEquals(123, dfa.getPropertyValue("poolSize")); - assertEquals(true, dfa.getPropertyValue("usingDirectBuffers")); + assertEquals(true, dfa.getPropertyValue("usingDirectBuffers")); + assertNotNull(dfa.getPropertyValue("interceptorFactoryChain")); } @Test diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/InterceptedSharedConnectionTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/InterceptedSharedConnectionTests-context.xml new file mode 100644 index 0000000000..aeadba9812 --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/InterceptedSharedConnectionTests-context.xml @@ -0,0 +1,72 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/InterceptedSharedConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/InterceptedSharedConnectionTests.java new file mode 100644 index 0000000000..81fcd19444 --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/InterceptedSharedConnectionTests.java @@ -0,0 +1,80 @@ +/* + * Copyright 2002-2010 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.ip.tcp; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.fail; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.support.AbstractApplicationContext; +import org.springframework.integration.Message; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.MessageBuilder; +import org.springframework.integration.core.MessageChannel; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + + +/** + * @author Gary Russell + * @since 2.0 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class InterceptedSharedConnectionTests { + + @Autowired + AbstractApplicationContext ctx; + + @Autowired + @Qualifier(value="inboundServer") + TcpReceivingChannelAdapter listener; + + /** + * Tests a loopback. The client-side outbound adapter sends a message over + * a connection from the client connection factory; the server side + * receives the message, puts in on a channel which is the input channel + * for the outbound adapter that's sharing the connections. The response + * comes back to an inbound adapter that is sharing the client's + * connection and we verify we get the echo back as expected. + * + * @throws Exception + */ + @Test + public void test1() throws Exception { + int n = 0; + Object o = ctx.getBean("inboundServer"); + while (!listener.isListening()) { + Thread.sleep(100); + if (n++ > 100) { + fail("Failed to listen"); + } + } + MessageChannel input = ctx.getBean("input", MessageChannel.class); + input.send(MessageBuilder.withPayload("Test").build()); + QueueChannel replies = ctx.getBean("replies", QueueChannel.class); + Message message = replies.receive(10000); + assertNotNull(message); + assertEquals("Test", message.getPayload()); + } + +} diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SharedConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SharedConnectionTests.java index 99e8bb480c..7cbc624894 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SharedConnectionTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SharedConnectionTests.java @@ -29,7 +29,6 @@ import org.springframework.integration.Message; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.MessageBuilder; import org.springframework.integration.core.MessageChannel; -import org.springframework.integration.ip.tcp.connection.TcpListener; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -48,7 +47,7 @@ public class SharedConnectionTests { @Autowired @Qualifier(value="inboundServer") - TcpListener listener; + TcpReceivingChannelAdapter listener; /** * Tests a loopback. The client-side outbound adapter sends a message over diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java index d1391438cb..a2ae51bd89 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java @@ -22,7 +22,11 @@ import static org.junit.Assert.fail; import java.io.IOException; import java.io.InputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; import java.net.Socket; +import java.net.SocketException; +import java.net.UnknownHostException; import java.util.HashSet; import java.util.LinkedList; import java.util.List; @@ -34,10 +38,14 @@ import java.util.concurrent.Executors; import javax.net.SocketFactory; import org.junit.Test; +import org.springframework.commons.serializer.java.JavaStreamingConverter; import org.springframework.integration.Message; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.ip.AbstractInternetProtocolReceivingChannelAdapter; import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; +import org.springframework.integration.ip.tcp.connection.HelloWorldInterceptorFactory; +import org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactory; +import org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactoryChain; import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory; import org.springframework.integration.ip.tcp.converter.ByteArrayCrLfConverter; @@ -577,6 +585,190 @@ public class TcpReceivingChannelAdapterTests { } } + @Test + public void newTestNetInterceptors() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); + interceptorsGuts(port, scf); + } + + @Test + public void newTestNetSingleNoOutboundInterceptors() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); + singleNoOutboundInterceptorsGuts(port, scf); + } + + @Test + public void newTestNetSingleSharedInterceptors() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); + singleSharedInterceptorsGuts(port, scf); + } + + @Test + public void newTestNioInterceptors() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); + interceptorsGuts(port, scf); + } + + @Test + public void newTestNioSingleNoOutboundInterceptors() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); + singleNoOutboundInterceptorsGuts(port, scf); + } + + @Test + public void newTestNioSingleSharedInterceptors() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); + singleSharedInterceptorsGuts(port, scf); + } + + private void interceptorsGuts(final int port, + AbstractServerConnectionFactory scf) throws Exception { + JavaStreamingConverter converter = new JavaStreamingConverter(); + scf.setInputConverter(converter); + scf.setOutputConverter(converter); + scf.setSingleUse(false); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(scf); + TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain(); + fc.setInterceptors(new TcpConnectionInterceptorFactory[] + {new HelloWorldInterceptorFactory(), + new HelloWorldInterceptorFactory()}); + scf.setInterceptorFactoryChain(fc); + scf.setSoTimeout(10000); + scf.start(); + int n = 0; + while (!scf.isListening()) { + Thread.sleep(100); + if (n++ > 100) { + fail("Failed to start listening"); + } + } + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + Socket socket = SocketFactory.getDefault().createSocket("localhost", port); + socket.setSoTimeout(10000); + new ObjectOutputStream(socket.getOutputStream()).writeObject("Hello"); + assertEquals("world!", new ObjectInputStream(socket.getInputStream()).readObject()); + new ObjectOutputStream(socket.getOutputStream()).writeObject("Hello"); + assertEquals("world!", new ObjectInputStream(socket.getInputStream()).readObject()); + new ObjectOutputStream(socket.getOutputStream()).writeObject("Test1"); + new ObjectOutputStream(socket.getOutputStream()).writeObject("Test2"); + Message message = channel.receive(10000); + assertNotNull(message); + assertEquals("Test1", message.getPayload()); + message = channel.receive(10000); + assertNotNull(message); + assertEquals("Test2", message.getPayload()); + } + + private void singleNoOutboundInterceptorsGuts(final int port, + AbstractServerConnectionFactory scf) throws Exception { + JavaStreamingConverter converter = new JavaStreamingConverter(); + scf.setInputConverter(converter); + scf.setOutputConverter(converter); + scf.setSingleUse(true); + scf.setSoTimeout(10000); + TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain(); + fc.setInterceptors(new TcpConnectionInterceptorFactory[] + {new HelloWorldInterceptorFactory(), + new HelloWorldInterceptorFactory()}); + scf.setInterceptorFactoryChain(fc); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(scf); + scf.start(); + int n = 0; + while (!scf.isListening()) { + Thread.sleep(100); + if (n++ > 100) { + fail("Failed to start listening"); + } + } + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + Socket socket = SocketFactory.getDefault().createSocket("localhost", port); + socket.setSoTimeout(10000); + new ObjectOutputStream(socket.getOutputStream()).writeObject("Hello"); + assertEquals("world!", new ObjectInputStream(socket.getInputStream()).readObject()); + new ObjectOutputStream(socket.getOutputStream()).writeObject("Hello"); + assertEquals("world!", new ObjectInputStream(socket.getInputStream()).readObject()); + new ObjectOutputStream(socket.getOutputStream()).writeObject("Test1"); + socket = SocketFactory.getDefault().createSocket("localhost", port); + new ObjectOutputStream(socket.getOutputStream()).writeObject("Hello"); + assertEquals("world!", new ObjectInputStream(socket.getInputStream()).readObject()); + new ObjectOutputStream(socket.getOutputStream()).writeObject("Hello"); + assertEquals("world!", new ObjectInputStream(socket.getInputStream()).readObject()); + new ObjectOutputStream(socket.getOutputStream()).writeObject("Test2"); + Message message = channel.receive(10000); + assertNotNull(message); + // with single use, results may come back in a different order + Set results = new HashSet(); + results.add(message.getPayload()); + message = channel.receive(10000); + assertNotNull(message); + results.add(message.getPayload()); + assertTrue(results.contains("Test1")); + assertTrue(results.contains("Test2")); + } + + private void singleSharedInterceptorsGuts(final int port, + AbstractServerConnectionFactory scf) throws Exception { + JavaStreamingConverter converter = new JavaStreamingConverter(); + scf.setInputConverter(converter); + scf.setOutputConverter(converter); + scf.setSingleUse(true); + scf.setSoTimeout(60000); + TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain(); + fc.setInterceptors(new TcpConnectionInterceptorFactory[] + {new HelloWorldInterceptorFactory(), + new HelloWorldInterceptorFactory()}); + scf.setInterceptorFactoryChain(fc); + TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); + handler.setConnectionFactory(scf); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(scf); + scf.start(); + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + int n = 0; + while (!scf.isListening()) { + Thread.sleep(100); + if (n++ > 100) { + fail("Failed to listen"); + } + } + Socket socket1 = SocketFactory.getDefault().createSocket("localhost", port); + socket1.setSoTimeout(60000); + new ObjectOutputStream(socket1.getOutputStream()).writeObject("Hello"); + assertEquals("world!", new ObjectInputStream(socket1.getInputStream()).readObject()); + new ObjectOutputStream(socket1.getOutputStream()).writeObject("Hello"); + assertEquals("world!", new ObjectInputStream(socket1.getInputStream()).readObject()); + new ObjectOutputStream(socket1.getOutputStream()).writeObject("Test1"); + + Socket socket2 = SocketFactory.getDefault().createSocket("localhost", port); + socket2.setSoTimeout(60000); + new ObjectOutputStream(socket2.getOutputStream()).writeObject("Hello"); + assertEquals("world!", new ObjectInputStream(socket2.getInputStream()).readObject()); + new ObjectOutputStream(socket2.getOutputStream()).writeObject("Hello"); + assertEquals("world!", new ObjectInputStream(socket2.getInputStream()).readObject()); + new ObjectOutputStream(socket2.getOutputStream()).writeObject("Test2"); + + Message message = channel.receive(10000); + assertNotNull(message); + handler.handleMessage(message); + message = channel.receive(10000); + assertNotNull(message); + handler.handleMessage(message); + + assertEquals("Test1", new ObjectInputStream(socket1.getInputStream()).readObject()); + assertEquals("Test2", new ObjectInputStream(socket2.getInputStream()).readObject()); + } + } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java index 509b3aa899..f1abf86bee 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java @@ -18,6 +18,7 @@ package org.springframework.integration.ip.tcp; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; import java.io.IOException; import java.io.InputStream; @@ -41,6 +42,9 @@ import org.springframework.integration.Message; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.MessageBuilder; import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; +import org.springframework.integration.ip.tcp.connection.HelloWorldInterceptorFactory; +import org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactory; +import org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactoryChain; import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory; import org.springframework.integration.ip.tcp.converter.ByteArrayCrLfConverter; @@ -272,7 +276,7 @@ public class TcpSendingMessageHandlerTests { } @Test - public void newTestNio() throws Exception { + public void newTestNioCrLf() throws Exception { final int port = SocketUtils.findAvailableServerSocket(); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean done = new AtomicBoolean(); @@ -827,7 +831,7 @@ public class TcpSendingMessageHandlerTests { while (true) { Socket socket = server.accept(); semaphore.release(); - byte[] b = new byte[8]; + byte[] b = new byte[9]; readFully(socket.getInputStream(), b); b = ("Reply" + (i++) + "\r\n").getBytes(); socket.getOutputStream().write(b); @@ -853,19 +857,267 @@ public class TcpSendingMessageHandlerTests { QueueChannel channel = new QueueChannel(); adapter.setOutputChannel(channel); assertTrue(latch.await(10, TimeUnit.SECONDS)); - for (int i = 100; i < 200; i++) { - handler.handleMessage(MessageBuilder.withPayload("Test" + i).build()); + int i = 0; + try { + for (i = 100; i < 200; i++) { + handler.handleMessage(MessageBuilder.withPayload("Test" + i).build()); + } + } catch (Exception e) { + e.printStackTrace(); + fail("Exception at " + i); } assertTrue(semaphore.tryAcquire(100, 20000, TimeUnit.MILLISECONDS)); Set replies = new HashSet(); - for (int i = 100; i < 200; i++) { + for (i = 100; i < 200; i++) { Message mOut = channel.receive(20000); assertNotNull(mOut); replies.add(new String((byte[])mOut.getPayload())); } - for (int i = 0; i < 100; i++) { + for (i = 0; i < 100; i++) { assertTrue("Reply" + i + " missing", replies.remove("Reply" + i)); } done.set(true); } + + @Test + public void newTestNetNegotiate() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + final CountDownLatch latch = new CountDownLatch(1); + final AtomicBoolean done = new AtomicBoolean(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + Socket socket = server.accept(); + int i = 0; + while (true) { + ObjectInputStream ois = new ObjectInputStream(socket.getInputStream()); + Object in; + ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream()); + if (i == 0) { + in = ois.readObject(); +// System.out.println(in); + oos.writeObject("world!"); + ois = new ObjectInputStream(socket.getInputStream()); + oos = new ObjectOutputStream(socket.getOutputStream()); + in = ois.readObject(); +// System.out.println(in); + oos.writeObject("world!"); + ois = new ObjectInputStream(socket.getInputStream()); + oos = new ObjectOutputStream(socket.getOutputStream()); + } + in = ois.readObject(); + oos.writeObject("Reply" + (++i)); + } + } catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + JavaStreamingConverter converter = new JavaStreamingConverter(); + ccf.setInputConverter(converter); + ccf.setOutputConverter(converter); + ccf.setSoTimeout(10000); + TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain(); + fc.setInterceptors(new TcpConnectionInterceptorFactory[] + {new HelloWorldInterceptorFactory(), + new HelloWorldInterceptorFactory()}); + ccf.setInterceptorFactoryChain(fc); + ccf.start(); + TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); + handler.setConnectionFactory(ccf); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(ccf); + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + Message mOut = channel.receive(10000); + assertNotNull(mOut); + assertEquals("Reply1", mOut.getPayload()); + mOut = channel.receive(10000); + assertNotNull(mOut); + assertEquals("Reply2", mOut.getPayload()); + done.set(true); + } + + @Test + public void newTestNioNegotiate() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + final CountDownLatch latch = new CountDownLatch(1); + final AtomicBoolean done = new AtomicBoolean(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + Socket socket = server.accept(); + int i = 0; + while (true) { + ObjectInputStream ois = new ObjectInputStream(socket.getInputStream()); + Object in; + ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream()); + if (i == 0) { + in = ois.readObject(); +// System.out.println(in); + oos.writeObject("world!"); + ois = new ObjectInputStream(socket.getInputStream()); + oos = new ObjectOutputStream(socket.getOutputStream()); + } + in = ois.readObject(); + oos.writeObject("Reply" + (++i)); + } + } catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); + JavaStreamingConverter converter = new JavaStreamingConverter(); + ccf.setInputConverter(converter); + ccf.setOutputConverter(converter); + ccf.setSoTimeout(10000); + TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain(); + fc.setInterceptors(new TcpConnectionInterceptorFactory[] {new HelloWorldInterceptorFactory()}); + ccf.setInterceptorFactoryChain(fc); + ccf.start(); + TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); + handler.setConnectionFactory(ccf); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(ccf); + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + Message mOut = channel.receive(10000); + assertNotNull(mOut); + assertEquals("Reply1", mOut.getPayload()); + mOut = channel.receive(10000); + assertNotNull(mOut); + assertEquals("Reply2", mOut.getPayload()); + done.set(true); + } + + @Test + public void newTestNetNegotiateSingleNoListen() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + final CountDownLatch latch = new CountDownLatch(1); + final AtomicBoolean done = new AtomicBoolean(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + Socket socket = server.accept(); + int i = 0; + while (true) { + ObjectInputStream ois = new ObjectInputStream(socket.getInputStream()); + Object in; + ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream()); + if (i == 0) { + in = ois.readObject(); +// System.out.println(in); + oos.writeObject("world!"); + ois = new ObjectInputStream(socket.getInputStream()); + oos = new ObjectOutputStream(socket.getOutputStream()); + in = ois.readObject(); +// System.out.println(in); + oos.writeObject("world!"); + ois = new ObjectInputStream(socket.getInputStream()); + oos = new ObjectOutputStream(socket.getOutputStream()); + } + in = ois.readObject(); + oos.writeObject("Reply" + (++i)); + } + } catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + JavaStreamingConverter converter = new JavaStreamingConverter(); + ccf.setInputConverter(converter); + ccf.setOutputConverter(converter); + ccf.setSoTimeout(10000); + TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain(); + fc.setInterceptors(new TcpConnectionInterceptorFactory[] + {new HelloWorldInterceptorFactory(), + new HelloWorldInterceptorFactory()}); + ccf.setInterceptorFactoryChain(fc); + ccf.setSingleUse(true); + ccf.start(); + TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); + handler.setConnectionFactory(ccf); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + done.set(true); + } + + @Test + public void newTestNioNegotiateSingleNoListen() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + final CountDownLatch latch = new CountDownLatch(1); + final AtomicBoolean done = new AtomicBoolean(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + Socket socket = server.accept(); + int i = 0; + while (true) { + ObjectInputStream ois = new ObjectInputStream(socket.getInputStream()); + Object in; + ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream()); + if (i == 0) { + in = ois.readObject(); +// System.out.println(in); + oos.writeObject("world!"); + ois = new ObjectInputStream(socket.getInputStream()); + oos = new ObjectOutputStream(socket.getOutputStream()); + in = ois.readObject(); +// System.out.println(in); + oos.writeObject("world!"); + ois = new ObjectInputStream(socket.getInputStream()); + oos = new ObjectOutputStream(socket.getOutputStream()); + } + in = ois.readObject(); + oos.writeObject("Reply" + (++i)); + } + } catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); + JavaStreamingConverter converter = new JavaStreamingConverter(); + ccf.setInputConverter(converter); + ccf.setOutputConverter(converter); + ccf.setSoTimeout(10000); + TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain(); + fc.setInterceptors(new TcpConnectionInterceptorFactory[] + {new HelloWorldInterceptorFactory(), + new HelloWorldInterceptorFactory()}); + ccf.setInterceptorFactoryChain(fc); + ccf.setSingleUse(true); + ccf.start(); + TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); + handler.setConnectionFactory(ccf); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + done.set(true); + } + } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptor.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptor.java new file mode 100644 index 0000000000..f40ec9a571 --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptor.java @@ -0,0 +1,110 @@ +/* + * Copyright 2002-2010 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.ip.tcp.connection; + +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.integration.Message; +import org.springframework.integration.MessagingException; +import org.springframework.integration.core.MessageBuilder; +import org.springframework.integration.ip.tcp.connection.AbstractTcpConnectionInterceptor; + +/** + * @author Gary Russell + * @since 2.0 + * + */ +public class HelloWorldInterceptor extends AbstractTcpConnectionInterceptor { + + Log logger = LogFactory.getLog(this.getClass()); + + private boolean negotiated; + + private Semaphore negotiationSemaphore = new Semaphore(0); + + private long timeout = 10000; + + private String hello = "Hello"; + + private String world = "world!"; + + public HelloWorldInterceptor() { + } + + /** + * @param hello + * @param world + */ + public HelloWorldInterceptor(String hello, String world) { + super(); + this.hello = hello; + this.world = world; + } + + @Override + public void onMessage(Message message) { + if (!this.negotiated) { + Object payload = message.getPayload(); + if (this.isServer()) { + if (payload.equals(hello)) { + try { + logger.debug("sending " + this.world); + super.send(MessageBuilder.withPayload(world).build()); + this.negotiated = true; + return; + } catch (Exception e) { + throw new MessagingException("Negotiation error", e); + } + } else { + throw new MessagingException("Negotiation error, expected '" + hello + + "' received '" + payload + "'"); + } + } else { + logger.debug("received " + payload); + if (payload.equals(world)) { + this.negotiated = true; + this.negotiationSemaphore.release(); + } else { + throw new MessagingException("Negotiation error - expected '" + world + + "' received " + payload); + } + return; + } + } + super.onMessage(message); + } + + @Override + public void send(Message message) throws Exception { + if (!this.negotiated) { + if (!this.isServer()) { + logger.debug("Sending " + hello); + super.send(MessageBuilder.withPayload(hello).build()); + this.negotiationSemaphore.tryAcquire(this.timeout, TimeUnit.MILLISECONDS); + if (!this.negotiated) { + throw new MessagingException("Negotiation error"); + } + } + } + super.send(message); + } + + + +} diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptorFactory.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptorFactory.java new file mode 100644 index 0000000000..ebdbd38c40 --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptorFactory.java @@ -0,0 +1,51 @@ +/* + * Copyright 2002-2010 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.ip.tcp.connection; + +import org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactory; + +/** + * @author Gary Russell + * @since 2.0 + * + */ +public class HelloWorldInterceptorFactory extends + TcpConnectionInterceptorFactory { + + private String hello = "Hello"; + + private String world = "world!"; + + public HelloWorldInterceptorFactory() { + } + + /** + * @param hello + * @param world + */ + public HelloWorldInterceptorFactory(String hello, String world) { + this.hello = hello; + this.world = world; + } + + + @Override + public TcpConnectionInterceptor getInterceptor() { + return new HelloWorldInterceptor(hello, world); + } + + +} diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java index c4dbf335a0..3e6ff36d3c 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java @@ -64,7 +64,7 @@ public class TcpNioConnectionTests { }); assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); try { - TcpNioConnection connection = factory.getConnection(); + TcpConnection connection = factory.getConnection(); connection.send(MessageBuilder.withPayload(new byte[1000000]).build()); } catch (Exception e) { assertTrue("Expected SocketTimeoutException, got " + e.getClass().getSimpleName() + @@ -97,7 +97,7 @@ public class TcpNioConnectionTests { }); assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); try { - TcpNioConnection connection = factory.getConnection(); + TcpConnection connection = factory.getConnection(); connection.send(MessageBuilder.withPayload("Test").build()); int n = 0; while (connection.isOpen()) { @@ -108,7 +108,7 @@ public class TcpNioConnectionTests { } assertTrue(!connection.isOpen()); } catch (Exception e) { - fail("Unexptected exception " + e); + fail("Unexpected exception " + e); } }