diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java index 660ef818c4..5fc8c35a69 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.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. You may obtain a copy of the License at @@ -31,8 +31,8 @@ import org.springframework.integration.core.MessageProducer; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.integration.store.MessageGroup; -import org.springframework.integration.store.MessageGroupCallback; import org.springframework.integration.store.MessageGroupStore; +import org.springframework.integration.store.MessageGroupStore.MessageGroupCallback; import org.springframework.integration.store.MessageStore; import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.integration.store.SimpleMessageStore; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupCallback.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupCallback.java deleted file mode 100755 index 09736af29e..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupCallback.java +++ /dev/null @@ -1,30 +0,0 @@ -/* - * Copyright 2002-2012 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.store; - -/** - * Invoked when a MessageGroupStore expires a group. - *

- * Note: This interface will become an inner interface of - * MessageGroupStore in release 3.0. - * @author Dave Syer - * - * @since 2.0 - * - */ -public interface MessageGroupCallback { - - void execute(MessageGroupStore messageGroupStore, MessageGroup group); - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java index 2a04cf437b..41026f3914 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/MessageGroupStore.java @@ -1,11 +1,11 @@ /* - * Copyright 2002-2011 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. 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. @@ -19,19 +19,20 @@ import org.springframework.jmx.export.annotation.ManagedAttribute; /** * Interface for storage operations on groups of messages linked by a group id. - * + * * @author Dave Syer * @author Oleg Zhurakousky - * + * @author Gary Russell + * * @since 2.0 - * + * */ public interface MessageGroupStore { /** * Optional attribute giving the number of messages in the store over all groups. Implementations may decline to * respond by throwing an exception. - * + * * @return the number of messages * @throws UnsupportedOperationException if not implemented */ @@ -40,13 +41,13 @@ public interface MessageGroupStore { /** * Optional attribute giving the number of message groups. Implementations may decline * to respond by throwing an exception. - * + * * @return the number message groups * @throws UnsupportedOperationException if not implemented */ @ManagedAttribute int getMessageGroupCount(); - + /** * Returns the size of this MessageGroup * @param groupId @@ -57,19 +58,19 @@ public interface MessageGroupStore { /** * Return all Messages currently in the MessageStore that were stored using * {@link #addMessageToGroup(Object, Message)} with this group id. - * + * * @return a group of messages, empty if none exists for this key */ MessageGroup getMessageGroup(Object groupId); /** * Store a message with an association to a group id. This can be used to group messages together. - * + * * @param groupId the group id to store the message under * @param message a message */ MessageGroup addMessageToGroup(Object groupId, Message message); - + /** * Persist a deletion on a single message from the group. The group is modified to reflect that 'messageToRemove' is * no longer present in the group. @@ -80,14 +81,14 @@ public interface MessageGroupStore { /** * Remove the message group with this id. - * + * * @param groupId the id of the group to remove */ void removeMessageGroup(Object groupId); /** * Register a callback for when a message group is expired through {@link #expireMessageGroups(long)}. - * + * * @param callback a callback to execute when a message group is cleaned up */ void registerMessageGroupExpiryCallback(MessageGroupCallback callback); @@ -97,36 +98,46 @@ public interface MessageGroupStore { * each of the registered callbacks on them in turn. For example: call with a timeout of 100 to expire all groups * that were created more than 100 milliseconds ago, and are not yet complete. Use a timeout of 0 (or negative to be * on the safe side) to expire all message groups. - * + * * @param timeout the timeout threshold to use * @return the number of message groups expired - * + * * @see #registerMessageGroupExpiryCallback(MessageGroupCallback) */ int expireMessageGroups(long timeout); - + /** * Allows you to set the sequence number of the last released Message. Used for Resequencing use cases * @param sequenceNumber */ void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber); - + /** * Returns the iterator of currently accumulated {@link MessageGroup}s */ Iterator iterator(); - - + + /** * Polls Message from this {@link MessageGroup} (in FIFO style if supported by the implementation) * while also removing the polled {@link Message} */ Message pollMessageFromGroup(Object groupId); - + /** - * Completes this MessageGroup. Completion of the MessageGroup generally means - * that this group should not be allowing any more mutating operation to be performed on it. + * Completes this MessageGroup. Completion of the MessageGroup generally means + * that this group should not be allowing any more mutating operation to be performed on it. * For example any attempt to add/remove new Message form the group should not be allowed. */ void completeGroup(Object groupId); + + /** + * Invoked when a MessageGroupStore expires a group. + */ + public interface MessageGroupCallback { + + void execute(MessageGroupStore messageGroupStore, MessageGroup group); + + } + } \ No newline at end of file diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreReaperTests.java b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreReaperTests.java index 115c037961..132fb4b4f8 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreReaperTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreReaperTests.java @@ -24,6 +24,7 @@ import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.store.MessageGroupStore.MessageGroupCallback; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -31,6 +32,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; * @author Dave Syer * @author Dave Turanski * @author Artem Bilan + * @author Gary Russell */ @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java index 1f29a4e280..e1b162eb43 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/store/MessageStoreTests.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,8 @@ package org.springframework.integration.store; +import static org.junit.Assert.assertEquals; + import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -23,22 +25,21 @@ import java.util.Iterator; import java.util.List; import org.junit.Test; - import org.springframework.integration.Message; import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.store.MessageGroupStore.MessageGroupCallback; import org.springframework.test.util.ReflectionTestUtils; -import static org.junit.Assert.assertEquals; - /** * @author Dave Syer + * @author Gary Russell */ public class MessageStoreTests { @Test public void shouldRegisterCallbacks() throws Exception { TestMessageStore store = new TestMessageStore(); - store.setExpiryCallbacks(Arrays. asList(new MessageGroupCallback() { + store.setExpiryCallbacks(Arrays. asList(new MessageGroupStore.MessageGroupCallback() { public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { } })); @@ -81,8 +82,8 @@ public class MessageStoreTests { MessageGroup testMessages = new SimpleMessageGroup(Arrays.asList(new GenericMessage("foo")), "bar"); private boolean removed = false; - - + + public Iterator iterator() { return Arrays.asList(testMessages).iterator(); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java b/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java index 649367a482..3190be47c9 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageStoreTests.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. @@ -29,12 +29,14 @@ import java.util.List; import org.junit.Test; import org.springframework.integration.Message; import org.springframework.integration.MessagingException; +import org.springframework.integration.store.MessageGroupStore.MessageGroupCallback; import org.springframework.integration.support.MessageBuilder; import org.springframework.test.util.ReflectionTestUtils; /** * @author Iwein Fuld * @author Dave Syer + * @author Gary Russell */ public class SimpleMessageStoreTests { @@ -116,7 +118,7 @@ public class SimpleMessageStoreTests { @Test public void shouldRegisterCallbacks() throws Exception { SimpleMessageStore store = new SimpleMessageStore(); - store.setExpiryCallbacks(Arrays. asList(new MessageGroupCallback() { + store.setExpiryCallbacks(Arrays. asList(new MessageGroupStore.MessageGroupCallback() { public void execute(MessageGroupStore messageGroupStore, MessageGroup group) { } })); 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 1eb427c24e..213b469c89 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 @@ -25,8 +25,11 @@ import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; +import org.springframework.integration.ip.tcp.connection.DefaultTcpNetSSLSocketFactorySupport; +import org.springframework.integration.ip.tcp.connection.DefaultTcpNetSocketFactorySupport; import org.springframework.integration.ip.tcp.connection.DefaultTcpNioConnectionSupport; import org.springframework.integration.ip.tcp.connection.DefaultTcpNioSSLConnectionSupport; +import org.springframework.integration.ip.tcp.connection.DefaultTcpSocketSupport; import org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactoryChain; import org.springframework.integration.ip.tcp.connection.TcpMessageMapper; import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory; @@ -34,12 +37,9 @@ import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionF import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpNioConnectionSupport; import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory; -import org.springframework.integration.ip.tcp.connection.support.DefaultTcpNetSSLSocketFactorySupport; -import org.springframework.integration.ip.tcp.connection.support.DefaultTcpNetSocketFactorySupport; -import org.springframework.integration.ip.tcp.connection.support.DefaultTcpSocketSupport; -import org.springframework.integration.ip.tcp.connection.support.TcpSSLContextSupport; -import org.springframework.integration.ip.tcp.connection.support.TcpSocketFactorySupport; -import org.springframework.integration.ip.tcp.connection.support.TcpSocketSupport; +import org.springframework.integration.ip.tcp.connection.TcpSSLContextSupport; +import org.springframework.integration.ip.tcp.connection.TcpSocketFactorySupport; +import org.springframework.integration.ip.tcp.connection.TcpSocketSupport; import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer; import org.springframework.util.Assert; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java index 244d66bf0c..e3f2c91868 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 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. @@ -27,7 +27,7 @@ import java.net.Socket; */ public abstract class AbstractClientConnectionFactory extends AbstractConnectionFactory { - private TcpConnection theConnection; + private TcpConnectionSupport theConnection; /** * Constructs a factory that will established connections to the host and port. @@ -43,20 +43,20 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection * true, a new connection is returned; otherwise a single connection is * reused for all requests while the connection remains open. */ - public TcpConnection getConnection() throws Exception { + public TcpConnectionSupport getConnection() throws Exception { this.checkActive(); if (this.isSingleUse()) { return obtainConnection(); } else { synchronized(this) { - TcpConnection connection = obtainConnection(); + TcpConnectionSupport connection = obtainConnection(); this.setTheConnection(connection); return connection; } } } - protected abstract TcpConnection obtainConnection() throws Exception; + protected abstract TcpConnectionSupport obtainConnection() throws Exception; /** * Transfers attributes such as (de)serializers, singleUse etc to a new connection. @@ -67,7 +67,7 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection * @param connection The new connection. * @param socket The new socket. */ - protected void initializeConnection(TcpConnection connection, Socket socket) { + protected void initializeConnection(TcpConnectionSupport connection, Socket socket) { TcpListener listener = this.getListener(); if (listener != null) { connection.registerListener(listener); @@ -85,14 +85,14 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection /** * @param theConnection the theConnection to set */ - protected void setTheConnection(TcpConnection theConnection) { + protected void setTheConnection(TcpConnectionSupport theConnection) { this.theConnection = theConnection; } /** * @return the theConnection */ - protected TcpConnection getTheConnection() { + protected TcpConnectionSupport getTheConnection() { return theConnection; } 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 62fa264585..d48434dace 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 @@ -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. @@ -40,8 +40,6 @@ import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; import org.springframework.integration.MessagingException; import org.springframework.integration.context.IntegrationObjectSupport; -import org.springframework.integration.ip.tcp.connection.support.DefaultTcpSocketSupport; -import org.springframework.integration.ip.tcp.connection.support.TcpSocketSupport; import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer; import org.springframework.util.Assert; @@ -482,7 +480,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport } } - protected TcpConnection wrapConnection(TcpConnection connection) throws Exception { + protected TcpConnectionSupport wrapConnection(TcpConnectionSupport connection) throws Exception { try { if (this.interceptorFactoryChain == null) { return connection; @@ -493,7 +491,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport return connection; } for (TcpConnectionInterceptorFactory factory : interceptorFactories) { - TcpConnectionInterceptor wrapper = factory.getInterceptor(); + TcpConnectionInterceptorSupport wrapper = factory.getInterceptor(); wrapper.setTheConnection(connection); // if no ultimate listener or sender, register each wrapper in turn if (this.listener == null) { 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 ce34531191..2fbb807acf 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 @@ -1,5 +1,5 @@ /* - * Copyright 2001-2011 the original author or authors. + * Copyright 2001-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. @@ -99,7 +99,7 @@ public abstract class AbstractServerConnectionFactory * @param connection The new connection. * @param socket The new socket. */ - protected void initializeConnection(TcpConnection connection, Socket socket) { + protected void initializeConnection(TcpConnectionSupport connection, Socket socket) { TcpListener listener = this.getListener(); if (listener != null) { connection.registerListener(listener); 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 0369a12903..6996ed00a2 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 @@ -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. @@ -38,7 +38,7 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact private final AbstractClientConnectionFactory targetConnectionFactory; - private final SimplePool pool; + private final SimplePool pool; private volatile TcpListener listener; @@ -47,9 +47,9 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact // override single-use to true to force "close" after use target.setSingleUse(true); this.targetConnectionFactory = target; - pool = new SimplePool(poolSize, new SimplePool.PoolItemCallback() { + pool = new SimplePool(poolSize, new SimplePool.PoolItemCallback() { - public TcpConnection createForPool() { + public TcpConnectionSupport createForPool() { try { return targetConnectionFactory.getConnection(); } catch (Exception e) { @@ -57,11 +57,11 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact } } - public boolean isStale(TcpConnection connection) { + public boolean isStale(TcpConnectionSupport connection) { return !connection.isOpen(); } - public void removedFromPool(TcpConnection connection) { + public void removedFromPool(TcpConnectionSupport connection) { connection.close(); } }); @@ -72,13 +72,11 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact } @Override - @SuppressWarnings("deprecation") public synchronized void setPoolSize(int poolSize) { this.pool.setPoolSize(poolSize); } @Override - @SuppressWarnings("deprecation") public int getPoolSize() { return this.pool.getPoolSize(); } @@ -96,19 +94,17 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact } @Override - public TcpConnection obtainConnection() throws Exception { + public TcpConnectionSupport obtainConnection() throws Exception { return new CachedConnection(this.pool.getItem()); } - private class CachedConnection extends AbstractTcpConnectionInterceptor { + private class CachedConnection extends TcpConnectionInterceptorSupport { private volatile boolean released; - public CachedConnection(TcpConnection connection) { + public CachedConnection(TcpConnectionSupport connection) { super.setTheConnection(connection); - if (connection instanceof AbstractTcpConnection) { - ((AbstractTcpConnection) connection).registerListener(this); - } + connection.registerListener(this); } @Override diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpNetSSLSocketFactorySupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpNetSSLSocketFactorySupport.java similarity index 96% rename from spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpNetSSLSocketFactorySupport.java rename to spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpNetSSLSocketFactorySupport.java index 7b7b8e678f..95aab3d15c 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpNetSSLSocketFactorySupport.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpNetSSLSocketFactorySupport.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.springframework.integration.ip.tcp.connection.support; +package org.springframework.integration.ip.tcp.connection; import javax.net.ServerSocketFactory; import javax.net.SocketFactory; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpNetSocketFactorySupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpNetSocketFactorySupport.java similarity index 94% rename from spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpNetSocketFactorySupport.java rename to spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpNetSocketFactorySupport.java index 0131fb4da9..172aa3a1c7 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpNetSocketFactorySupport.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpNetSocketFactorySupport.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.springframework.integration.ip.tcp.connection.support; +package org.springframework.integration.ip.tcp.connection; import javax.net.ServerSocketFactory; import javax.net.SocketFactory; 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 4fc0992233..4ffdfcb60d 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 @@ -21,7 +21,6 @@ import javax.net.ssl.SSLContext; import javax.net.ssl.SSLEngine; import org.springframework.beans.factory.InitializingBean; -import org.springframework.integration.ip.tcp.connection.support.TcpSSLContextSupport; import org.springframework.util.Assert; /** diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpSSLContextSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpSSLContextSupport.java similarity index 98% rename from spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpSSLContextSupport.java rename to spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpSSLContextSupport.java index b7c54d4264..3e6ab6eb19 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpSSLContextSupport.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpSSLContextSupport.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.springframework.integration.ip.tcp.connection.support; +package org.springframework.integration.ip.tcp.connection; import java.io.IOException; import java.security.GeneralSecurityException; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpSocketSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpSocketSupport.java similarity index 93% rename from spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpSocketSupport.java rename to spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpSocketSupport.java index 4a031c1619..d400b7f3e8 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpSocketSupport.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/DefaultTcpSocketSupport.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.springframework.integration.ip.tcp.connection.support; +package org.springframework.integration.ip.tcp.connection; import java.net.ServerSocket; import java.net.Socket; 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 2679e73563..b5c326b8ed 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 @@ -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. @@ -99,8 +99,8 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac } @Override - protected TcpConnection obtainConnection() throws Exception { - TcpConnection connection = this.getTheConnection(); + protected TcpConnectionSupport obtainConnection() throws Exception { + TcpConnectionSupport connection = this.getTheConnection(); if (connection != null && connection.isOpen()) { return connection; } @@ -150,7 +150,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac * @since 2.2 * */ - private class FailoverTcpConnection implements TcpConnection, TcpListener { + private class FailoverTcpConnection extends TcpConnectionSupport implements TcpListener { private final List factories; @@ -160,7 +160,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac private volatile AbstractClientConnectionFactory currentFactory; - private volatile TcpConnection delegate; + private volatile TcpConnectionSupport delegate; private volatile boolean open = true; @@ -212,6 +212,7 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac } } + @Override public void close() { this.delegate.close(); this.open = false; @@ -264,10 +265,12 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac throw new UnsupportedOperationException("Not supported on FailoverTcpConnection"); } + @Override public String getHostName() { return this.delegate.getHostName(); } + @Override public String getHostAddress() { return this.delegate.getHostAddress(); } @@ -276,54 +279,67 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac return this.delegate.getPort(); } + @Override public void registerListener(TcpListener listener) { this.delegate.registerListener(listener); } + @Override public void registerSender(TcpSender sender) { this.delegate.registerSender(sender); } + @Override public String getConnectionId() { return this.connectionId; } + @Override public void setSingleUse(boolean singleUse) { this.delegate.setSingleUse(singleUse); } + @Override public boolean isSingleUse() { return this.delegate.isSingleUse(); } + @Override public boolean isServer() { return this.delegate.isServer(); } + @Override public void setMapper(TcpMessageMapper mapper) { this.delegate.setMapper(mapper); } + @Override public Deserializer getDeserializer() { return this.delegate.getDeserializer(); } + @Override public void setDeserializer(Deserializer deserializer) { this.delegate.setDeserializer(deserializer); } + @Override public Serializer getSerializer() { return this.delegate.getSerializer(); } + @Override public void setSerializer(Serializer serializer) { this.delegate.setSerializer(serializer); } + @Override public TcpListener getListener() { return this.delegate.getListener(); } + @Override public long incrementAndGetConnectionSequence() { return this.delegate.incrementAndGetConnectionSequence(); } 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 ebc9f79a67..427104851f 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 @@ -1,5 +1,5 @@ /* - * Copyright 2001-2011 the original author or authors. + * Copyright 2001-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. @@ -75,47 +75,22 @@ public interface TcpConnection extends Runnable { */ int getPort(); - /** - * Sets the listener that will receive incoming Messages. - * @param listener The listener - */ - void registerListener(TcpListener listener); - - /** - * Registers a sender. Used on server side sockets so a - * sender can determine which connection to send a reply - * to. - * @param sender the sender - */ - void registerSender(TcpSender sender); - /** * @return a string uniquely representing a connection. */ String getConnectionId(); - - /** - * When true, the socket is used once and discarded. - * @param singleUse the singleUse - */ - void setSingleUse(boolean singleUse); /** * * @return True if connection is used once. */ boolean isSingleUse(); - + /** * * @return True if connection is used once. */ boolean isServer(); - - /** - * @param mapper the mapper - */ - void setMapper(TcpMessageMapper mapper); /** * @@ -123,22 +98,12 @@ public interface TcpConnection extends Runnable { */ Deserializer getDeserializer(); - /** - * @param deserializer the deserializer to set - */ - void setDeserializer(Deserializer deserializer); - /** * * @return the serializer */ Serializer getSerializer(); - - /** - * @param serializer the serializer to set - */ - void setSerializer(Serializer serializer); - + /** * @return this connection's listener */ @@ -148,5 +113,5 @@ public interface TcpConnection extends Runnable { * @return the next sequence number for a message received on this socket */ long incrementAndGetConnectionSequence(); - + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptor.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptor.java index 4db2d00083..1833397ef0 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptor.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptor.java @@ -1,5 +1,5 @@ /* - * Copyright 2001-2011 the original author or authors. + * Copyright 2001-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,7 +21,5 @@ package org.springframework.integration.ip.tcp.connection; * */ public interface TcpConnectionInterceptor extends TcpConnection, TcpListener, TcpSender { - - void setTheConnection(TcpConnection connection); } \ No newline at end of file diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorFactory.java index 84edf4167b..766866aae0 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2001-2011 the original author or authors. + * Copyright 2001-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,8 +17,8 @@ package org.springframework.integration.ip.tcp.connection; /** - * Interface for TCP connection interceptor factories. - * + * Interface for TCP connection interceptor factories. + * * @author Gary Russell * @since 2.0 * @@ -28,9 +28,9 @@ public interface TcpConnectionInterceptorFactory { /** * Called for each new connection; * a new interceptor must be returned on each call. - * - * @return the TcpInterceptor + * + * @return the TcpInterceptor */ - abstract TcpConnectionInterceptor getInterceptor(); + abstract TcpConnectionInterceptorSupport getInterceptor(); } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java similarity index 85% rename from spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java rename to spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java index fa22449331..a0e4456473 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.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. @@ -23,20 +23,21 @@ import org.springframework.integration.Message; /** * Base class for TcpConnectionIntercepters; passes all method calls through * to the underlying {@link TcpConnection}. - * + * * @author Gary Russell * @since 2.0 */ -public abstract class AbstractTcpConnectionInterceptor implements TcpConnectionInterceptor { +public abstract class TcpConnectionInterceptorSupport extends TcpConnectionSupport implements TcpConnectionInterceptor { + + private TcpConnectionSupport theConnection; - private TcpConnection theConnection; - private TcpListener tcpListener; private TcpSender tcpSender; - + private Boolean realSender; + @Override public void close() { this.theConnection.close(); } @@ -49,10 +50,12 @@ public abstract class AbstractTcpConnectionInterceptor implements TcpConnectionI return this.theConnection.getPayload(); } + @Override public String getHostName() { return this.theConnection.getHostName(); } + @Override public String getHostAddress() { return this.theConnection.getHostAddress(); } @@ -61,20 +64,24 @@ public abstract class AbstractTcpConnectionInterceptor implements TcpConnectionI return this.theConnection.getPort(); } + @Override public void registerListener(TcpListener listener) { this.tcpListener = listener; this.theConnection.registerListener(this); } + @Override public void registerSender(TcpSender sender) { this.tcpSender = sender; this.theConnection.registerSender(this); } + @Override public String getConnectionId() { return this.theConnection.getConnectionId(); } + @Override public boolean isSingleUse() { return this.theConnection.isSingleUse(); } @@ -83,30 +90,37 @@ public abstract class AbstractTcpConnectionInterceptor implements TcpConnectionI this.theConnection.run(); } + @Override public void setSingleUse(boolean singleUse) { this.theConnection.setSingleUse(singleUse); } + @Override public void setMapper(TcpMessageMapper mapper) { this.theConnection.setMapper(mapper); } + @Override public Deserializer getDeserializer() { return this.theConnection.getDeserializer(); } + @Override public void setDeserializer(Deserializer deserializer) { this.theConnection.setDeserializer(deserializer); } + @Override public Serializer getSerializer() { return this.theConnection.getSerializer(); } + @Override public void setSerializer(Serializer serializer) { this.theConnection.setSerializer(serializer); } + @Override public boolean isServer() { return this.theConnection.isServer(); } @@ -126,7 +140,7 @@ public abstract class AbstractTcpConnectionInterceptor implements TcpConnectionI * Returns the underlying connection (or next interceptor) * @return the connection */ - public TcpConnection getTheConnection() { + public TcpConnectionSupport getTheConnection() { return this.theConnection; } @@ -134,13 +148,14 @@ public abstract class AbstractTcpConnectionInterceptor implements TcpConnectionI * Sets the underlying connection (or next interceptor) * @param theConnection the connection */ - public void setTheConnection(TcpConnection theConnection) { + public void setTheConnection(TcpConnectionSupport theConnection) { this.theConnection = theConnection; } /** * @return the listener */ + @Override public TcpListener getListener() { return tcpListener; } @@ -157,21 +172,23 @@ public abstract class AbstractTcpConnectionInterceptor implements TcpConnectionI } } + @Override public long incrementAndGetConnectionSequence() { return this.theConnection.incrementAndGetConnectionSequence(); } - - TcpSender getSender() { + + @Override + public TcpSender getSender() { return this.tcpSender; } - + protected boolean hasRealSender() { if (this.realSender != null) { return this.realSender; } TcpSender sender = this.getSender(); - while (sender instanceof AbstractTcpConnectionInterceptor) { - sender = ((AbstractTcpConnectionInterceptor) sender).getSender(); + while (sender instanceof TcpConnectionInterceptorSupport) { + sender = ((TcpConnectionInterceptorSupport) sender).getSender(); } this.realSender = sender != null; return this.realSender; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java similarity index 92% rename from spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java rename to spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java index 44e5e58375..7022c07432 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java @@ -1,5 +1,5 @@ /* - * Copyright 2001-2011 the original author or authors. + * Copyright 2001-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. @@ -39,7 +39,7 @@ import org.springframework.util.Assert; * @since 2.0 * */ -public abstract class AbstractTcpConnection implements TcpConnection { +public abstract class TcpConnectionSupport implements TcpConnection { protected final Log logger = LogFactory.getLog(this.getClass()); @@ -71,7 +71,11 @@ public abstract class AbstractTcpConnection implements TcpConnection { private volatile String hostAddress = "unknown"; - public AbstractTcpConnection(Socket socket, boolean server, boolean lookupHost) { + public TcpConnectionSupport() { + server = false; + } + + public TcpConnectionSupport(Socket socket, boolean server, boolean lookupHost) { this.server = server; InetAddress inetAddress = socket.getInetAddress(); if (inetAddress != null) { @@ -182,7 +186,8 @@ public abstract class AbstractTcpConnection implements TcpConnection { } /** - * @param listener the listener to set + * Sets the listener that will receive incoming Messages. + * @param listener The listener. */ public void registerListener(TcpListener listener) { this.listener = listener; @@ -199,7 +204,10 @@ public abstract class AbstractTcpConnection implements TcpConnection { } /** - * @param sender the sender to set + * Registers a sender. Used on server side connections so a + * sender can determine which connection to send a reply + * to. + * @param sender the sender. */ public void registerSender(TcpSender sender) { this.sender = sender; 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 c8ca7cb9fe..7993420fb1 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 @@ -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. @@ -22,8 +22,6 @@ import java.net.SocketException; import javax.net.SocketFactory; -import org.springframework.integration.ip.tcp.connection.support.DefaultTcpNetSocketFactorySupport; -import org.springframework.integration.ip.tcp.connection.support.TcpSocketFactorySupport; import org.springframework.util.Assert; /** @@ -52,8 +50,8 @@ public class TcpNetClientConnectionFactory extends * @throws Exception */ @Override - protected TcpConnection obtainConnection() throws Exception { - TcpConnection theConnection = this.getTheConnection(); + protected TcpConnectionSupport obtainConnection() throws Exception { + TcpConnectionSupport theConnection = this.getTheConnection(); if (theConnection != null && theConnection.isOpen()) { return theConnection; } @@ -62,7 +60,7 @@ public class TcpNetClientConnectionFactory extends } Socket socket = createSocket(this.getHost(), this.getPort()); setSocketAttributes(socket); - TcpConnection connection = new TcpNetConnection(socket, false, this.isLookupHost()); + TcpConnectionSupport connection = new TcpNetConnection(socket, false, this.isLookupHost()); 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 9e64f88a43..a753f292eb 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 @@ -1,5 +1,5 @@ /* - * Copyright 2001-2011 the original author or authors. + * Copyright 2001-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. @@ -32,7 +32,7 @@ import org.springframework.integration.ip.tcp.serializer.SoftEndOfStreamExceptio * @since 2.0 * */ -public class TcpNetConnection extends AbstractTcpConnection { +public class TcpNetConnection extends TcpConnectionSupport { private final Socket socket; 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 60d4d60530..c64af71b6b 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 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. @@ -25,8 +25,6 @@ import java.net.SocketTimeoutException; import javax.net.ServerSocketFactory; -import org.springframework.integration.ip.tcp.connection.support.DefaultTcpNetSocketFactorySupport; -import org.springframework.integration.ip.tcp.connection.support.TcpSocketFactorySupport; import org.springframework.util.Assert; /** @@ -53,7 +51,7 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto /** * If no listener registers, exits. * Accepts incoming connections and creates TcpConnections for each new connection. - * Invokes {{@link #initializeConnection(TcpConnection, Socket)} and executes the + * Invokes {{@link #initializeConnection(TcpConnectionSupport, Socket)} and executes the * connection {@link TcpConnection#run()} using the task executor. * I/O errors on the server socket/channel are logged and the factory is stopped. */ @@ -100,7 +98,7 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto logger.debug("Accepted connection from " + socket.getInetAddress().getHostAddress()); } setSocketAttributes(socket); - TcpConnection connection = new TcpNetConnection(socket, true, this.isLookupHost()); + TcpConnectionSupport connection = new TcpNetConnection(socket, true, this.isLookupHost()); 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 adc3540c2d..8846b4e727 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 @@ -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. @@ -67,7 +67,7 @@ public class TcpNioClientConnectionFactory extends * @throws SocketException */ @Override - protected TcpConnection obtainConnection() throws Exception { + protected TcpConnectionSupport obtainConnection() throws Exception { int n = 0; while (this.selector == null) { try { @@ -79,7 +79,7 @@ public class TcpNioClientConnectionFactory extends throw new Exception("Factory failed to start"); } } - TcpConnection theConnection = this.getTheConnection(); + TcpConnectionSupport theConnection = this.getTheConnection(); if (theConnection != null && theConnection.isOpen()) { return theConnection; } @@ -92,7 +92,7 @@ public class TcpNioClientConnectionFactory extends socketChannel, false, this.isLookupHost()); connection.setUsingDirectBuffers(this.usingDirectBuffers); connection.setTaskExecutor(this.getTaskExecutor()); - TcpConnection wrappedConnection = wrapConnection(connection); + TcpConnectionSupport wrappedConnection = wrapConnection(connection); initializeConnection(wrappedConnection, socketChannel.socket()); socketChannel.configureBlocking(false); if (this.getSoTimeout() > 0) { 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 43c1810f58..374827f52a 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 @@ -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. @@ -46,7 +46,7 @@ import org.springframework.util.Assert; * @since 2.0 * */ -public class TcpNioConnection extends AbstractTcpConnection { +public class TcpNioConnection extends TcpConnectionSupport { private static final long DEFAULT_PIPE_TIMEOUT = 60000; 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 78fa196f07..0e027405e9 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 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. @@ -63,7 +63,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto /** * If no listener registers, exits. * Accepts incoming connections and creates TcpConnections for each new connection. - * Invokes {{@link #initializeConnection(TcpConnection, Socket)} and executes the + * Invokes {{@link #initializeConnection(TcpConnectionSupport, Socket)} and executes the * connection {@link TcpConnection#run()} using the task executor. * I/O errors on the server socket/channel are logged and the factory is stopped. */ @@ -176,7 +176,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto .createNewConnection(socketChannel, true, this.isLookupHost()); connection.setUsingDirectBuffers(this.usingDirectBuffers); - TcpConnection wrappedConnection = wrapConnection(connection); + TcpConnectionSupport wrappedConnection = wrapConnection(connection); this.initializeConnection(wrappedConnection, socketChannel.socket()); return connection; } catch (Exception e) { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/TcpSSLContextSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpSSLContextSupport.java similarity index 94% rename from spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/TcpSSLContextSupport.java rename to spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpSSLContextSupport.java index 793c371dab..680fd3ff86 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/TcpSSLContextSupport.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpSSLContextSupport.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.springframework.integration.ip.tcp.connection.support; +package org.springframework.integration.ip.tcp.connection; import java.io.IOException; import java.security.GeneralSecurityException; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/TcpSocketFactorySupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpSocketFactorySupport.java similarity index 94% rename from spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/TcpSocketFactorySupport.java rename to spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpSocketFactorySupport.java index 619da0ae22..1b6ff9110a 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/TcpSocketFactorySupport.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpSocketFactorySupport.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.springframework.integration.ip.tcp.connection.support; +package org.springframework.integration.ip.tcp.connection; import java.net.ServerSocket; import java.net.Socket; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/TcpSocketSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpSocketSupport.java similarity index 95% rename from spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/TcpSocketSupport.java rename to spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpSocketSupport.java index 26609c0940..9cd26219ed 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/TcpSocketSupport.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpSocketSupport.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.springframework.integration.ip.tcp.connection.support; +package org.springframework.integration.ip.tcp.connection; import java.net.ServerSocket; import java.net.Socket; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/package-info.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/package-info.java deleted file mode 100644 index 82644b750a..0000000000 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/package-info.java +++ /dev/null @@ -1,5 +0,0 @@ -/** - * Provides classes supporting the creation/manipulation of sockets, - * SSLContexts etc. - */ -package org.springframework.integration.ip.tcp.connection.support; \ No newline at end of file diff --git a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.2.xsd b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.2.xsd index c90c3c7ebe..ea9c405cd4 100644 --- a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.2.xsd +++ b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.2.xsd @@ -572,7 +572,7 @@ sockets are used. - + @@ -585,7 +585,7 @@ and ServerSocket instances after creation and after configured attributes are ap - + @@ -601,7 +601,7 @@ used to create SSLServerSocketFactory and SSLSocketFactory instances. - + 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 c2a7e411cf..c3b2f079cc 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 @@ -572,7 +572,7 @@ sockets are used. - + @@ -585,7 +585,7 @@ and ServerSocket instances after creation and after configured attributes are ap - + @@ -601,7 +601,7 @@ used to create SSLServerSocketFactory and SSLSocketFactory instances. - + 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 51e03db442..02bb17fa62 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 @@ -70,7 +70,7 @@ ssl-context-support="sslContextSupport" /> - + @@ -87,11 +87,11 @@ socket-factory-support="socketFactorySupport" /> - + - + factories = new ArrayList(); factories.add(factory1); factories.add(factory2); - TcpConnection conn1 = makeMockConnection(); - TcpConnection conn2 = makeMockConnection(); + TcpConnectionSupport conn1 = makeMockConnection(); + TcpConnectionSupport conn2 = makeMockConnection(); when(factory1.getConnection()).thenReturn(conn1); when(factory2.getConnection()).thenReturn(conn2); when(factory1.isActive()).thenReturn(true); @@ -94,8 +94,8 @@ public class FailoverClientConnectionFactoryTests { List factories = new ArrayList(); factories.add(factory1); factories.add(factory2); - TcpConnection conn1 = makeMockConnection(); - TcpConnection conn2 = makeMockConnection(); + TcpConnectionSupport conn1 = makeMockConnection(); + TcpConnectionSupport conn2 = makeMockConnection(); when(factory1.getConnection()).thenReturn(conn1); when(factory2.getConnection()).thenReturn(conn2); when(factory1.isActive()).thenReturn(true); @@ -116,8 +116,8 @@ public class FailoverClientConnectionFactoryTests { List factories = new ArrayList(); factories.add(factory1); factories.add(factory2); - TcpConnection conn1 = makeMockConnection(); - TcpConnection conn2 = makeMockConnection(); + TcpConnectionSupport conn1 = makeMockConnection(); + TcpConnectionSupport conn2 = makeMockConnection(); when(factory1.getConnection()).thenReturn(conn1); when(factory2.getConnection()).thenReturn(conn2); when(factory1.isActive()).thenReturn(true); @@ -165,7 +165,7 @@ public class FailoverClientConnectionFactoryTests { List factories = new ArrayList(); factories.add(factory1); factories.add(factory2); - TcpConnection conn1 = makeMockConnection(); + TcpConnectionSupport conn1 = makeMockConnection(); doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) throws Throwable { return null; @@ -189,8 +189,8 @@ public class FailoverClientConnectionFactoryTests { List factories = new ArrayList(); factories.add(factory1); factories.add(factory2); - TcpConnection conn1 = makeMockConnection(); - TcpConnection conn2 = makeMockConnection(); + TcpConnectionSupport conn1 = makeMockConnection(); + TcpConnectionSupport conn2 = makeMockConnection(); when(factory1.getConnection()).thenReturn(conn1); when(factory2.getConnection()).thenReturn(conn2); when(factory1.isActive()).thenReturn(true); @@ -218,8 +218,8 @@ public class FailoverClientConnectionFactoryTests { Mockito.verify(conn1, times(3)).send(message); } - public TcpConnection makeMockConnection() { - TcpConnection connection = mock(TcpConnection.class); + public TcpConnectionSupport makeMockConnection() { + TcpConnectionSupport connection = mock(TcpConnectionSupport.class); when(connection.isOpen()).thenReturn(true); return connection; } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptor.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptor.java index ea8250526a..dbc251bf97 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptor.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptor.java @@ -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. @@ -29,7 +29,7 @@ import org.springframework.integration.support.MessageBuilder; * @since 2.0 * */ -public class HelloWorldInterceptor extends AbstractTcpConnectionInterceptor { +public class HelloWorldInterceptor extends TcpConnectionInterceptorSupport { Log logger = LogFactory.getLog(this.getClass()); @@ -47,9 +47,9 @@ public class HelloWorldInterceptor extends AbstractTcpConnectionInterceptor { private volatile boolean pendingSend; - public HelloWorldInterceptor() { + public HelloWorldInterceptor() { } - + /** * @param hello * @param world @@ -75,7 +75,7 @@ public class HelloWorldInterceptor extends AbstractTcpConnectionInterceptor { throw new MessagingException("Negotiation error", e); } } else { - throw new MessagingException("Negotiation error, expected '" + hello + + throw new MessagingException("Negotiation error, expected '" + hello + "' received '" + payload + "'"); } } else { @@ -84,7 +84,7 @@ public class HelloWorldInterceptor extends AbstractTcpConnectionInterceptor { this.negotiated = true; this.negotiationSemaphore.release(); } else { - throw new MessagingException("Negotiation error - expected '" + world + + throw new MessagingException("Negotiation error - expected '" + world + "' received " + payload); } return true; @@ -93,7 +93,7 @@ public class HelloWorldInterceptor extends AbstractTcpConnectionInterceptor { try { return super.onMessage(message); } finally { - // on the server side, we don't want to close if we are expecting a response + // on the server side, we don't want to close if we are expecting a response if (!(this.isServer() && this.hasRealSender()) && !this.pendingSend) { this.checkDeferredClose(); } @@ -144,6 +144,6 @@ public class HelloWorldInterceptor extends AbstractTcpConnectionInterceptor { } } - + } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptorFactory.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptorFactory.java index 1f4dc8978a..23e8093a15 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptorFactory.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptorFactory.java @@ -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. @@ -30,7 +30,7 @@ public class HelloWorldInterceptorFactory implements public HelloWorldInterceptorFactory() { } - + /** * @param hello * @param world @@ -41,9 +41,9 @@ public class HelloWorldInterceptorFactory implements } - public TcpConnectionInterceptor getInterceptor() { + public TcpConnectionInterceptorSupport getInterceptor() { return new HelloWorldInterceptor(hello, world); } - + } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SocketSupportTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SocketSupportTests.java index 14ae88522d..4e479aff0a 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SocketSupportTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SocketSupportTests.java @@ -40,11 +40,6 @@ import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; import org.springframework.integration.Message; -import org.springframework.integration.ip.tcp.connection.support.DefaultTcpNetSSLSocketFactorySupport; -import org.springframework.integration.ip.tcp.connection.support.DefaultTcpSSLContextSupport; -import org.springframework.integration.ip.tcp.connection.support.TcpSSLContextSupport; -import org.springframework.integration.ip.tcp.connection.support.TcpSocketFactorySupport; -import org.springframework.integration.ip.tcp.connection.support.TcpSocketSupport; import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer; import org.springframework.integration.ip.util.TestingUtilities; import org.springframework.integration.message.GenericMessage; 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 e111cdde8e..50846a58f7 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 @@ -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. @@ -63,7 +63,7 @@ public class TcpMessageMapperTests { TcpMessageMapper mapper = new TcpMessageMapper(); Socket socket = SocketFactory.getDefault().createSocket(); - TcpConnection connection = new AbstractTcpConnection(socket, false, false) { + TcpConnection connection = new TcpConnectionSupport(socket, false, false) { public void run() { } public void send(Message message) throws Exception { @@ -115,7 +115,7 @@ public class TcpMessageMapperTests { TcpMessageMapper mapper = new TcpMessageMapper(); mapper.setApplySequence(true); Socket socket = SocketFactory.getDefault().createSocket(); - TcpConnection connection = new AbstractTcpConnection(socket, false, false) { + TcpConnection connection = new TcpConnectionSupport(socket, false, false) { public void run() { } public void send(Message message) throws Exception { diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java index cfc9b9143f..88548bbdfb 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.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. @@ -38,7 +38,6 @@ import javax.sql.DataSource; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; - import org.springframework.beans.factory.annotation.Autowired; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; @@ -47,8 +46,8 @@ import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.history.MessageHistory; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.store.MessageGroup; -import org.springframework.integration.store.MessageGroupCallback; import org.springframework.integration.store.MessageGroupStore; +import org.springframework.integration.store.MessageGroupStore.MessageGroupCallback; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.util.UUIDConverter; import org.springframework.jdbc.core.JdbcTemplate; @@ -62,6 +61,7 @@ import org.springframework.transaction.annotation.Transactional; * @author Oleg Zhurakousky * @author Gunnar Hillert * @author Artem Bilan + * @author Gary Russell */ @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class)