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 79d6041de4..e55cbc0a8d 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-2012 the original author or authors. + * Copyright 2002-2013 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. @@ -123,6 +123,8 @@ public abstract class IpAdapterParserUtils { public static final String BACKLOG = "backlog"; + public static final String MAPPER = "mapper"; + private IpAdapterParserUtils() {} /** diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpNamespaceHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpNamespaceHandler.java index f2b3600da1..52745dae02 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpNamespaceHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpNamespaceHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2013 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. @@ -20,7 +20,7 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa /** * Namespace handler for Spring Integration's ip namespace. - * + * * @author Gary Russell * @since 2.0 */ @@ -31,9 +31,10 @@ public class IpNamespaceHandler extends AbstractIntegrationNamespaceHandler { this.registerBeanDefinitionParser("udp-outbound-channel-adapter", new UdpOutboundChannelAdapterParser()); this.registerBeanDefinitionParser("tcp-inbound-gateway", new TcpInboundGatewayParser()); this.registerBeanDefinitionParser("tcp-outbound-gateway", new TcpOutboundGatewayParser()); - this.registerBeanDefinitionParser("tcp-connection-factory", new TcpConnectionParser()); + this.registerBeanDefinitionParser("tcp-connection-factory", new TcpConnectionFactoryParser()); this.registerBeanDefinitionParser("tcp-inbound-channel-adapter", new TcpInboundChannelAdapterParser()); this.registerBeanDefinitionParser("tcp-outbound-channel-adapter", new TcpOutboundChannelAdapterParser()); + this.registerBeanDefinitionParser("tcp-connection-event-inbound-channel-adapter", new TcpConnectionEventInboundChannelAdapterParser()); } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionEventInboundChannelAdapterParser.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionEventInboundChannelAdapterParser.java new file mode 100644 index 0000000000..4a92d4252e --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionEventInboundChannelAdapterParser.java @@ -0,0 +1,43 @@ +/* + * Copyright 2002-2013 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.config; + +import org.springframework.beans.factory.support.AbstractBeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.AbstractChannelAdapterParser; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.ip.tcp.connection.TcpConnectionEventListeningMessageProducer; +import org.w3c.dom.Element; + +/** + * @author Gary Russell + * @since 3.0 + */ +public class TcpConnectionEventInboundChannelAdapterParser extends AbstractChannelAdapterParser { + + @Override + protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) { + BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder + .rootBeanDefinition(TcpConnectionEventListeningMessageProducer.class); + adapterBuilder.addPropertyReference("outputChannel", channelName); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(adapterBuilder, element, "error-channel", "errorChannel"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(adapterBuilder, element, "event-types"); + return adapterBuilder.getBeanDefinition(); + } + +} 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 213b469c89..34308fde77 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-2012 the original author or authors. + * Copyright 2002-2013 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. @@ -20,6 +20,8 @@ import java.util.concurrent.Executor; import org.springframework.beans.factory.BeanNameAware; import org.springframework.beans.factory.config.AbstractFactoryBean; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.context.SmartLifecycle; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; @@ -50,7 +52,8 @@ import org.springframework.util.Assert; * @author Gary Russell * @since 2.0.5 */ -public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean implements SmartLifecycle, BeanNameAware { +public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean implements SmartLifecycle, BeanNameAware, + ApplicationEventPublisherAware { private volatile AbstractConnectionFactory connectionFactory; @@ -108,6 +111,8 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean getObjectType() { return this.connectionFactory != null ? this.connectionFactory.getClass() @@ -168,6 +173,7 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean message) { TcpConnection connection = null; Assert.notNull(this.clientConnectionFactory, "'clientConnectionFactory' cannot be null"); @@ -334,15 +315,6 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements this.isClientMode = isClientMode; } - /** - * @param scheduler the scheduler to set - * @deprecated Use {@link #setTaskScheduler(TaskScheduler)} - */ - @Deprecated - public void setScheduler(TaskScheduler scheduler) { - this.setTaskScheduler(scheduler); - } - @Override // super class is protected public void setTaskScheduler(TaskScheduler taskScheduler) { super.setTaskScheduler(taskScheduler); 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 d48434dace..54bef29fe4 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 @@ -19,11 +19,14 @@ package org.springframework.integration.ip.tcp.connection; import java.io.IOException; import java.net.Socket; import java.net.SocketException; +import java.net.SocketTimeoutException; import java.nio.channels.CancelledKeyException; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.ServerSocketChannel; import java.nio.channels.SocketChannel; +import java.util.ArrayList; +import java.util.Collections; import java.util.Iterator; import java.util.LinkedList; import java.util.List; @@ -35,6 +38,8 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.context.SmartLifecycle; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; @@ -51,7 +56,7 @@ import org.springframework.util.Assert; * */ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport - implements ConnectionFactory, SmartLifecycle { + implements ConnectionFactory, SmartLifecycle, ApplicationEventPublisherAware { protected static final int DEFAULT_REPLY_TIMEOUT = 10000; @@ -95,7 +100,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport private volatile boolean lookupHost = true; - private volatile List connections = new LinkedList(); + private volatile List connections = new LinkedList(); private volatile TcpSocketSupport tcpSocketSupport = new DefaultTcpSocketSupport(); @@ -105,6 +110,8 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport private volatile int nioHarvestInterval = DEFAULT_NIO_HARVEST_INTERVAL; + private volatile ApplicationEventPublisher applicationEventPublisher; + private static final int DEFAULT_NIO_HARVEST_INTERVAL = 2000; public AbstractConnectionFactory(int port) { @@ -117,6 +124,14 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport this.port = port; } + public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { + this.applicationEventPublisher = applicationEventPublisher; + } + + protected ApplicationEventPublisher getApplicationEventPublisher() { + return applicationEventPublisher; + } + /** * Sets socket attributes on the socket. * @param socket The socket. @@ -290,17 +305,6 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport return mapper; } - /** - * @deprecated This property is no longer used. If you wish - * to use a fixed thread pool, provide your own Executor - * in {@link #setTaskExecutor(Executor)}. - * @return the poolSize - */ - @Deprecated - public int getPoolSize() { - return 0; - } - /** * Registers a TcpListener to receive messages after * the payload has been converted from the input data. @@ -371,16 +375,6 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport } - /** - * @deprecated Default task executor is now a cached rather - * than a fixed pool executor. To use a pool, supply an - * appropriate Executor in {@link AbstractConnectionFactory#setTaskExecutor(Executor)}. - * Use {@link AbstractServerConnectionFactory#setBacklog(int)} to set the connection backlog. - */ - @Deprecated - public void setPoolSize(int poolSize) { - } - public void setInterceptorFactoryChain(TcpConnectionInterceptorFactoryChain interceptorFactoryChain) { this.interceptorFactoryChain = interceptorFactoryChain; } @@ -447,7 +441,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport this.active = false; this.close(); synchronized (this.connections) { - Iterator iterator = this.connections.iterator(); + Iterator iterator = this.connections.iterator(); while (iterator.hasNext()) { TcpConnection connection = iterator.next(); connection.close(); @@ -554,6 +548,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport this.port + " : " + connection.getConnectionId()); } + connection.publishConnectionExceptionEvent(new SocketTimeoutException("Timing out connection")); connection.timeout(); } } @@ -650,7 +645,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport callback.run(); } - protected void addConnection(TcpConnection connection) { + protected void addConnection(TcpConnectionSupport connection) { synchronized (this.connections) { if (!this.active) { connection.close(); @@ -660,18 +655,34 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport } } - protected void harvestClosedConnections() { + /** + * Cleans up this.connections by removing any closed connections. + * @return a list of open connection ids. + */ + private List removeClosedConnectionsAndReturnOpenConnectionIds() { synchronized (this.connections) { - Iterator iterator = this.connections.iterator(); + List openConnectionIds = new ArrayList(); + Iterator iterator = this.connections.iterator(); while (iterator.hasNext()) { TcpConnection connection = iterator.next(); if (!connection.isOpen()) { iterator.remove(); } + else { + openConnectionIds.add(connection.getConnectionId()); + } } + return openConnectionIds; } } + /** + * Cleans up this.connections by removing any closed connections. + */ + protected void harvestClosedConnections() { + this.removeClosedConnectionsAndReturnOpenConnectionIds(); + } + public boolean isRunning() { return this.active; } @@ -705,4 +716,40 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport this.tcpSocketSupport = tcpSocketSupport; } + /** + * Returns a list of (currently) open {@link TcpConnection} connection ids; allows, + * for example, broadcast operations to all open connections. + * @return the list of connection ids. + */ + public List getOpenConnectionIds() { + return Collections.unmodifiableList(this.removeClosedConnectionsAndReturnOpenConnectionIds()); + } + + /** + * Close a connection with the specified connection id. + * @param connectionId the connection id. + * @return true if the connection was closed. + */ + public boolean closeConnection(String connectionId) { + Assert.notNull(connectionId, "'connectionId' to close must not be null"); + synchronized(this.connections) { + boolean closed = false; + for (TcpConnectionSupport connection : connections) { + if (connectionId.equals(connection.getConnectionId())) { + try { + connection.close(); + closed = true; + break; + } + catch (Exception e) { + if (logger.isDebugEnabled()) { + logger.debug("Failed to close connection " + connectionId, e); + } + connection.publishConnectionExceptionEvent(e); + } + } + } + return closed; + } + } } 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 2fbb807acf..6699b18b63 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 @@ -165,17 +165,6 @@ public abstract class AbstractServerConnectionFactory this.backlog = backlog; } - /** - * @deprecated Default task executor is now a cached rather - * than a fixed pool executor. - * Use {@link #setBacklog(int)} to set the connection backlog. - */ - @Override - @Deprecated - public void setPoolSize(int poolSize) { - this.setBacklog(poolSize); - } - public int beforeShutdown() { this.shuttingDown = true; return 0; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java index 6996ed00a2..701265fc64 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java @@ -71,12 +71,10 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact this.pool.setWaitTimeout(connectionWaitTimeout); } - @Override public synchronized void setPoolSize(int poolSize) { this.pool.setPoolSize(poolSize); } - @Override public int getPoolSize() { return this.pool.getPoolSize(); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpNioConnectionSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpNioConnectionSupport.java index 9d8ad36be4..35e70ced0d 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpNioConnectionSupport.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpNioConnectionSupport.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 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. @@ -17,6 +17,8 @@ package org.springframework.integration.ip.tcp.connection; import java.nio.channels.SocketChannel; +import org.springframework.context.ApplicationEventPublisher; + /** * Implementation of {@link TcpNioConnectionSupport} for non-SSL @@ -27,9 +29,9 @@ import java.nio.channels.SocketChannel; */ public class DefaultTcpNioConnectionSupport implements TcpNioConnectionSupport { - public TcpNioConnection createNewConnection(SocketChannel socketChannel, - boolean server, boolean lookupHost) throws Exception { - return new TcpNioConnection(socketChannel, server, lookupHost); + public TcpNioConnection createNewConnection(SocketChannel socketChannel, boolean server, boolean lookupHost, + ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName) throws Exception { + return new TcpNioConnection(socketChannel, server, lookupHost, applicationEventPublisher, connectionFactoryName); } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpNioSSLConnectionSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpNioSSLConnectionSupport.java index 4ffdfcb60d..d1b6993c6d 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpNioSSLConnectionSupport.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpNioSSLConnectionSupport.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 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. @@ -21,6 +21,7 @@ import javax.net.ssl.SSLContext; import javax.net.ssl.SSLEngine; import org.springframework.beans.factory.InitializingBean; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.util.Assert; /** @@ -41,11 +42,14 @@ public class DefaultTcpNioSSLConnectionSupport implements TcpNioConnectionSuppor this.sslContextSupport = sslContextSupport; } - public TcpNioConnection createNewConnection(SocketChannel socketChannel, - boolean server, boolean lookupHost) throws Exception { - + /** + * Creates a {@link TcpNioSSLConnection}. + */ + public TcpNioConnection createNewConnection(SocketChannel socketChannel, boolean server, boolean lookupHost, + ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName) throws Exception { SSLEngine sslEngine = this.sslContext.createSSLEngine(); - TcpNioSSLConnection tcpNioSSLConnection = new TcpNioSSLConnection(socketChannel, server, lookupHost, sslEngine); + TcpNioSSLConnection tcpNioSSLConnection = new TcpNioSSLConnection(socketChannel, server, lookupHost, + applicationEventPublisher, connectionFactoryName, sslEngine); tcpNioSSLConnection.init(); return tcpNioSSLConnection; } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java index b5c326b8ed..1e71cc7878 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java @@ -20,8 +20,6 @@ import java.util.Iterator; import java.util.List; import java.util.UUID; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; import org.springframework.integration.Message; @@ -39,8 +37,6 @@ import org.springframework.util.Assert; */ public class FailoverClientConnectionFactory extends AbstractClientConnectionFactory { - private static final Log logger = LogFactory.getLog(FailoverClientConnectionFactory.class); - private final List factories; public FailoverClientConnectionFactory(List factories) { @@ -279,6 +275,10 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac return this.delegate.getPort(); } + public Object getDeserializerStateKey() { + return this.delegate.getDeserializerStateKey(); + } + @Override public void registerListener(TcpListener listener) { this.delegate.registerListener(listener); 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 427104851f..9311ca7237 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 @@ -114,4 +114,12 @@ public interface TcpConnection extends Runnable { */ long incrementAndGetConnectionSequence(); + /** + * @return a key that can be used to reference state in a {@link Deserializer} that + * maintains state for this connection. Currently, this would be the InputStream + * associated with the connection, but the object should be treated as opaque + * and ONLY used as a key. + */ + Object getDeserializerStateKey(); + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionEvent.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionEvent.java new file mode 100644 index 0000000000..96dfe5319d --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionEvent.java @@ -0,0 +1,91 @@ +/* + * Copyright 2002-2013 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.context.ApplicationEvent; + +/** + * ApplicationEvent representing certain operations on a {@link TcpConnection}. + * @author Gary Russell + * @since 3.0 + * + */ +public class TcpConnectionEvent extends ApplicationEvent { + + private static final long serialVersionUID = 5323436446362192129L; + + /** + * Valid EventTypes - subclasses should provide similar enums for + * their types. + * + */ + public enum TcpConnectionEventType implements EventType { + OPEN, + CLOSE, + EXCEPTION; + } + + private final EventType type; + + private final String connectionFactoryName; + + private final Throwable throwable; + + public TcpConnectionEvent(TcpConnectionSupport connection, EventType type, + String connectionFactoryName) { + super(connection); + this.type = type; + this.throwable = null; + this.connectionFactoryName = connectionFactoryName; + } + + public TcpConnectionEvent(TcpConnectionSupport connection, Throwable t, + String connectionFactoryName) { + super(connection); + this.type = TcpConnectionEventType.EXCEPTION; + this.throwable = t; + this.connectionFactoryName = connectionFactoryName; + } + + public EventType getType() { + return type; + } + + public String getConnectionId() { + return ((TcpConnection) this.getSource()).getConnectionId(); + } + + public String getConnectionFactoryName() { + return connectionFactoryName; + } + + @Override + public String toString() { + return "TcpConnectionEvent [type=" + this.getType() + + ", factory=" + this.connectionFactoryName + + ", connectionId=" + this.getConnectionId() + + (this.throwable == null ? "" : ", Exception=" + this.throwable) + "]"; + } + + /** + * A marker interface allowing the definition of enums for allowed event types. + * + */ + public interface EventType { + + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionEventListeningMessageProducer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionEventListeningMessageProducer.java new file mode 100644 index 0000000000..02c6e54d5b --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionEventListeningMessageProducer.java @@ -0,0 +1,73 @@ +/* + * Copyright 2002-2013 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.HashSet; +import java.util.Set; + +import org.springframework.context.ApplicationListener; +import org.springframework.integration.Message; +import org.springframework.integration.core.MessageProducer; +import org.springframework.integration.endpoint.MessageProducerSupport; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; + +/** + * {@link MessageProducer} that produces Messages with @link {@link TcpConnectionEvent} + * payloads. + * @author Gary Russell + * @since 3.0 + * + */ +public class TcpConnectionEventListeningMessageProducer extends MessageProducerSupport + implements ApplicationListener { + + private volatile Set> eventTypes = + new HashSet>(); + + /** + * Set the list of event types (classes that extend TcpConnectionEvent) that + * this adapter should send to the message channel. By default, all event + * types will be sent. + */ + @SuppressWarnings("unchecked") + public void setEventTypes(Class[] eventTypes) { + Assert.notEmpty(eventTypes, "at least one event type is required"); + Set> eventTypeSet = new HashSet>(); + eventTypeSet.addAll(CollectionUtils.arrayToList(eventTypes)); + this.eventTypes = eventTypeSet; + } + + public void onApplicationEvent(TcpConnectionEvent event) { + if (CollectionUtils.isEmpty(this.eventTypes)) { + this.sendMessage(messageFromEvent(event)); + } + else { + for (Class eventType : this.eventTypes) { + if (eventType.isAssignableFrom(event.getClass())) { + this.sendMessage(messageFromEvent(event)); + break; + } + } + } + } + + protected Message messageFromEvent(TcpConnectionEvent event) { + return MessageBuilder.withPayload(event).build(); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java index a0e4456473..3a988fa4cc 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java @@ -64,6 +64,10 @@ public abstract class TcpConnectionInterceptorSupport extends TcpConnectionSuppo return this.theConnection.getPort(); } + public Object getDeserializerStateKey() { + return this.theConnection.getDeserializerStateKey(); + } + @Override public void registerListener(TcpListener listener) { this.tcpListener = listener; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java index 7022c07432..1d2ec18560 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java @@ -20,13 +20,16 @@ import java.net.InetAddress; import java.net.Socket; import java.net.SocketException; import java.util.UUID; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; import org.springframework.integration.Message; +import org.springframework.integration.ip.tcp.connection.TcpConnectionEvent.TcpConnectionEventType; import org.springframework.integration.ip.tcp.serializer.AbstractByteArraySerializer; import org.springframework.util.Assert; @@ -71,11 +74,32 @@ public abstract class TcpConnectionSupport implements TcpConnection { private volatile String hostAddress = "unknown"; + private volatile String connectionFactoryName = "unknown"; + + private final ApplicationEventPublisher applicationEventPublisher; + + private final AtomicBoolean closePublished = new AtomicBoolean(); + public TcpConnectionSupport() { - server = false; + this.server = false; + this.applicationEventPublisher = null; } - public TcpConnectionSupport(Socket socket, boolean server, boolean lookupHost) { + /** + * Creates a TcpConnectionSupport object and publishes a {@link TcpConnectionEvent#OPEN} + * event, if so configured. + * @param socket the underlying socket. + * @param server true if this connection is a server connection + * @param lookupHost true if reverse lookup of the host name should be performed, + * otherwise, the ip address will be used for identification purposes. + * @param applicationEventPublisher the publisher to which OPEN, CLOSE and EXCEPTION events will + * be sent; may be null if event publishing is not required. + * @param connectionFactoryName the name of the connection factory creating this connection; used + * during event publishing, may be null, in which case "unknown" will be used. + */ + public TcpConnectionSupport(Socket socket, boolean server, boolean lookupHost, + ApplicationEventPublisher applicationEventPublisher, + String connectionFactoryName) { this.server = server; InetAddress inetAddress = socket.getInetAddress(); if (inetAddress != null) { @@ -91,6 +115,11 @@ public abstract class TcpConnectionSupport implements TcpConnection { try { this.soLinger = socket.getSoLinger(); } catch (SocketException e) { } + this.applicationEventPublisher = applicationEventPublisher; + if (connectionFactoryName != null) { + this.connectionFactoryName = connectionFactoryName; + } + this.publishConnectionOpenEvent(); } public void afterSend(Message message) throws Exception { @@ -115,6 +144,10 @@ public abstract class TcpConnectionSupport implements TcpConnection { if (this.sender != null) { this.sender.removeDeadConnection(this); } + // close() may be called multiple times; only publish once + if (!this.closePublished.getAndSet(true)) { + this.publishConnectionCloseEvent(); + } } /** @@ -266,4 +299,51 @@ public abstract class TcpConnectionSupport implements TcpConnection { return this.connectionId; } + protected void publishConnectionOpenEvent() { + TcpConnectionEvent event = new TcpConnectionEvent(this, TcpConnectionEventType.OPEN, + this.connectionFactoryName); + doPublish(event); + } + + protected void publishConnectionCloseEvent() { + TcpConnectionEvent event = new TcpConnectionEvent(this, TcpConnectionEventType.CLOSE, + this.connectionFactoryName); + doPublish(event); + } + + protected void publishConnectionExceptionEvent(Throwable t) { + TcpConnectionEvent event = new TcpConnectionEvent(this, t, + this.connectionFactoryName); + doPublish(event); + } + + /** + * Allow interceptors etc to publish events, perhaps subclasses of + * TcpConnectionEvent. The event source must be this connection. + * @param event the event to publish. + */ + public void publishEvent(TcpConnectionEvent event) { + Assert.isTrue(event.getSource() == this, "Can only publish events with this as the source"); + this.doPublish(event); + } + + private void doPublish(TcpConnectionEvent event) { + try { + if (this.applicationEventPublisher == null) { + logger.warn("No publisher available to publish " + event); + } + else { + this.applicationEventPublisher.publishEvent(event); + } + } + catch (Exception e) { + if (logger.isDebugEnabled()) { + logger.debug("Failed to publish " + event, e); + } + else if (logger.isWarnEnabled()) { + logger.warn("Failed to publish " + event + ":" + e.getMessage()); + } + } + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java index 1456aed85f..4deae52ed6 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2013 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. @@ -16,6 +16,7 @@ package org.springframework.integration.ip.tcp.connection; import java.io.UnsupportedEncodingException; +import java.util.Map; import org.springframework.integration.Message; import org.springframework.integration.MessageHandlingException; @@ -31,7 +32,9 @@ import org.springframework.integration.support.MessageBuilder; * charset (UTF-8 by default). * Inbound messages include headers representing the remote end of the * connection as well as a connection id that can be used by a {@link TcpSender} - * to correlate which connection to send a reply. + * to correlate which connection to send a reply. If applySequence is set, adds + * standard correlationId/sequenceNumber headers allowing for downstream (unbounded) + * resequencing. * @author Gary Russell * @since 2.0 * @@ -50,27 +53,35 @@ public class TcpMessageMapper implements Message message = null; Object payload = connection.getPayload(); if (payload != null) { + MessageBuilder messageBuilder = MessageBuilder.withPayload(payload); String connectionId = connection.getConnectionId(); + messageBuilder + .setHeader(IpHeaders.HOSTNAME, connection.getHostName()) + .setHeader(IpHeaders.IP_ADDRESS, connection.getHostAddress()) + .setHeader(IpHeaders.REMOTE_PORT, connection.getPort()) + .setHeader(IpHeaders.CONNECTION_ID, connectionId); if (this.applySequence) { - message = MessageBuilder.withPayload(payload) - .setHeader(IpHeaders.HOSTNAME, connection.getHostName()) - .setHeader(IpHeaders.IP_ADDRESS, connection.getHostAddress()) - .setHeader(IpHeaders.REMOTE_PORT, connection.getPort()) - .setHeader(IpHeaders.CONNECTION_ID, connectionId) - .setCorrelationId(connectionId) - .setSequenceNumber((int) connection.incrementAndGetConnectionSequence()) - .build(); - } else { - message = MessageBuilder.withPayload(payload) - .setHeader(IpHeaders.HOSTNAME, connection.getHostName()) - .setHeader(IpHeaders.IP_ADDRESS, connection.getHostAddress()) - .setHeader(IpHeaders.REMOTE_PORT, connection.getPort()) - .setHeader(IpHeaders.CONNECTION_ID, connectionId) - .build(); + messageBuilder + .setCorrelationId(connectionId) + .setSequenceNumber((int) connection.incrementAndGetConnectionSequence()); } + Map customHeaders = this.supplyCustomHeaders(connection); + if (customHeaders != null) { + messageBuilder.copyHeadersIfAbsent(customHeaders); + } + message = messageBuilder.build(); } return message; + } + /** + * Override to provide additional headers. The standard headers cannot be overridden + * and any such headers will be ignored if provided in the result. + * @param connection the connection. + * @return A Map of headers to be added to the message. + */ + protected Map supplyCustomHeaders(TcpConnection connection) { + return null; } public Object fromMessage(Message message) throws Exception { 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 7993420fb1..6acaee17cf 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 @@ -60,7 +60,8 @@ public class TcpNetClientConnectionFactory extends } Socket socket = createSocket(this.getHost(), this.getPort()); setSocketAttributes(socket); - TcpConnectionSupport connection = new TcpNetConnection(socket, false, this.isLookupHost()); + TcpConnectionSupport connection = new TcpNetConnection(socket, false, this.isLookupHost(), + this.getApplicationEventPublisher(), this.getComponentName()); connection = wrapConnection(connection); initializeConnection(connection, socket); this.getTaskExecutor().execute(connection); 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 e9a4fd310b..2472f9a747 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 @@ -20,6 +20,7 @@ import java.net.Socket; import java.net.SocketException; import java.net.SocketTimeoutException; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; import org.springframework.integration.Message; @@ -47,9 +48,31 @@ public class TcpNetConnection extends TcpConnectionSupport { * @param socket the socket * @param server if true this connection was created as * a result of an incoming request. + * @param lookupHost true if hostname lookup should be performed, otherwise the connection will + * be identified using the ip address. + * @deprecated Use {@link #TcpNetConnection(Socket, boolean, boolean, ApplicationEventPublisher, String)} + * TODO: Remove in 3.1/4.0 */ + @Deprecated public TcpNetConnection(Socket socket, boolean server, boolean lookupHost) { - super(socket, server, lookupHost); + this(socket, server, lookupHost, null, null); + } + + /** + * Constructs a TcpNetConnection for the socket. + * @param socket the socket + * @param server if true this connection was created as + * a result of an incoming request. + * @param lookupHost true if hostname lookup should be performed, otherwise the connection will + * be identified using the ip address. + * @param applicationEventPublisher the publisher to which OPEN, CLOSE and EXCEPTION events will + * be sent; may be null if event publishing is not required. + * @param connectionFactoryName the name of the connection factory creating this connection; used + * during event publishing, may be null, in which case "unknown" will be used. + */ + public TcpNetConnection(Socket socket, boolean server, boolean lookupHost, + ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName) { + super(socket, server, lookupHost, applicationEventPublisher, connectionFactoryName); this.socket = socket; } @@ -73,7 +96,13 @@ public class TcpNetConnection extends TcpConnectionSupport { public synchronized void send(Message message) throws Exception { Object object = this.getMapper().fromMessage(message); this.lastSend = System.currentTimeMillis(); - ((Serializer) this.getSerializer()).serialize(object, this.socket.getOutputStream()); + try { + ((Serializer) this.getSerializer()).serialize(object, this.socket.getOutputStream()); + } + catch (Exception e) { + this.publishConnectionExceptionEvent(e); + throw e; + } this.afterSend(message); } @@ -85,6 +114,15 @@ public class TcpNetConnection extends TcpConnectionSupport { return this.socket.getPort(); } + public Object getDeserializerStateKey() { + try { + return this.socket.getInputStream(); + } + catch (Exception e) { + return null; + } + } + /** * If there is no listener, and this connection is not for single use, * this method exits. When there is a listener, the method runs in a @@ -103,7 +141,9 @@ public class TcpNetConnection extends TcpConnectionSupport { return; } boolean okToRun = true; - logger.debug("Reading..."); + if (logger.isDebugEnabled()) { + logger.debug(this.getConnectionId() + " Reading..."); + } boolean intercepted = false; while (okToRun) { Message message = null; @@ -112,6 +152,7 @@ public class TcpNetConnection extends TcpConnectionSupport { this.lastRead = System.currentTimeMillis(); } catch (Exception e) { + this.publishConnectionExceptionEvent(e); if (handleReadException(e)) { okToRun = false; } 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 c64af71b6b..631628ca67 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 @@ -98,7 +98,8 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto logger.debug("Accepted connection from " + socket.getInetAddress().getHostAddress()); } setSocketAttributes(socket); - TcpConnectionSupport connection = new TcpNetConnection(socket, true, this.isLookupHost()); + TcpConnectionSupport connection = new TcpNetConnection(socket, true, this.isLookupHost(), + this.getApplicationEventPublisher(), this.getComponentName()); connection = wrapConnection(connection); this.initializeConnection(connection, socket); this.getTaskExecutor().execute(connection); 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 8846b4e727..ff2ec2c4ea 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 @@ -89,7 +89,7 @@ public class TcpNioClientConnectionFactory extends SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress(this.getHost(), this.getPort())); setSocketAttributes(socketChannel.socket()); TcpNioConnection connection = this.tcpNioConnectionSupport.createNewConnection( - socketChannel, false, this.isLookupHost()); + socketChannel, false, this.isLookupHost(), this.getApplicationEventPublisher(), this.getComponentName()); connection.setUsingDirectBuffers(this.usingDirectBuffers); connection.setTaskExecutor(this.getTaskExecutor()); TcpConnectionSupport wrappedConnection = wrapConnection(connection); 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 374827f52a..b2e0bcf6f3 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 @@ -33,6 +33,7 @@ import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.core.serializer.Serializer; import org.springframework.integration.Message; import org.springframework.integration.MessagingException; @@ -79,9 +80,24 @@ public class TcpNioConnection extends TcpConnectionSupport { * @param socketChannel the socketChannel * @param server if true this connection was created as * a result of an incoming request. + * @deprecated Use {@link #TcpNioConnection(SocketChannel, boolean, boolean, ApplicationEventPublisher, String)} + * TODO: Remove in 3.1/4.0 */ + @Deprecated public TcpNioConnection(SocketChannel socketChannel, boolean server, boolean lookupHost) throws Exception { - super(socketChannel.socket(), server, lookupHost); + this(socketChannel, server, lookupHost, null, null); + } + + /** + * Constructs a TcpNetConnection for the SocketChannel. + * @param socketChannel the socketChannel + * @param server if true this connection was created as + * a result of an incoming request. + */ + public TcpNioConnection(SocketChannel socketChannel, boolean server, boolean lookupHost, + ApplicationEventPublisher applicationEventPublisher, + String connectionFactoryName) throws Exception { + super(socketChannel.socket(), server, lookupHost, applicationEventPublisher, connectionFactoryName); this.socketChannel = socketChannel; int receiveBufferSize = socketChannel.socket().getReceiveBufferSize(); if (receiveBufferSize <= 0) { @@ -118,7 +134,13 @@ public class TcpNioConnection extends TcpConnectionSupport { synchronized(this.getMapper()) { Object object = this.getMapper().fromMessage(message); this.lastSend = System.currentTimeMillis(); - ((Serializer) this.getSerializer()).serialize(object, this.getChannelOutputStream()); + try { + ((Serializer) this.getSerializer()).serialize(object, this.getChannelOutputStream()); + } + catch (Exception e) { + this.publishConnectionExceptionEvent(e); + throw e; + } this.afterSend(message); } } @@ -131,6 +153,10 @@ public class TcpNioConnection extends TcpConnectionSupport { return this.socketChannel.socket().getPort(); } + public Object getDeserializerStateKey() { + return this.channelInputStream; + } + /** * Allocates a ByteBuffer of the requested length using normal or * direct buffers, depending on the usingDirectBuffers field. @@ -326,7 +352,12 @@ public class TcpNioConnection extends TcpConnectionSupport { throw new MessagingException("Timed out writing to ChannelInputStream, probably due to insufficient threads in " + "a fixed thread pool; consider increasing this task executor pool size"); } - } finally { + } + catch (Exception e) { + this.publishConnectionExceptionEvent(e); + throw e; + } + finally { this.writingToPipe = false; } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionSupport.java index 49273ed69d..3bbfcda1b7 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionSupport.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionSupport.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 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. @@ -17,6 +17,8 @@ package org.springframework.integration.ip.tcp.connection; import java.nio.channels.SocketChannel; +import org.springframework.context.ApplicationEventPublisher; + /** * Used by NIO connection factories to instantiate a {@link TcpNioConnection} object. @@ -27,7 +29,21 @@ import java.nio.channels.SocketChannel; */ public interface TcpNioConnectionSupport { + /** + * Create a new {@link TcpNioConnection} object wrapping the {@link SocketChannel} + * @param socketChannel the SocketChannel. + * @param server true if this connection is a server connection. + * @param lookupHost true if hostname lookup should be performed, otherwise the connection will + * be identified using the ip address. + * @param applicationEventPublisher the publisher to which OPEN, CLOSE and EXCEPTION events will + * be sent; may be null if event publishing is not required. + * @param connectionFactoryName the name of the connection factory creating this connection; used + * during event publishing, may be null, in which case "unknown" will be used. + * @return the TcpNioConnection + * @throws Exception + */ TcpNioConnection createNewConnection(SocketChannel socketChannel, - boolean server, boolean lookupHost) throws Exception; - + boolean server, boolean lookupHost, + ApplicationEventPublisher applicationEventPublisher, + String connectionFactoryName) throws Exception; } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioSSLConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioSSLConnection.java index f0d2974a4d..7b38b2af5d 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioSSLConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioSSLConnection.java @@ -27,6 +27,7 @@ import javax.net.ssl.SSLEngineResult.HandshakeStatus; import javax.net.ssl.SSLEngineResult.Status; import javax.net.ssl.SSLException; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.integration.MessagingException; import org.springframework.util.Assert; @@ -66,12 +67,25 @@ public class TcpNioSSLConnection extends TcpNioConnection { private boolean needMoreNetworkData; + /** + * @deprecated Use {@link #TcpNioSSLConnection(SocketChannel, boolean, boolean, ApplicationEventPublisher, String, SSLEngine)} + * TODO: Remove in 3.1/4.0 + */ + @Deprecated public TcpNioSSLConnection(SocketChannel socketChannel, boolean server, boolean lookupHost, SSLEngine sslEngine) throws Exception { - super(socketChannel, server, lookupHost); + super(socketChannel, server, lookupHost, null, null); this.sslEngine = sslEngine; } + public TcpNioSSLConnection(SocketChannel socketChannel, boolean server, boolean lookupHost, + ApplicationEventPublisher applicationEventPublisher, String connectionFactoryName, + SSLEngine sslEngine) throws Exception { + super(socketChannel, server, lookupHost, applicationEventPublisher, connectionFactoryName); + this.sslEngine = sslEngine; + } + + /** * Overrides super class method to perform decryption and/or participate * in handshaking. Decrypted data is sent to the super class to be 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 0e027405e9..7485c5d614 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 @@ -174,7 +174,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto try { TcpNioConnection connection = this.tcpNioConnectionSupport .createNewConnection(socketChannel, true, - this.isLookupHost()); + this.isLookupHost(), this.getApplicationEventPublisher(), this.getComponentName()); connection.setUsingDirectBuffers(this.usingDirectBuffers); TcpConnectionSupport wrappedConnection = wrapConnection(connection); this.initializeConnection(wrappedConnection, socketChannel.socket()); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java index f630962a79..debe05f81c 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2013 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. @@ -24,19 +24,19 @@ import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; /** - * Base class for (de)serializers that provide a mechanism to + * Base class for (de)serializers that provide a mechanism to * reconstruct a byte array from an arbitrary stream. - * + * * @author Gary Russell * @since 2.0 * */ public abstract class AbstractByteArraySerializer implements - Serializer, + Serializer, Deserializer { protected int maxMessageSize = 2048; - + protected final Log logger = LogFactory.getLog(this.getClass()); /** @@ -59,9 +59,25 @@ public abstract class AbstractByteArraySerializer implements protected void checkClosure(int bite) throws IOException { if (bite < 0) { - logger.debug("Socket closed during message assembly"); + logger.debug("Socket closed during message assembly"); throw new IOException("Socket closed during message assembly"); } } + /** + * Copy size bytes to a new buffer exactly size bytes long. + * @param buffer The buffer containing the data. + * @param size The number of bytes to copy. + * @return The new buffer, or the buffer parameter if it is + * already the correct size. + */ + protected byte[] copyToSizedArray(byte[] buffer, int size) { + if (size == buffer.length) { + return buffer; + } + byte[] assembledData = new byte[size]; + System.arraycopy(buffer, 0, assembledData, 0, size); + return assembledData; + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java index 9d99a53477..4bdc94fd92 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2013 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. @@ -40,6 +40,13 @@ public class ByteArrayCrLfSerializer extends AbstractByteArraySerializer { */ public byte[] deserialize(InputStream inputStream) throws IOException { byte[] buffer = new byte[this.maxMessageSize]; + int n = this.fillToCrLf(inputStream, buffer); + byte[] assembledData = this.copyToSizedArray(buffer, n); + return assembledData; + } + + public int fillToCrLf(InputStream inputStream, byte[] buffer) + throws IOException, SoftEndOfStreamException { int n = 0; int bite; if (logger.isDebugEnabled()) { @@ -61,9 +68,7 @@ public class ByteArrayCrLfSerializer extends AbstractByteArraySerializer { + this.maxMessageSize); } }; - byte[] assembledData = new byte[n-1]; - System.arraycopy(buffer, 0, assembledData, 0, n-1); - return assembledData; + return n-1; // trim \r } /** diff --git a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-3.0.xsd b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-3.0.xsd index c3b2f079cc..041899e01d 100644 --- a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-3.0.xsd +++ b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-3.0.xsd @@ -517,16 +517,6 @@ cached thread pool task executor is used. - - - -Deprecated since 2.2; previously it specified the thread pool size if -an external task executor was not provided; it was also used to set -the connection backlog for server factories. That setting is now -specified by the backlog attribute. - - - @@ -606,6 +596,72 @@ used to create SSLServerSocketFactory and SSLSocketFactory instances. + + + +A reference to a TcpMessageMapper implementation. Allows customization of headers in inbound messages by overriding +setCustomHeaders(). Default is TcpMessageMapper. + + + + + + + + + + + + + + + Configures an inbound Channel Adapter which listens for TCP Connection + events, converts them to Messages and sends them to a Message Channel. + + + + + + + + Comma delimited list of event types (classes that extend TcpConnectionEvent) that this adapter + should send to the message channel. By default, all event types will be sent [OPTIONAL]. + Note, it is NOT possible to filter by subtype, just class - for example, the standard TcpConnectionEvent + class has 3 subtypes (OPEN, CLOSE, EXCEPTION). This feature is intended to allow the adapter to + be used, say, to obtain just subclasses of TcpConnectionEvent (perhaps generated by a + TcpConnectionInterceptor, perhaps to signal handshaking of some kind). + + + + + + + + + + + + The channel to which Messages generated from Application Context events will be sent. + + + + + + + + + + + + If a (synchronous) downstream exception is thrown and an error-channel is specified, + a MessagingException will be sent to this channel. Otherwise, any such exception + will be propagated to the caller. + + + + 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 02bb17fa62..549ec82a5a 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 @@ -59,8 +59,11 @@ lookup-host="false" apply-sequence="true" ssl-context-support="sslContextSupport" + mapper="mapper" /> + + - - + + \ No newline at end of file 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 02907b22d7..4377050533 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 @@ -53,6 +53,10 @@ import org.springframework.integration.ip.tcp.connection.AbstractConnectionFacto import org.springframework.integration.ip.tcp.connection.DefaultTcpNetSSLSocketFactorySupport; import org.springframework.integration.ip.tcp.connection.DefaultTcpNioSSLConnectionSupport; import org.springframework.integration.ip.tcp.connection.DefaultTcpSSLContextSupport; +import org.springframework.integration.ip.tcp.connection.TcpConnectionEvent; +import org.springframework.integration.ip.tcp.connection.TcpConnectionEventListeningMessageProducer; +import org.springframework.integration.ip.tcp.connection.TcpConnectionSupport; +import org.springframework.integration.ip.tcp.connection.TcpMessageMapper; import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory; @@ -168,9 +172,6 @@ public class ParserUnitTests { @Autowired AbstractConnectionFactory server1; - @Autowired - AbstractConnectionFactory serverBackwardsCompatible; - @Autowired AbstractConnectionFactory server2; @@ -255,6 +256,12 @@ public class ParserUnitTests { @Autowired TcpSSLContextSupport contextSupport; + @Autowired + TcpMessageMapper mapper; + + @Autowired + TcpConnectionEventListeningMessageProducer eventAdapter; + private static volatile int adviceCalled; @Test @@ -304,8 +311,9 @@ public class ParserUnitTests { assertFalse(cfS1.isLookupHost()); assertFalse(tcpIn.isAutoStartup()); assertEquals(124, tcpIn.getPhase()); - assertTrue((Boolean) TestUtils.getPropertyValue( - TestUtils.getPropertyValue(cfS1, "mapper"), "applySequence")); + TcpMessageMapper cfS1Mapper = TestUtils.getPropertyValue(cfS1, "mapper", TcpMessageMapper.class); + assertSame(mapper, cfS1Mapper); + assertTrue((Boolean) TestUtils.getPropertyValue(cfS1Mapper, "applySequence")); Object socketSupport = TestUtils.getPropertyValue(cfS1, "tcpSocketFactorySupport"); assertTrue(socketSupport instanceof DefaultTcpNetSSLSocketFactorySupport); assertNotNull(TestUtils.getPropertyValue(socketSupport, "sslContext")); @@ -507,13 +515,6 @@ public class ParserUnitTests { assertNotNull(dfa.getPropertyValue("interceptorFactoryChain")); } - @Test - public void testConnDeprecatedPoolSize() { - assertTrue(serverBackwardsCompatible instanceof TcpNetServerConnectionFactory); - DirectFieldAccessor dfa = new DirectFieldAccessor(serverBackwardsCompatible); - assertEquals(123, dfa.getPropertyValue("backlog")); - } - @Test public void testConnClient2() { assertTrue(client2 instanceof TcpNetClientConnectionFactory); @@ -648,6 +649,14 @@ public class ParserUnitTests { assertSame(socketSupport, dfa.getPropertyValue("tcpSocketSupport")); } + @Test + public void testEventAdapter() { + Set eventTypes = TestUtils.getPropertyValue(this.eventAdapter, "eventTypes", Set.class); + assertEquals(2, eventTypes.size()); + assertTrue(eventTypes.contains(EventSubclass1.class)); + assertTrue(eventTypes.contains(EventSubclass2.class)); + } + public static class FooAdvice extends AbstractRequestHandlerAdvice { @Override @@ -657,4 +666,20 @@ public class ParserUnitTests { } } + + @SuppressWarnings("serial") + public static class EventSubclass1 extends TcpConnectionEvent { + + public EventSubclass1(TcpConnectionSupport connection, EventType type, String connectionFactoryName) { + super(connection, type, connectionFactoryName); + } + } + + @SuppressWarnings("serial") + public static class EventSubclass2 extends TcpConnectionEvent { + + public EventSubclass2(TcpConnectionSupport connection, EventType type, String connectionFactoryName) { + super(connection, type, connectionFactoryName); + } + } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests-context.xml index ed8ed9b0d4..dc8857782c 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests-context.xml +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests-context.xml @@ -33,15 +33,21 @@ /> - + + + + + + + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java index 930ac2ef56..0aae75c482 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java @@ -35,6 +35,8 @@ import org.springframework.integration.history.MessageHistory; import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpConnection; +import org.springframework.integration.ip.tcp.connection.TcpConnectionEvent; +import org.springframework.integration.ip.tcp.connection.TcpConnectionEvent.TcpConnectionEventType; import org.springframework.integration.ip.tcp.serializer.ByteArrayRawSerializer; import org.springframework.integration.ip.util.TestingUtilities; import org.springframework.integration.support.MessageBuilder; @@ -64,6 +66,9 @@ public class ConnectionToConnectionTests { @Autowired private QueueChannel serverSideChannel; + @Autowired + private QueueChannel events; + // Test jvm shutdown public static void main(String[] args) { ConfigurableApplicationContext ctx = new ClassPathXmlApplicationContext( @@ -78,6 +83,7 @@ public class ConnectionToConnectionTests { ctx.close(); } + @SuppressWarnings("unchecked") @Test public void testConnect() throws Exception { TestingUtilities.waitListening(server, null); @@ -95,6 +101,39 @@ public class ConnectionToConnectionTests { assertNotNull(message); assertEquals("Test", new String((byte[]) message.getPayload())); } + int clientOpens = 0; + int clientCloses = 0; + int serverOpens = 0; + int serverCloses = 0; + int clientExceptions = 0; + Message eventMessage; + while ((eventMessage = (Message) events.receive(1000)) != null) { + TcpConnectionEvent event = eventMessage.getPayload(); + if ("client".equals(event.getConnectionFactoryName())) { + if (TcpConnectionEventType.OPEN == event.getType()) { + clientOpens++; + } + else if (TcpConnectionEventType.CLOSE == event.getType()) { + clientCloses++; + } + else if (TcpConnectionEventType.EXCEPTION == event.getType()) { + clientExceptions++; + } + } + else if ("server".equals(event.getConnectionFactoryName())) { + if (TcpConnectionEventType.OPEN == event.getType()) { + serverOpens++; + } + else if (TcpConnectionEventType.CLOSE == event.getType()) { + serverCloses++; + } + } + } + assertEquals(100, clientOpens); + assertEquals(100, clientCloses); + assertEquals(100, clientExceptions); + assertEquals(100, serverOpens); + assertEquals(100, serverCloses); } @Test diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java new file mode 100644 index 0000000000..2f4c8419a7 --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java @@ -0,0 +1,72 @@ +/* + * Copyright 2002-2013 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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.fail; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; + +import java.io.OutputStream; +import java.net.Socket; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.Test; +import org.mockito.Mockito; +import org.springframework.context.ApplicationEvent; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.core.serializer.Serializer; +import org.springframework.integration.message.GenericMessage; + +/** + * @author Gary Russell + * @since 3.0 + * + */ +public class ConnectionEventTests { + + @Test + public void test() throws Exception { + Socket socket = mock(Socket.class); + final AtomicReference theEvent = new AtomicReference(); + TcpNetConnection conn = new TcpNetConnection(socket, false, false, new ApplicationEventPublisher() { + + public void publishEvent(ApplicationEvent event) { + theEvent.set(event); + } + }, "foo"); + assertNotNull(theEvent.get()); + assertEquals("TcpConnectionEvent [type=OPEN, factory=foo, connectionId=" + conn.getConnectionId() + "]", theEvent.get().toString()); + @SuppressWarnings("unchecked") + Serializer serializer = mock(Serializer.class); + doThrow(new RuntimeException("foo")).when(serializer).serialize(Mockito.any(Object.class), Mockito.any(OutputStream.class)); + conn.setMapper(new TcpMessageMapper()); + conn.setSerializer(serializer); + try { + conn.send(new GenericMessage("bar")); + fail("Expected exception"); + } + catch (Exception e) {} + assertNotNull(theEvent.get()); + assertEquals("TcpConnectionEvent [type=EXCEPTION, factory=foo, connectionId=" + conn.getConnectionId() + + ", Exception=java.lang.RuntimeException: foo]", theEvent.get().toString()); + conn.close(); + assertNotNull(theEvent.get()); + assertEquals("TcpConnectionEvent [type=CLOSE, factory=foo, connectionId=" + conn.getConnectionId() + "]", theEvent.get().toString()); + } + +} diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java new file mode 100644 index 0000000000..5d29394d81 --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java @@ -0,0 +1,106 @@ +/* + * Copyright 2002-2013 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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import static org.mockito.Mockito.mock; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import org.junit.Test; +import org.springframework.context.ApplicationEvent; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.integration.Message; +import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter; +import org.springframework.integration.ip.tcp.connection.TcpConnectionEvent.TcpConnectionEventType; +import org.springframework.integration.ip.util.TestingUtilities; +import org.springframework.integration.test.util.SocketUtils; + +/** + * @author Gary Russell + * @since 3.0 + * + */ +public class ConnectionFactoryTests { + + @Test + public void testObtainConnectionIds() throws Exception { + final List events = + Collections.synchronizedList(new ArrayList()); + ApplicationEventPublisher publisher = new ApplicationEventPublisher() { + public void publishEvent(ApplicationEvent event) { + events.add((TcpConnectionEvent) event); + } + }; + int port = SocketUtils.findAvailableServerSocket(); + TcpNetServerConnectionFactory serverFactory = new TcpNetServerConnectionFactory(port); + serverFactory.setBeanName("serverFactory"); + serverFactory.setApplicationEventPublisher(publisher); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(serverFactory); + adapter.start(); + TestingUtilities.waitListening(serverFactory, null); + TcpNetClientConnectionFactory clientFactory = new TcpNetClientConnectionFactory("localhost", port); + clientFactory.registerListener(new TcpListener() { + public boolean onMessage(Message message) { + return false; + } + }); + clientFactory.setBeanName("clientFactory"); + clientFactory.setApplicationEventPublisher(publisher); + clientFactory.start(); + TcpConnectionSupport client = clientFactory.getConnection(); + List clients = clientFactory.getOpenConnectionIds(); + assertEquals(1, clients.size()); + assertTrue(clients.contains(client.getConnectionId())); + List servers = serverFactory.getOpenConnectionIds(); + assertEquals(1, servers.size()); + assertTrue(serverFactory.closeConnection(servers.get(0))); + servers = serverFactory.getOpenConnectionIds(); + assertEquals(0, servers.size()); + Thread.sleep(1000); + clients = clientFactory.getOpenConnectionIds(); + assertEquals(0, clients.size()); + assertEquals(6, events.size()); // OPEN, CLOSE, EXCEPTION for each side + + FooEvent event = new FooEvent(client, TcpConnectionEventType.OPEN, "foo"); + client.publishEvent(event); + assertEquals(7, events.size()); + + try { + event = new FooEvent(mock(TcpConnectionSupport.class), TcpConnectionEventType.OPEN, "foo"); + client.publishEvent(event); + fail("Expected exception"); + } + catch (IllegalArgumentException e) { + assertTrue("Can only publish events with this as the source".equals(e.getMessage())); + } + } + + @SuppressWarnings("serial") + private class FooEvent extends TcpConnectionEvent { + + public FooEvent(TcpConnectionSupport connection, EventType type, String connectionFactoryName) { + super(connection, type, connectionFactoryName); + } + + } + +} diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpConnectionEventListenerTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpConnectionEventListenerTests.java new file mode 100644 index 0000000000..fa12601e41 --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpConnectionEventListenerTests.java @@ -0,0 +1,104 @@ +/* + * Copyright 2002-2013 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 static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; + +import org.junit.Test; +import org.mockito.Mockito; +import org.springframework.integration.Message; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.ip.tcp.connection.TcpConnectionEvent.TcpConnectionEventType; + +/** + * @author Gary Russell + * @since 3.0 + * + */ +public class TcpConnectionEventListenerTests { + + @Test + public void testNoFilter() { + TcpConnectionEventListeningMessageProducer eventProducer = new TcpConnectionEventListeningMessageProducer(); + QueueChannel outputChannel = new QueueChannel(); + eventProducer.setOutputChannel(outputChannel); + eventProducer.afterPropertiesSet(); + TcpConnectionSupport connection = Mockito.mock(TcpConnectionSupport.class); + TcpConnectionEvent event1 = new TcpConnectionEvent(connection, TcpConnectionEventType.OPEN, "foo"); + eventProducer.onApplicationEvent(event1); + FooEvent event2 = new FooEvent(connection, TcpConnectionEventType.OPEN, "foo"); + eventProducer.onApplicationEvent(event2); + BarEvent event3 = new BarEvent(connection, TcpConnectionEventType.OPEN, "foo"); + eventProducer.onApplicationEvent(event3); + Message message = outputChannel.receive(0); + assertNotNull(message); + assertSame(event1, message.getPayload()); + message = outputChannel.receive(0); + assertNotNull(message); + assertSame(event2, message.getPayload()); + message = outputChannel.receive(0); + assertNotNull(message); + assertSame(event3, message.getPayload()); + message = outputChannel.receive(0); + assertNull(message); + } + + @SuppressWarnings("unchecked") + @Test + public void testFilter() { + TcpConnectionEventListeningMessageProducer eventProducer = new TcpConnectionEventListeningMessageProducer(); + QueueChannel outputChannel = new QueueChannel(); + eventProducer.setOutputChannel(outputChannel); + eventProducer.setEventTypes(new Class[] {FooEvent.class, BarEvent.class}); + eventProducer.afterPropertiesSet(); + TcpConnectionSupport connection = Mockito.mock(TcpConnectionSupport.class); + TcpConnectionEvent event1 = new TcpConnectionEvent(connection, TcpConnectionEventType.OPEN, "foo"); + eventProducer.onApplicationEvent(event1); + FooEvent event2 = new FooEvent(connection, TcpConnectionEventType.OPEN, "foo"); + eventProducer.onApplicationEvent(event2); + BarEvent event3 = new BarEvent(connection, TcpConnectionEventType.OPEN, "foo"); + eventProducer.onApplicationEvent(event3); + Message message = outputChannel.receive(0); + assertNotNull(message); + assertSame(event2, message.getPayload()); + message = outputChannel.receive(0); + assertNotNull(message); + assertSame(event3, message.getPayload()); + message = outputChannel.receive(0); + assertNull(message); + } + + @SuppressWarnings("serial") + private class FooEvent extends TcpConnectionEvent { + + public FooEvent(TcpConnectionSupport connection, EventType type, String connectionFactoryName) { + super(connection, type, connectionFactoryName); + } + + } + + @SuppressWarnings("serial") + private class BarEvent extends TcpConnectionEvent { + + public BarEvent(TcpConnectionSupport connection, EventType type, String connectionFactoryName) { + super(connection, type, connectionFactoryName); + } + + } + +} diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java index 50846a58f7..ed88488195 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java @@ -16,10 +16,13 @@ package org.springframework.integration.ip.tcp.connection; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; import java.net.Socket; +import java.util.Collections; +import java.util.Map; import javax.net.SocketFactory; @@ -30,13 +33,11 @@ import org.springframework.integration.support.MessageBuilder; /** * @author Gary Russell + * @since 2.0 * */ public class TcpMessageMapperTests { - /** - * - */ private static final String TEST_PAYLOAD = "abcdefghijkl"; @Test @@ -63,7 +64,7 @@ public class TcpMessageMapperTests { TcpMessageMapper mapper = new TcpMessageMapper(); Socket socket = SocketFactory.getDefault().createSocket(); - TcpConnection connection = new TcpConnectionSupport(socket, false, false) { + TcpConnection connection = new TcpConnectionSupport(socket, false, false, null, null) { public void run() { } public void send(Message message) throws Exception { @@ -89,6 +90,9 @@ public class TcpMessageMapperTests { public String getConnectionId() { return "anId"; } + public Object getDeserializerStateKey() { + return null; + } }; Message message = mapper.toMessage(connection); assertEquals(TEST_PAYLOAD, new String((byte[]) message.getPayload())); @@ -111,11 +115,18 @@ public class TcpMessageMapperTests { } @Test - public void testToMessageSequenceNew() throws Exception { - TcpMessageMapper mapper = new TcpMessageMapper(); + public void testToMessageSequenceNewWithCustomHeader() throws Exception { + TcpMessageMapper mapper = new TcpMessageMapper() { + + @Override + protected Map supplyCustomHeaders(TcpConnection connection) { + return Collections.singletonMap("foo", "bar"); + } + + }; mapper.setApplySequence(true); Socket socket = SocketFactory.getDefault().createSocket(); - TcpConnection connection = new TcpConnectionSupport(socket, false, false) { + TcpConnection connection = new TcpConnectionSupport(socket, false, false, null, null) { public void run() { } public void send(Message message) throws Exception { @@ -141,6 +152,9 @@ public class TcpMessageMapperTests { public String getConnectionId() { return "anId"; } + public Object getDeserializerStateKey() { + return null; + } }; Message message = mapper.toMessage(connection); assertEquals(TEST_PAYLOAD, new String((byte[]) message.getPayload())); @@ -166,6 +180,8 @@ public class TcpMessageMapperTests { .getHeaders().getSequenceNumber()); assertEquals(message.getHeaders().get(IpHeaders.CONNECTION_ID), message .getHeaders().getCorrelationId()); + assertNotNull(message.getHeaders().get("foo")); + assertEquals("bar", message.getHeaders().get("foo")); } @@ -191,5 +207,4 @@ public class TcpMessageMapperTests { } - } \ No newline at end of file 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 f91c0eb134..590351f157 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 @@ -262,7 +262,7 @@ public class TcpNioConnectionTests { } }).when(channel).read(Mockito.any(ByteBuffer.class)); when(socket.getReceiveBufferSize()).thenReturn(1024); - final TcpNioConnection connection = new TcpNioConnection(channel, false, false); + final TcpNioConnection connection = new TcpNioConnection(channel, false, false, null, null); connection.setTaskExecutor(exec); connection.setPipeTimeout(200); Method method = TcpNioConnection.class.getDeclaredMethod("doRead"); @@ -309,7 +309,7 @@ public class TcpNioConnectionTests { return 1027; } }).when(channel).read(Mockito.any(ByteBuffer.class)); - final TcpNioConnection connection = new TcpNioConnection(channel, false, false); + final TcpNioConnection connection = new TcpNioConnection(channel, false, false, null, null); connection.setTaskExecutor(exec); connection.registerListener(new TcpListener(){ public boolean onMessage(Message message) { diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index d437851637..9a9de7ddba 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -12,6 +12,46 @@
New Components +
+ TCP/IP Connection Events and Connection Management + + The (supplied) TcpConnections now emit + ApplicationEvents (specifically + TcpConnectionEvents) when connections are + opened, closed, or an exception occurs. This allows applications + to be informed of changes to TCP connections using the normal + Spring ApplicationListener + mechanism. + + + AbstractTcpConnection has been renamed + TcpConnectionSupport; custom connections that + are subclasses of this class, can use its methods to publish events. + Similarly, AbstractTcpConnectionInterceptor has + been renamed to TcpConnectionInterceptorSupport. + + + In addition, a new <int-ip:tcp-connection-event-channel-adapter/> + is provided; by default, this adapter sends all TcpConnectionEvents + to a Channel. + + + Further, the TCP Connection Factories, now provide a new method + getOpenConnectionIds(), which returns a list of identifiers for all + open connections; this allows applications, for example, to broadcast to all + open connections. + + + Finally, the connection factories also provide a new method + closeConnection(String connectionId) which allows applications + to explicitly close a connection using its ID. + + + Further documentation for these features will follow; for now, refer to the + schema documentation for information about the new adapter, and JavaDocs for + it as well as the other features. + +