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

INT-2126 Add TcpConnection Event Publisher

Browse Source 
Reference: https://jira.springsource.org/browse/INT-2126

* Add TcpConnectionEvent for OPEN, CLOSE, EXCEPTION.
* Add Event Type Enum - Provides type safety for event types.
* Add <int-ip:tcp-connection-event-inbound-channel-adapter />
* Add the ability to filter events (restrict to one or more subclasses)
* Add publishEvent() to TcpConnectionSupport to permit, for example, connection interceptors to publish events via the connection, but the event source can only be the connection used to publish the event.
* Add documentation to Reference Documentation section *What's New*

INT-2871 Provide Mechanism to Find TCP Connections
Reference: https://jira.springsource.org/browse/INT-2871

* Add getOpenConnectionIds() to Abstract Connection Factory
* Add closeConnection(String connectionID)
* Introduce removeClosedConnectionsAndReturnOpenConnectionIds

INT-2877 TCP Extension Improvements
Reference: https://jira.springsource.org/browse/INT-2877

* Make improvements to make extension easier.
* Add setCustomHeaders() to mapper.
* Add mapper to connection factory namespace parser.
* Add plumbing for stateful Deserializers.
* Refactor setting standard headers
* rename setCustomHeaders to supplyCustomHeaders - returning a Map and not exposing the MessageBuilder to subclasses

INT-2872 Remove Deprecated Items in ip Module
Reference: https://jira.springsource.org/browse/INT-2872

* pool-size Attribute
* setScheduler
* TcpSendingMessageHandler.getConnection()
* Deprecated attribute on TCP Connection Factory.
This commit is contained in:
Gary Russell
2013-01-04 14:44:37 -05:00
committed by Gunnar Hillert
parent 0374b56ee1
commit b045e8c2be
42 changed files with 1100 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java
View File

@@ -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() {}
/**

7
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpNamespaceHandler.java
View File

@@ -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 <em>ip</em> 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());
}
}

43
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionEventInboundChannelAdapterParser.java Normal file
View File

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

25
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java
View File

@@ -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<AbstractConnectionFactory> implements SmartLifecycle, BeanNameAware {
public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<AbstractConnectionFactory> implements SmartLifecycle, BeanNameAware,
ApplicationEventPublisherAware {
private volatile AbstractConnectionFactory connectionFactory;
@@ -108,6 +111,8 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
private volatile TcpSocketFactorySupport socketFactorySupport;
private volatile ApplicationEventPublisher applicationEventPublisher;
@Override
public Class<?> getObjectType() {
return this.connectionFactory != null ? this.connectionFactory.getClass()
@@ -168,6 +173,7 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
factory.setTaskExecutor(this.taskExecutor);
factory.setBeanName(this.beanName);
factory.setTcpSocketSupport(this.socketSupport);
factory.setApplicationEventPublisher(this.applicationEventPublisher);
}
private void setServerAttributes(AbstractServerConnectionFactory factory) {
@@ -360,17 +366,6 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
this.singleUse = singleUse;
}
/**
* @param poolSize
* @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#setPoolSize(int)
* @deprecated
*/
@Deprecated
public void setPoolSize(int poolSize) {
logger.warn("poolSize is deprecated; use backlog instead");
this.backlog = poolSize;
}
/**
* @param backlog
* @see AbstractServerConnectionFactory#setBacklog(int)
@@ -477,5 +472,9 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
this.socketFactorySupport = tcpSocketFactorySupport;
}
public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
this.applicationEventPublisher = applicationEventPublisher;
}
}

6
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionParser.java → spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryParser.java
View File

@@ -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.
@@ -29,7 +29,7 @@ import org.w3c.dom.Element;
* @since 2.0
*
*/
public class TcpConnectionParser extends AbstractBeanDefinitionParser {
public class TcpConnectionFactoryParser extends AbstractBeanDefinitionParser {
@Override
protected AbstractBeanDefinition parseInternal(Element element,
@@ -88,6 +88,8 @@ public class TcpConnectionParser extends AbstractBeanDefinitionParser {
IpAdapterParserUtils.SOCKET_FACTORY_SUPPORT);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.SOCKET_SUPPORT);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.MAPPER);
return builder.getBeanDefinition();
}

6
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundGatewayParser.java
View File

@@ -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.
@@ -26,7 +26,7 @@ import org.w3c.dom.Element;
* @since 2.0
*/
public class TcpInboundGatewayParser extends AbstractInboundGatewayParser {
private static final String BASE_PACKAGE = "org.springframework.integration.ip.tcp";
@Override
@@ -46,7 +46,7 @@ public class TcpInboundGatewayParser extends AbstractInboundGatewayParser {
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.TCP_CONNECTION_FACTORY);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.SCHEDULER);
IpAdapterParserUtils.SCHEDULER, "taskScheduler");
}
}

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

@@ -32,7 +32,6 @@ import org.springframework.integration.ip.tcp.connection.ClientModeConnectionMan
import org.springframework.integration.ip.tcp.connection.TcpConnection;
import org.springframework.integration.ip.tcp.connection.TcpListener;
import org.springframework.integration.ip.tcp.connection.TcpSender;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
/**
@@ -218,15 +217,6 @@ public class TcpInboundGateway extends MessagingGatewaySupport implements
this.isClientMode = isClientMode;
}
/**
* @param scheduler the scheduler to set
* @deprecated Use {@link TcpInboundGateway#setTaskScheduler(TaskScheduler)}
*/
@Deprecated
public void setScheduler(TaskScheduler scheduler) {
this.setTaskScheduler(scheduler);
}
/**
* @return the retryInterval
*/

10
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java
View File

@@ -28,7 +28,6 @@ import org.springframework.integration.ip.tcp.connection.ClientModeCapable;
import org.springframework.integration.ip.tcp.connection.ClientModeConnectionManager;
import org.springframework.integration.ip.tcp.connection.ConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpListener;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
/**
@@ -191,15 +190,6 @@ public class TcpReceivingChannelAdapter
this.isClientMode = isClientMode;
}
/**
* @param scheduler the scheduler to set
* @deprecated Use {@link #setTaskScheduler(TaskScheduler)}
*/
@Deprecated
public void setScheduler(TaskScheduler scheduler) {
this.setTaskScheduler(scheduler);
}
/**
* @return the retryInterval
*/

28
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java
View File

@@ -69,25 +69,6 @@ public class TcpSendingMessageHandler extends AbstractMessageHandler implements
private volatile boolean active;
/**
* @deprecated Use {@link #obtainConnection(Message)}.
* TODO: remove in 3.0
*/
@Deprecated
protected TcpConnection getConnection() {
TcpConnection connection = null;
if (this.clientConnectionFactory == null) {
return null;
}
try {
connection = this.clientConnectionFactory.getConnection();
}
catch (Exception e) {
logger.error("Error creating connection", e);
}
return connection;
}
protected TcpConnection obtainConnection(Message<?> 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);

101
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java
View File

@@ -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<TcpConnection> connections = new LinkedList<TcpConnection>();
private volatile List<TcpConnectionSupport> connections = new LinkedList<TcpConnectionSupport>();
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<TcpConnection> iterator = this.connections.iterator();
Iterator<TcpConnectionSupport> 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<String> removeClosedConnectionsAndReturnOpenConnectionIds() {
synchronized (this.connections) {
Iterator<TcpConnection> iterator = this.connections.iterator();
List<String> openConnectionIds = new ArrayList<String>();
Iterator<TcpConnectionSupport> 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<String> 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;
}
}
}

11
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java
View File

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

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java
View File

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

10
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpNioConnectionSupport.java
View File

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

14
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpNioSSLConnectionSupport.java
View File

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

8
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java
View File

@@ -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<AbstractClientConnectionFactory> factories;
public FailoverClientConnectionFactory(List<AbstractClientConnectionFactory> 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);

8
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java
View File

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

91
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionEvent.java Normal file
View File

@@ -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 {
}
}

73
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionEventListeningMessageProducer.java Normal file
View File

@@ -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<TcpConnectionEvent> {
private volatile Set<Class<? extends TcpConnectionEvent>> eventTypes =
new HashSet<Class<? extends TcpConnectionEvent>>();
/**
* 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<? extends TcpConnectionEvent>[] eventTypes) {
Assert.notEmpty(eventTypes, "at least one event type is required");
Set<Class<? extends TcpConnectionEvent>> eventTypeSet = new HashSet<Class<? extends TcpConnectionEvent>>();
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<? extends TcpConnectionEvent> eventType : this.eventTypes) {
if (eventType.isAssignableFrom(event.getClass())) {
this.sendMessage(messageFromEvent(event));
break;
}
}
}
}
protected Message<TcpConnectionEvent> messageFromEvent(TcpConnectionEvent event) {
return MessageBuilder.withPayload(event).build();
}
}

4
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java
View File

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

84
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java
View File

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

45
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java
View File

@@ -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<Object> message = null;
Object payload = connection.getPayload();
if (payload != null) {
MessageBuilder<Object> 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<String, ?> 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 <String, ?> headers to be added to the message.
*/
protected Map<String, ?> supplyCustomHeaders(TcpConnection connection) {
return null;
}
public Object fromMessage(Message<?> message) throws Exception {

3
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java
View File

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

47
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java
View File

@@ -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<Object>) this.getSerializer()).serialize(object, this.socket.getOutputStream());
try {
((Serializer<Object>) 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;
}

3
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java
View File

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

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java
View File

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

37
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java
View File

@@ -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<Object>) this.getSerializer()).serialize(object, this.getChannelOutputStream());
try {
((Serializer<Object>) 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;
}
}

22
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionSupport.java
View File

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

16
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioSSLConnection.java
View File

@@ -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

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java
View File

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

28
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/AbstractByteArraySerializer.java
View File

@@ -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<byte[]>,
Serializer<byte[]>,
Deserializer<byte[]> {
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;
}
}

13
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java
View File

@@ -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
}
/**

76
spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-3.0.xsd
View File

@@ -517,16 +517,6 @@ cached thread pool task executor is used.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="pool-size" type="xsd:string" >
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="backlog" type="xsd:string" >
<xsd:annotation>
<xsd:documentation>
@@ -606,6 +596,72 @@ used to create SSLServerSocketFactory and SSLSocketFactory instances.
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="mapper" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
A reference to a TcpMessageMapper implementation. Allows customization of headers in inbound messages by overriding
setCustomHeaders(). Default is TcpMessageMapper.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.ip.tcp.connection.TcpMessageMapper"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<xsd:element name="tcp-connection-event-inbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Configures an inbound Channel Adapter which listens for TCP Connection
events, converts them to Messages and sends them to a Message Channel.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:attribute name="id" type="xsd:string" use="optional" />
<xsd:attribute name="event-types" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation>
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).
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.integration.MessageChannel" />
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
The channel to which Messages generated from Application Context events will be sent.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="error-channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.integration.MessageChannel" />
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="auto-startup" type="xsd:string" default="true" />
</xsd:complexType>
</xsd:element>

13
spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml
View File

@@ -59,8 +59,11 @@
lookup-host="false"
apply-sequence="true"
ssl-context-support="sslContextSupport"
mapper="mapper"
/>
<bean id="mapper" class="org.springframework.integration.ip.tcp.connection.TcpMessageMapper" />
<ip:tcp-connection-factory id="cfS1Nio"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(5210)}"
@@ -281,13 +284,6 @@
interceptor-factory-chain="interceptors"
/>
<ip:tcp-connection-factory
id="serverBackwardsCompatible"
type="server"
port="#{client1.port}"
pool-size="123"
/>
<ip:tcp-connection-factory
id="client2"
type="client"
@@ -423,4 +419,7 @@
</constructor-arg>
</bean>
<ip:tcp-connection-event-inbound-channel-adapter id="eventAdapter" channel="nullChannel"
event-types="org.springframework.integration.ip.config.ParserUnitTests$EventSubclass1, org.springframework.integration.ip.config.ParserUnitTests$EventSubclass2"/>
</beans>

49
spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java
View File

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

10
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests-context.xml
View File

@@ -33,15 +33,21 @@
/>
<int-ip:tcp-inbound-gateway id="looper"
request-channel="queue"
request-channel="serverSideChannel"
connection-factory="server"
reply-timeout="1"
/>
<int:channel id="queue">
<int:channel id="serverSideChannel">
<int:queue/>
</int:channel>
<task:executor id="exec" pool-size="10"/>
<int-ip:tcp-connection-event-inbound-channel-adapter channel="events" />
<int:channel id="events">
<int:queue />
</int:channel>
</beans>

39
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java
View File

@@ -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<TcpConnectionEvent> eventMessage;
while ((eventMessage = (Message<TcpConnectionEvent>) 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

72
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java Normal file
View File

@@ -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<ApplicationEvent> theEvent = new AtomicReference<ApplicationEvent>();
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<Object> 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<String>("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());
}
}

106
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionFactoryTests.java Normal file
View File

@@ -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<TcpConnectionEvent> events =
Collections.synchronizedList(new ArrayList<TcpConnectionEvent>());
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<String> clients = clientFactory.getOpenConnectionIds();
assertEquals(1, clients.size());
assertTrue(clients.contains(client.getConnectionId()));
List<String> 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);
}
}
}

104
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpConnectionEventListenerTests.java Normal file
View File

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

31
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java
View File

@@ -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<Object> 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<String, ?> 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<Object> 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 {
}
}

4
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java
View File

@@ -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) {

40
src/reference/docbook/whats-new.xml
View File

@@ -12,6 +12,46 @@
<section id="3.0-new-components">
<title>New Components</title>
<section id="3.0-tcp-events">
<title>TCP/IP Connection Events and Connection Management</title>
<para>
The (supplied) <classname>TcpConnection</classname>s now emit
<interfacename>ApplicationEvent</interfacename>s (specifically
<classname>TcpConnectionEvent</classname>s) when connections are
opened, closed, or an exception occurs. This allows applications
to be informed of changes to TCP connections using the normal
Spring <interfacename>ApplicationListener</interfacename>
mechanism.
</para>
<para>
<classname>AbstractTcpConnection</classname> has been renamed
<classname>TcpConnectionSupport</classname>; custom connections that
are subclasses of this class, can use its methods to publish events.
Similarly, <classname>AbstractTcpConnectionInterceptor</classname> has
been renamed to <classname>TcpConnectionInterceptorSupport</classname>.
</para>
<para>
In addition, a new <code>&lt;int-ip:tcp-connection-event-channel-adapter/&gt;</code>
is provided; by default, this adapter sends all <classname>TcpConnectionEvents</classname>
to a <interfacename>Channel</interfacename>.
</para>
<para>
Further, the TCP Connection Factories, now provide a new method
<code>getOpenConnectionIds()</code>, which returns a list of identifiers for all
open connections; this allows applications, for example, to broadcast to all
open connections.
</para>
<para>
Finally, the connection factories also provide a new method
<code>closeConnection(String connectionId)</code> which allows applications
to explicitly close a connection using its ID.
</para>
<para>
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.
</para>
</section>
</section>
<section id="3.0-general">