diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java b/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java index 709e06704e..8c32ff8022 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java @@ -192,7 +192,7 @@ public class SimplePool implements Pool { } else if (this.callback.isStale(item)) { if (logger.isDebugEnabled()) { - logger.debug("Received a stale item, will attempt to get a new one."); + logger.debug("Received a stale item " + item + ", will attempt to get a new one."); } doRemoveItem(item); item = doGetItem(); @@ -241,6 +241,9 @@ public class SimplePool implements Pool { } private void doRemoveItem(T item) { + if (logger.isDebugEnabled()){ + logger.debug("Removing " + item + " from the pool"); + } this.allocated.remove(item); this.inUse.remove(item); this.callback.removedFromPool(item); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java index 39077b51e5..d28e8762c8 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java @@ -45,6 +45,10 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe private final SimplePool> pool; + private final boolean isSharedSessionCapable; + + private volatile long sharedSessionEpoch; + /** * Create a CachingSessionFactory with an unlimited number of sessions. * @param sessionFactory the underlying session factory. @@ -65,18 +69,22 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe public CachingSessionFactory(SessionFactory sessionFactory, int sessionCacheSize) { this.sessionFactory = sessionFactory; this.pool = new SimplePool>(sessionCacheSize, new SimplePool.PoolItemCallback>() { + @Override public Session createForPool() { return CachingSessionFactory.this.sessionFactory.getSession(); } + @Override public boolean isStale(Session session) { return !session.isOpen(); } + @Override public void removedFromPool(Session session) { session.close(); } }); + this.isSharedSessionCapable = sessionFactory instanceof SharedSessionCapable; } @@ -100,78 +108,138 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe /** * Get a session from the pool (or block if none available). */ + @Override public Session getSession() { - return new CachedSession(this.pool.getItem()); + return new CachedSession(this.pool.getItem(), this.sharedSessionEpoch); } /** * Remove (close) any unused sessions in the pool. */ + @Override public void destroy() { this.pool.removeAllIdleItems(); } + /** + * Clear the cache of sessions; also any in-use sessions will be closed when + * returned to the cache. + */ + public synchronized void resetCache() { + if (logger.isDebugEnabled()) { + logger.debug("Cache reset; idle sessions will be removed, in-use sessions will be closed when returned"); + } + if (this.isSharedSessionCapable && ((SharedSessionCapable) this.sessionFactory).isSharedSession()) { + ((SharedSessionCapable) this.sessionFactory).resetSharedSession(); + } + long sharedSessionEpoch = System.nanoTime(); + /* + * Spin until we get a new value - nano precision but may be lower resolution. + * We reset the epoch AFTER resetting the shared session so there is no possibility + * of an "old" session being created in the new epoch. There is a slight possibility + * that a "new" session might appear in the old epoch and thus be closed when returned to + * the cache. + */ + while (sharedSessionEpoch == this.sharedSessionEpoch) { + sharedSessionEpoch = System.nanoTime(); + } + this.sharedSessionEpoch = sharedSessionEpoch; + this.pool.removeAllIdleItems(); + } private class CachedSession implements Session { private final Session targetSession; - private boolean released; + private volatile boolean released; - private CachedSession(Session targetSession) { + /** + * The epoch in which this session was created. + */ + private final long sharedSessionEpoch; + + private CachedSession(Session targetSession, long sharedSessionEpoch) { this.targetSession = targetSession; + this.sharedSessionEpoch = sharedSessionEpoch; } + @Override public synchronized void close() { if (released) { if (logger.isDebugEnabled()){ - logger.debug("Session already released."); + logger.debug("Session " + targetSession + " already released."); } } else { if (logger.isDebugEnabled()){ - logger.debug("Releasing Session back to the pool."); + logger.debug("Releasing Session " + targetSession + " back to the pool."); + } + if (this.sharedSessionEpoch != CachingSessionFactory.this.sharedSessionEpoch) { + if (logger.isDebugEnabled()){ + logger.debug("Closing session " + targetSession + " after reset."); + } + this.targetSession.close(); } pool.releaseItem(targetSession); released = true; } } + @Override public boolean remove(String path) throws IOException{ return this.targetSession.remove(path); } + @Override public F[] list(String path) throws IOException{ return this.targetSession.list(path); } + @Override public void read(String source, OutputStream os) throws IOException{ this.targetSession.read(source, os); } + @Override public void write(InputStream inputStream, String destination) throws IOException{ this.targetSession.write(inputStream, destination); } + @Override public boolean isOpen() { return this.targetSession.isOpen(); } + @Override public void rename(String pathFrom, String pathTo) throws IOException { this.targetSession.rename(pathFrom, pathTo); } + @Override public boolean mkdir(String directory) throws IOException { return this.targetSession.mkdir(directory); } + @Override public boolean exists(String path) throws IOException{ return this.targetSession.exists(path); } + @Override public String[] listNames(String path) throws IOException { return this.targetSession.listNames(path); } + + @Override + public InputStream readRaw(String source) throws IOException { + return this.targetSession.readRaw(source); + } + + @Override + public boolean finalizeRaw() throws IOException { + return this.targetSession.finalizeRaw(); + } + } } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java index 4a5c325002..2b9469c74d 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/Session.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. @@ -35,21 +35,36 @@ public interface Session { boolean remove(String path) throws IOException; T[] list(String path) throws IOException; - + void read(String source, OutputStream outputStream) throws IOException; void write(InputStream inputStream, String destination) throws IOException; - + boolean mkdir(String directory) throws IOException; - + void rename(String pathFrom, String pathTo) throws IOException; void close(); - + boolean isOpen(); - + boolean exists(String path) throws IOException; String[] listNames(String path) throws IOException; + /** + * Retrieve a remote file as a raw {@link InputStream}. + * @param source The path of the remote file + * @return The raw inputStream. + */ + InputStream readRaw(String source) throws IOException; + + /** + * Invoke after closing the InputStream from {@link #readRaw(String)}. + * Required by some session providers. + * @return true if successful. + * @throws IOException + */ + boolean finalizeRaw() throws IOException; + } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/SharedSessionCapable.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/SharedSessionCapable.java new file mode 100644 index 0000000000..f9db8f99bd --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/SharedSessionCapable.java @@ -0,0 +1,39 @@ +/* + * Copyright 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.file.remote.session; + + +/** + * A {@link SessionFactory} that implements this interface is capable of supporting a shared session. + * + * @author Gary Russell + * @since 3.0 + * + */ +public interface SharedSessionCapable { + + /** + * @return true if this factory uses a shared session. + */ + public abstract boolean isSharedSession(); + + /** + * Resets the shared session so the next {@code #getSession()} will return a session + * using a new connection. + */ + public abstract void resetSharedSession(); + +} diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java index 62a776950d..52743c95cf 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java @@ -55,7 +55,6 @@ import org.springframework.messaging.MessagingException; /** * @author Gary Russell * @since 2.1 - * */ @SuppressWarnings("rawtypes") public class RemoteFileOutboundGatewayTests { @@ -63,11 +62,11 @@ public class RemoteFileOutboundGatewayTests { private final String tmpDir = System.getProperty("java.io.tmpdir"); - @Test(expected=IllegalArgumentException.class) + @Test(expected = IllegalArgumentException.class) public void testBad() throws Exception { SessionFactory sessionFactory = mock(SessionFactory.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "bad", "payload"); + (sessionFactory, "bad", "payload"); gw.afterPropertiesSet(); } @@ -106,7 +105,7 @@ public class RemoteFileOutboundGatewayTests { SessionFactory sessionFactory = mock(SessionFactory.class); Session session = mock(Session.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "ls", "payload"); + (sessionFactory, "ls", "payload"); gw.afterPropertiesSet(); when(sessionFactory.getSession()).thenReturn(session); TestLsEntry[] files = fileList(); @@ -128,6 +127,7 @@ public class RemoteFileOutboundGatewayTests { /** * Test a wildcard mget where the full path is returned for each file + * * @throws Exception */ @Test @@ -138,19 +138,25 @@ public class RemoteFileOutboundGatewayTests { private void testMGetWildGuts(final String path1, final String path2) { SessionFactory sessionFactory = mock(SessionFactory.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "mget", "payload"); - gw.setLocalDirectory(new File(this.tmpDir )); + (sessionFactory, "mget", "payload"); + gw.setLocalDirectory(new File(this.tmpDir)); gw.afterPropertiesSet(); new File(this.tmpDir + "/f1").delete(); new File(this.tmpDir + "/f2").delete(); when(sessionFactory.getSession()).thenReturn(new Session() { int n; + + @Override public boolean remove(String path) throws IOException { return false; } + + @Override public Object[] list(String path) throws IOException { return null; } + + @Override public void read(String source, OutputStream outputStream) throws IOException { if (n++ == 0) { @@ -161,25 +167,49 @@ public class RemoteFileOutboundGatewayTests { } outputStream.write("testData".getBytes()); } + + @Override public void write(InputStream inputStream, String destination) throws IOException { } + + @Override public boolean mkdir(String directory) throws IOException { return false; } + + @Override public void rename(String pathFrom, String pathTo) throws IOException { } + + @Override public void close() { } + + @Override public boolean isOpen() { return false; } + + @Override public boolean exists(String path) throws IOException { return false; } + + @Override public String[] listNames(String path) throws IOException { - return new String[] {path1, path2}; + return new String[]{path1, path2}; + } + + @Override + public InputStream readRaw(String source) throws IOException { + return null; + } + + @Override + public boolean finalizeRaw() throws IOException { + return false; } }); @SuppressWarnings("unchecked") @@ -196,40 +226,69 @@ public class RemoteFileOutboundGatewayTests { public void testMGetSingle() throws Exception { SessionFactory sessionFactory = mock(SessionFactory.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "mget", "payload"); - gw.setLocalDirectory(new File(this.tmpDir )); + (sessionFactory, "mget", "payload"); + gw.setLocalDirectory(new File(this.tmpDir)); gw.afterPropertiesSet(); new File(this.tmpDir + "/f1").delete(); when(sessionFactory.getSession()).thenReturn(new Session() { + @Override public boolean remove(String path) throws IOException { return false; } + + @Override public Object[] list(String path) throws IOException { return null; } + + @Override public void read(String source, OutputStream outputStream) throws IOException { outputStream.write("testData".getBytes()); } + + @Override public void write(InputStream inputStream, String destination) throws IOException { } + + @Override public boolean mkdir(String directory) throws IOException { return false; } + + @Override public void rename(String pathFrom, String pathTo) throws IOException { } + + @Override public void close() { } + + @Override public boolean isOpen() { return false; } + + @Override public boolean exists(String path) throws IOException { return false; } + + @Override public String[] listNames(String path) throws IOException { - return new String[] {"f1"}; + return new String[]{"f1"}; + } + + @Override + public InputStream readRaw(String source) throws IOException { + return null; + } + + @Override + public boolean finalizeRaw() throws IOException { + return false; } }); @SuppressWarnings("unchecked") @@ -241,47 +300,76 @@ public class RemoteFileOutboundGatewayTests { out.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)); } - @Test(expected=MessagingException.class) + @Test(expected = MessagingException.class) public void testMGetEmpty() throws Exception { SessionFactory sessionFactory = mock(SessionFactory.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "mget", "payload"); - gw.setLocalDirectory(new File(this.tmpDir )); + (sessionFactory, "mget", "payload"); + gw.setLocalDirectory(new File(this.tmpDir)); gw.setOptions(" -x "); gw.afterPropertiesSet(); new File(this.tmpDir + "/f1").delete(); new File(this.tmpDir + "/f2").delete(); when(sessionFactory.getSession()).thenReturn(new Session() { + @Override public boolean remove(String path) throws IOException { return false; } + + @Override public Object[] list(String path) throws IOException { return null; } + + @Override public void read(String source, OutputStream outputStream) throws IOException { outputStream.write("testData".getBytes()); } + + @Override public void write(InputStream inputStream, String destination) throws IOException { } + + @Override public boolean mkdir(String directory) throws IOException { return false; } + + @Override public void rename(String pathFrom, String pathTo) throws IOException { } + + @Override public void close() { } + + @Override public boolean isOpen() { return false; } + + @Override public boolean exists(String path) throws IOException { return false; } + + @Override public String[] listNames(String path) throws IOException { return new String[0]; } + + @Override + public InputStream readRaw(String source) throws IOException { + return null; + } + + @Override + public boolean finalizeRaw() throws IOException { + return false; + } }); gw.handleRequestMessage(new GenericMessage("testremote/*")); } @@ -290,7 +378,7 @@ public class RemoteFileOutboundGatewayTests { public void testMove() throws Exception { SessionFactory sessionFactory = mock(SessionFactory.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "mv", "payload"); + (sessionFactory, "mv", "payload"); gw.afterPropertiesSet(); Session session = mock(Session.class); final AtomicReference args = new AtomicReference(); @@ -302,7 +390,7 @@ public class RemoteFileOutboundGatewayTests { return null; } }).when(session).rename(anyString(), anyString()); - when (sessionFactory.getSession()).thenReturn(session); + when(sessionFactory.getSession()).thenReturn(session); Message requestMessage = MessageBuilder.withPayload("foo") .setHeader(FileHeaders.RENAME_TO, "bar") .build(); @@ -316,7 +404,7 @@ public class RemoteFileOutboundGatewayTests { public void testMoveWithExpression() throws Exception { SessionFactory sessionFactory = mock(SessionFactory.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "mv", "payload"); + (sessionFactory, "mv", "payload"); gw.setRenameExpression("payload.substring(1)"); gw.afterPropertiesSet(); Session session = mock(Session.class); @@ -329,7 +417,7 @@ public class RemoteFileOutboundGatewayTests { return null; } }).when(session).rename(anyString(), anyString()); - when (sessionFactory.getSession()).thenReturn(session); + when(sessionFactory.getSession()).thenReturn(session); Message out = (Message) gw.handleRequestMessage(new GenericMessage("foo")); assertEquals("oo", out.getHeaders().get(FileHeaders.RENAME_TO)); assertEquals("foo", out.getHeaders().get(FileHeaders.REMOTE_FILE)); @@ -341,7 +429,7 @@ public class RemoteFileOutboundGatewayTests { public void testMoveWithMkDirs() throws Exception { SessionFactory sessionFactory = mock(SessionFactory.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "mv", "payload"); + (sessionFactory, "mv", "payload"); gw.setRenameExpression("'foo/bar/baz'"); gw.afterPropertiesSet(); Session session = mock(Session.class); @@ -356,12 +444,13 @@ public class RemoteFileOutboundGatewayTests { }).when(session).rename(anyString(), anyString()); final List madeDirs = new ArrayList(); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { madeDirs.add((String) invocation.getArguments()[0]); return null; } }).when(session).mkdir(anyString()); - when (sessionFactory.getSession()).thenReturn(session); + when(sessionFactory.getSession()).thenReturn(session); Message requestMessage = MessageBuilder.withPayload("foo") .setHeader(FileHeaders.RENAME_TO, "bar") .build(); @@ -390,7 +479,7 @@ public class RemoteFileOutboundGatewayTests { SessionFactory sessionFactory = mock(SessionFactory.class); Session session = mock(Session.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "ls", "payload"); + (sessionFactory, "ls", "payload"); gw.setOptions("-f"); gw.afterPropertiesSet(); when(sessionFactory.getSession()).thenReturn(session); @@ -407,23 +496,23 @@ public class RemoteFileOutboundGatewayTests { } public TestLsEntry[] level1List() { - return new TestLsEntry[] { - new TestLsEntry("f1", 123, false, false, 1234, "-r--r--r--"), - new TestLsEntry("d1", 0, true, false, 12345, "drw-r--r--"), - new TestLsEntry("f2", 12345, false, false, 123456, "-rw-r--r--") + return new TestLsEntry[]{ + new TestLsEntry("f1", 123, false, false, 1234, "-r--r--r--"), + new TestLsEntry("d1", 0, true, false, 12345, "drw-r--r--"), + new TestLsEntry("f2", 12345, false, false, 123456, "-rw-r--r--") }; } public TestLsEntry[] level2List() { - return new TestLsEntry[] { - new TestLsEntry("d2", 0, true, false, 12345, "drw-r--r--"), - new TestLsEntry("f3", 12345, false, false, 123456, "-rw-r--r--") + return new TestLsEntry[]{ + new TestLsEntry("d2", 0, true, false, 12345, "drw-r--r--"), + new TestLsEntry("f3", 12345, false, false, 123456, "-rw-r--r--") }; } public TestLsEntry[] level3List() { - return new TestLsEntry[] { - new TestLsEntry("f4", 12345, false, false, 123456, "-rw-r--r--") + return new TestLsEntry[]{ + new TestLsEntry("f4", 12345, false, false, 123456, "-rw-r--r--") }; } @@ -432,7 +521,7 @@ public class RemoteFileOutboundGatewayTests { SessionFactory sessionFactory = mock(SessionFactory.class); Session session = mock(Session.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "ls", "payload"); + (sessionFactory, "ls", "payload"); gw.setOptions("-f -R"); gw.afterPropertiesSet(); when(sessionFactory.getSession()).thenReturn(session); @@ -459,7 +548,7 @@ public class RemoteFileOutboundGatewayTests { SessionFactory sessionFactory = mock(SessionFactory.class); Session session = mock(Session.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "ls", "payload"); + (sessionFactory, "ls", "payload"); gw.setOptions("-f -R -dirs"); gw.afterPropertiesSet(); when(sessionFactory.getSession()).thenReturn(session); @@ -488,7 +577,7 @@ public class RemoteFileOutboundGatewayTests { SessionFactory sessionFactory = mock(SessionFactory.class); Session session = mock(Session.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "ls", "payload"); + (sessionFactory, "ls", "payload"); gw.afterPropertiesSet(); when(sessionFactory.getSession()).thenReturn(session); TestLsEntry[] files = new TestLsEntry[0]; @@ -504,7 +593,7 @@ public class RemoteFileOutboundGatewayTests { SessionFactory sessionFactory = mock(SessionFactory.class); Session session = mock(Session.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "ls", "payload"); + (sessionFactory, "ls", "payload"); gw.setOptions("-1"); gw.afterPropertiesSet(); when(sessionFactory.getSession()).thenReturn(session); @@ -523,7 +612,7 @@ public class RemoteFileOutboundGatewayTests { SessionFactory sessionFactory = mock(SessionFactory.class); Session session = mock(Session.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "ls", "payload"); + (sessionFactory, "ls", "payload"); gw.setOptions("-1 -f"); gw.afterPropertiesSet(); when(sessionFactory.getSession()).thenReturn(session); @@ -542,7 +631,7 @@ public class RemoteFileOutboundGatewayTests { SessionFactory sessionFactory = mock(SessionFactory.class); Session session = mock(Session.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "ls", "payload"); + (sessionFactory, "ls", "payload"); gw.setOptions("-1 -dirs"); gw.afterPropertiesSet(); when(sessionFactory.getSession()).thenReturn(session); @@ -562,7 +651,7 @@ public class RemoteFileOutboundGatewayTests { SessionFactory sessionFactory = mock(SessionFactory.class); Session session = mock(Session.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "ls", "payload"); + (sessionFactory, "ls", "payload"); gw.setOptions("-1 -dirs -links"); gw.afterPropertiesSet(); when(sessionFactory.getSession()).thenReturn(session); @@ -583,7 +672,7 @@ public class RemoteFileOutboundGatewayTests { SessionFactory sessionFactory = mock(SessionFactory.class); Session session = mock(Session.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "ls", "payload"); + (sessionFactory, "ls", "payload"); gw.setOptions("-1 -a -f -dirs -links"); gw.afterPropertiesSet(); when(sessionFactory.getSession()).thenReturn(session); @@ -606,7 +695,7 @@ public class RemoteFileOutboundGatewayTests { SessionFactory sessionFactory = mock(SessionFactory.class); Session session = mock(Session.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "ls", "payload"); + (sessionFactory, "ls", "payload"); gw.setOptions("-1 -a -f -dirs -links"); gw.setFilter(new TestPatternFilter("*4")); gw.afterPropertiesSet(); @@ -624,45 +713,75 @@ public class RemoteFileOutboundGatewayTests { public void testGet() throws Exception { SessionFactory sessionFactory = mock(SessionFactory.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "get", "payload"); - gw.setLocalDirectory(new File(this.tmpDir )); + (sessionFactory, "get", "payload"); + gw.setLocalDirectory(new File(this.tmpDir)); gw.afterPropertiesSet(); new File(this.tmpDir + "/f1").delete(); - when(sessionFactory.getSession()).thenReturn(new Session(){ + when(sessionFactory.getSession()).thenReturn(new Session() { private boolean open = true; + + @Override public boolean remove(String path) throws IOException { return false; } + + @Override public TestLsEntry[] list(String path) throws IOException { - return new TestLsEntry[] { + return new TestLsEntry[]{ new TestLsEntry("f1", 1234, false, false, 12345, "-rw-r--r--") }; } + + @Override public void read(String source, OutputStream outputStream) throws IOException { outputStream.write("testfile".getBytes()); } + + @Override public void write(InputStream inputStream, String destination) throws IOException { } + + @Override public boolean mkdir(String directory) throws IOException { return true; } + + @Override public void rename(String pathFrom, String pathTo) throws IOException { } + + @Override public void close() { open = false; } + + @Override public boolean isOpen() { return open; } + + @Override public boolean exists(String path) throws IOException { return true; } + + @Override public String[] listNames(String path) throws IOException { return null; } + + @Override + public InputStream readRaw(String source) throws IOException { + return null; + } + + @Override + public boolean finalizeRaw() throws IOException { + return false; + } }); @SuppressWarnings("unchecked") Message out = (Message) gw.handleRequestMessage(new GenericMessage("f1")); @@ -680,7 +799,7 @@ public class RemoteFileOutboundGatewayTests { public void testGet_P() throws Exception { SessionFactory sessionFactory = mock(SessionFactory.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "get", "payload"); + (sessionFactory, "get", "payload"); gw.setLocalDirectory(new File(this.tmpDir)); gw.setOptions("-P"); gw.afterPropertiesSet(); @@ -688,41 +807,71 @@ public class RemoteFileOutboundGatewayTests { Calendar cal = Calendar.getInstance(); cal.add(Calendar.MONTH, -1); final Date modified = new Date(cal.getTime().getTime() / 1000 * 1000); - when(sessionFactory.getSession()).thenReturn(new Session(){ + when(sessionFactory.getSession()).thenReturn(new Session() { private boolean open = true; + + @Override public boolean remove(String path) throws IOException { return false; } + + @Override public TestLsEntry[] list(String path) throws IOException { - return new TestLsEntry[] { + return new TestLsEntry[]{ new TestLsEntry("f1", 1234, false, false, modified.getTime(), "-rw-r--r--") }; } + + @Override public void read(String source, OutputStream outputStream) throws IOException { outputStream.write("testfile".getBytes()); } + + @Override public void write(InputStream inputStream, String destination) throws IOException { } + + @Override public boolean mkdir(String directory) throws IOException { return true; } + + @Override public void rename(String pathFrom, String pathTo) throws IOException { } + + @Override public void close() { open = false; } + + @Override public boolean isOpen() { return open; } + + @Override public boolean exists(String path) throws IOException { return true; } + + @Override public String[] listNames(String path) throws IOException { return null; } + + @Override + public InputStream readRaw(String source) throws IOException { + return null; + } + + @Override + public boolean finalizeRaw() throws IOException { + return false; + } }); @SuppressWarnings("unchecked") Message out = (Message) gw.handleRequestMessage(new GenericMessage("x/f1")); @@ -743,44 +892,74 @@ public class RemoteFileOutboundGatewayTests { new File(this.tmpDir + "/x").delete(); SessionFactory sessionFactory = mock(SessionFactory.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "get", "payload"); + (sessionFactory, "get", "payload"); gw.setLocalDirectory(new File(this.tmpDir + "/x")); gw.afterPropertiesSet(); - when(sessionFactory.getSession()).thenReturn(new Session(){ + when(sessionFactory.getSession()).thenReturn(new Session() { private boolean open = true; + + @Override public boolean remove(String path) throws IOException { return false; } + + @Override public TestLsEntry[] list(String path) throws IOException { - return new TestLsEntry[] { + return new TestLsEntry[]{ new TestLsEntry("f1", 1234, false, false, 12345, "-rw-r--r--") }; } + + @Override public void read(String source, OutputStream outputStream) throws IOException { outputStream.write("testfile".getBytes()); } + + @Override public void write(InputStream inputStream, String destination) throws IOException { } + + @Override public boolean mkdir(String directory) throws IOException { return true; } + + @Override public void rename(String pathFrom, String pathTo) throws IOException { } + + @Override public void close() { open = false; } + + @Override public boolean isOpen() { return open; } + + @Override public boolean exists(String path) throws IOException { return true; } + + @Override public String[] listNames(String path) throws IOException { return null; } + + @Override + public InputStream readRaw(String source) throws IOException { + return null; + } + + @Override + public boolean finalizeRaw() throws IOException { + return false; + } }); gw.handleRequestMessage(new GenericMessage("f1")); File out = new File(this.tmpDir + "/x/f1"); @@ -793,7 +972,7 @@ public class RemoteFileOutboundGatewayTests { SessionFactory sessionFactory = mock(SessionFactory.class); Session session = mock(Session.class); TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway - (sessionFactory, "rm", "payload"); + (sessionFactory, "rm", "payload"); gw.afterPropertiesSet(); when(sessionFactory.getSession()).thenReturn(session); when(session.remove("testremote/x/f1")).thenReturn(Boolean.TRUE); @@ -812,9 +991,9 @@ public class RemoteFileOutboundGatewayTests { class TestRemoteFileOutboundGateway extends AbstractRemoteFileOutboundGateway { - @SuppressWarnings({ "rawtypes", "unchecked" }) + @SuppressWarnings({"rawtypes", "unchecked"}) public TestRemoteFileOutboundGateway(SessionFactory sessionFactory, - String command, String expression) { + String command, String expression) { super(sessionFactory, Command.toCommand(command), expression); this.setBeanFactory(mock(BeanFactory.class)); } @@ -868,7 +1047,7 @@ class TestLsEntry extends AbstractFileInfo { private final String permissions; public TestLsEntry(String filename, long size, boolean dir, boolean link, - long modified, String permissions) { + long modified, String permissions) { this.filename = filename; this.size = size; this.dir = dir; @@ -877,30 +1056,37 @@ class TestLsEntry extends AbstractFileInfo { this.permissions = permissions; } + @Override public boolean isDirectory() { return this.dir; } + @Override public long getModified() { return this.modified; } + @Override public String getFilename() { return this.filename; } + @Override public boolean isLink() { return this.link; } + @Override public long getSize() { return this.size; } + @Override public String getPermissions() { return this.permissions; } + @Override public TestLsEntry getFileInfo() { return this; } @@ -911,7 +1097,7 @@ class TestLsEntry extends AbstractFileInfo { } -class TestPatternFilter extends AbstractSimplePatternFileListFilter{ +class TestPatternFilter extends AbstractSimplePatternFileListFilter { public TestPatternFilter(String path) { super(path); diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/session/CachingSessionFactoryTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/session/CachingSessionFactoryTests.java new file mode 100644 index 0000000000..f0cca26600 --- /dev/null +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/session/CachingSessionFactoryTests.java @@ -0,0 +1,147 @@ +/* + * Copyright 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.file.remote.session; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; + +import org.junit.Test; + +import org.springframework.integration.test.util.TestUtils; + +/** + * @author Gary Russell + * @since 3.0 + * + */ +public class CachingSessionFactoryTests { + + @Test + public void testCacheAndReset() { + TestSessionFactory factory = new TestSessionFactory(); + CachingSessionFactory cache = new CachingSessionFactory(factory); + Session sess1 = cache.getSession(); + assertEquals("session:1", TestUtils.getPropertyValue(sess1, "targetSession.id")); + Session sess2 = cache.getSession(); + assertEquals("session:2", TestUtils.getPropertyValue(sess2, "targetSession.id")); + sess1.close(); + // session back to pool; should be open and reused. + assertTrue(sess1.isOpen()); + sess1 = cache.getSession(); + assertEquals("session:1", TestUtils.getPropertyValue(sess1, "targetSession.id")); + sess1.close(); + assertTrue(sess1.isOpen()); + // reset the cache; should close idle (sess1); sess2 should closed later + cache.resetCache(); + assertFalse(sess1.isOpen()); + sess1 = cache.getSession(); + assertEquals("session:3", TestUtils.getPropertyValue(sess1, "targetSession.id")); + sess1.close(); + assertTrue(sess1.isOpen()); + // session from previous epoch is closed on return + sess2.close(); + assertFalse(sess2.isOpen()); + cache.resetCache(); + assertFalse(sess1.isOpen()); + } + + private class TestSessionFactory implements SessionFactory { + + private int n; + + @Override + public Session getSession() { + return new TestSession("session:" + ++n); + } + + } + + private class TestSession implements Session { + + @SuppressWarnings("unused") + private final String id; + + private volatile boolean open = true; + + private TestSession(String id) { + this.id = id; + } + + @Override + public boolean remove(String path) throws IOException { + return false; + } + + @Override + public String[] list(String path) throws IOException { + return null; + } + + @Override + public void read(String source, OutputStream outputStream) throws IOException { + } + + @Override + public void write(InputStream inputStream, String destination) throws IOException { + } + + @Override + public boolean mkdir(String directory) throws IOException { + return false; + } + + @Override + public void rename(String pathFrom, String pathTo) throws IOException { + } + + @Override + public void close() { + this.open = false; + } + + @Override + public boolean isOpen() { + return this.open; + } + + @Override + public boolean exists(String path) throws IOException { + return false; + } + + @Override + public String[] listNames(String path) throws IOException { + return null; + } + + @Override + public InputStream readRaw(String source) throws IOException { + return null; + } + + @Override + public boolean finalizeRaw() throws IOException { + return false; + } + + } + +} diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java index 470d1116ca..24718c8f5e 100644 --- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.java +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpSession.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. @@ -19,17 +19,20 @@ package org.springframework.integration.ftp.session; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; +import java.util.concurrent.atomic.AtomicBoolean; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.commons.net.ftp.FTPClient; import org.apache.commons.net.ftp.FTPFile; +import org.apache.commons.net.ftp.FTPReply; + import org.springframework.integration.file.remote.session.Session; import org.springframework.util.Assert; /** * Implementation of {@link Session} for FTP. - * + * * @author Mark Fisher * @author Oleg Zhurakousky * @author Gary Russell @@ -41,6 +44,7 @@ public class FtpSession implements Session { private final FTPClient client; + private final AtomicBoolean readingRaw = new AtomicBoolean(); public FtpSession(FTPClient client) { Assert.notNull(client, "client must not be null"); @@ -48,6 +52,7 @@ public class FtpSession implements Session { } + @Override public boolean remove(String path) throws IOException { Assert.hasText(path, "path must not be null"); boolean completed = this.client.deleteFile(path); @@ -56,17 +61,20 @@ public class FtpSession implements Session { } return completed; } - + + @Override public FTPFile[] list(String path) throws IOException { Assert.hasText(path, "path must not be null"); return this.client.listFiles(path); } + @Override public String[] listNames(String path) throws IOException { Assert.hasText(path, "path must not be null"); return this.client.listNames(path); } + @Override public void read(String path, OutputStream fos) throws IOException { Assert.hasText(path, "path must not be null"); Assert.notNull(fos, "outputStream must not be null"); @@ -78,12 +86,40 @@ public class FtpSession implements Session { logger.info("File has been successfully transfered from: " + path); } + @Override + public InputStream readRaw(String source) throws IOException { + if (!this.readingRaw.compareAndSet(false, true)) { + throw new IOException("Previous raw read was not finalized"); + } + InputStream inputStream = this.client.retrieveFileStream(source); + if (inputStream == null) { + throw new IOException("Failed to obtain InputStream for remote file " + source + ": " + this.client.getReplyCode()); + } + return inputStream; + } + + @Override + public boolean finalizeRaw() throws IOException { + if (!this.readingRaw.compareAndSet(true, false)) { + throw new IOException("Raw read is not in process"); + } + if (this.client.completePendingCommand()) { + int replyCode = this.client.getReplyCode(); + if (logger.isDebugEnabled()) { + logger.debug(this + " finalizeRaw - reply code: " + replyCode); + } + return FTPReply.isPositiveCompletion(replyCode); + } + throw new IOException("completePendingCommandFailed"); + } + + @Override public void write(InputStream inputStream, String path) throws IOException { Assert.notNull(inputStream, "inputStream must not be null"); Assert.hasText(path, "path must not be null"); boolean completed = this.client.storeFile(path, inputStream); if (!completed) { - throw new IOException("Failed to write to '" + path + throw new IOException("Failed to write to '" + path + "'. Server replied with: " + this.client.getReplyString()); } if (logger.isInfoEnabled()) { @@ -91,6 +127,7 @@ public class FtpSession implements Session { } } + @Override public void close() { try { this.client.disconnect(); @@ -102,6 +139,7 @@ public class FtpSession implements Session { } } + @Override public boolean isOpen() { try { this.client.noop(); @@ -112,38 +150,41 @@ public class FtpSession implements Session { return true; } + @Override public void rename(String pathFrom, String pathTo) throws IOException{ this.client.deleteFile(pathTo); boolean completed = this.client.rename(pathFrom, pathTo); if (!completed) { - throw new IOException("Failed to rename '" + pathFrom + + throw new IOException("Failed to rename '" + pathFrom + "' to " + pathTo + "'. Server replied with: " + this.client.getReplyString()); } if (logger.isInfoEnabled()) { logger.info("File has been successfully renamed from: " + pathFrom + " to " + pathTo); } } - + + @Override public boolean mkdir(String remoteDirectory) throws IOException { return this.client.makeDirectory(remoteDirectory); } - + + @Override public boolean exists(String path) throws IOException{ Assert.hasText(path, "'path' must not be empty"); - + String currentWorkingPath = this.client.printWorkingDirectory(); Assert.state(currentWorkingPath != null, "working directory cannot be determined, therefore exists check can not be completed"); boolean exists = false; try { - if (this.client.changeWorkingDirectory(path)){ + if (this.client.changeWorkingDirectory(path)) { exists = true; } - } + } finally { this.client.changeWorkingDirectory(currentWorkingPath); } - + return exists; } } diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/TesFtpServer.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/TesFtpServer.java index 9d059f3f10..e0a62d2d3d 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/TesFtpServer.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/TesFtpServer.java @@ -17,6 +17,7 @@ package org.springframework.integration.ftp; import java.io.File; +import java.io.FileOutputStream; import java.io.IOException; import java.util.Arrays; @@ -82,13 +83,22 @@ public class TesFtpServer { sourceFtpDirectory.mkdir(); File file = new File(sourceFtpDirectory, "ftpSource1.txt"); file.createNewFile(); + FileOutputStream fos = new FileOutputStream(file); + fos.write("source1".getBytes()); + fos.close(); file = new File(sourceFtpDirectory, "ftpSource2.txt"); file.createNewFile(); + fos = new FileOutputStream(file); + fos.write("source2".getBytes()); + fos.close(); File subSourceFtpDirectory = new File(sourceFtpDirectory, "subFtpSource"); subSourceFtpDirectory.mkdir(); file = new File(subSourceFtpDirectory, "subFtpSource1.txt"); file.createNewFile(); + fos = new FileOutputStream(file); + fos.write("subSource1".getBytes()); + fos.close(); targetFtpDirectory = new File(ftpRootFolder, "ftpTarget"); targetFtpDirectory.mkdirs(); diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java index 36b5cb76b1..179a370ae7 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java @@ -19,8 +19,10 @@ package org.springframework.integration.ftp.outbound; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import java.io.ByteArrayOutputStream; import java.io.File; import java.util.List; @@ -31,12 +33,14 @@ import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.ftp.TesFtpServer; import org.springframework.messaging.Message; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.util.FileCopyUtils; /** * @author Artem Bilan @@ -170,4 +174,21 @@ public class FtpServerOutboundTests { } + @Test + public void testInt3100RawGET() throws Exception { + Session session = this.ftpServer.ftpSessionFactory().getSession(); + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + FileCopyUtils.copy(session.readRaw("ftpSource/ftpSource1.txt"), baos); + assertTrue(session.finalizeRaw()); + assertEquals("source1", new String(baos.toByteArray())); + + baos = new ByteArrayOutputStream(); + FileCopyUtils.copy(session.readRaw("ftpSource/ftpSource2.txt"), baos); + assertTrue(session.finalizeRaw()); + assertEquals("source2", new String(baos.toByteArray())); + + session.close(); + } + + } diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpInboundEndpointParser.java b/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpInboundEndpointParser.java index d60a447358..ef8685dd4d 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpInboundEndpointParser.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpInboundEndpointParser.java @@ -17,6 +17,7 @@ package org.springframework.integration.http.config; import java.util.List; + import org.w3c.dom.Element; import org.springframework.beans.factory.BeanDefinitionStoreException; @@ -74,7 +75,7 @@ public class HttpInboundEndpointParser extends AbstractSingleBeanDefinitionParse throws BeanDefinitionStoreException { String id = super.resolveId(element, definition, parserContext); - if (!element.hasAttribute(getInputChannelAttributeName())) { + if (!this.expectReply && !element.hasAttribute("channel")) { // the created channel will get the 'id', so the adapter's bean name includes a suffix id = id + ".adapter"; } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java index 3276096978..54abca2327 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java @@ -18,14 +18,17 @@ package org.springframework.integration.jms.config; import org.w3c.dom.Element; +import org.springframework.beans.factory.BeanDefinitionStoreException; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.parsing.BeanComponentDefinition; +import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.integration.jms.JmsMessageDrivenEndpoint; +import org.springframework.jms.listener.DefaultMessageListenerContainer; import org.springframework.util.StringUtils; /** @@ -79,6 +82,22 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition return JmsMessageDrivenEndpoint.class.getName(); } + @Override + protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext) + throws BeanDefinitionStoreException { + String id = super.resolveId(element, definition, parserContext); + + if (!this.expectReply && !element.hasAttribute("channel")) { + // the created channel will get the 'id', so the adapter's bean name includes a suffix + id = id + ".adapter"; + } + if (!StringUtils.hasText(id)) { + id = BeanDefinitionReaderUtils.generateBeanName(definition, parserContext.getRegistry()); + } + + return id; + } + @Override protected boolean shouldGenerateId() { return false; @@ -100,7 +119,11 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition } private String parseMessageListenerContainer(Element element, ParserContext parserContext) { + String containerClass = element.getAttribute("container-class"); if (element.hasAttribute("container")) { + if (StringUtils.hasText(containerClass)) { + parserContext.getReaderContext().error("Cannot have both 'container' and 'container-class'", element); + } for (String containerAttribute : containerAttributes) { if (element.hasAttribute(containerAttribute)) { parserContext.getReaderContext().error("The '" + containerAttribute + @@ -110,8 +133,13 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition return element.getAttribute("container"); } // otherwise, we build a DefaultMessageListenerContainer instance - BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition( - "org.springframework.jms.listener.DefaultMessageListenerContainer"); + BeanDefinitionBuilder builder; + if (StringUtils.hasText(containerClass)) { + builder = BeanDefinitionBuilder.genericBeanDefinition(containerClass); + } + else { + builder = BeanDefinitionBuilder.genericBeanDefinition(DefaultMessageListenerContainer.class); + } String destinationAttribute = this.expectReply ? "request-destination" : "destination"; String destinationNameAttribute = this.expectReply ? "request-destination-name" : "destination-name"; String pubSubDomainAttribute = this.expectReply ? "request-pub-sub-domain" : "pub-sub-domain"; @@ -198,7 +226,11 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel"); } else { - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "channel", "requestChannel"); + String channelName = element.getAttribute("channel"); + if (!StringUtils.hasText(channelName)) { + channelName = IntegrationNamespaceUtils.createDirectChannel(element, parserContext); + } + builder.addPropertyReference("requestChannel", channelName); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout", "requestTimeout"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-payload", "extractRequestPayload"); } diff --git a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-3.0.xsd b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-3.0.xsd index 2661295b59..7663d10173 100644 --- a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-3.0.xsd +++ b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-3.0.xsd @@ -1201,6 +1201,11 @@ + @@ -1208,6 +1213,22 @@ + + + + + + + + + + + diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsMessageDrivenChannelAdapterParserTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsMessageDrivenChannelAdapterParserTests.java index a27980560d..eda222fd42 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsMessageDrivenChannelAdapterParserTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsMessageDrivenChannelAdapterParserTests.java @@ -26,18 +26,20 @@ import org.junit.Test; import org.springframework.beans.DirectFieldAccessor; import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.messaging.Message; -import org.springframework.messaging.PollableChannel; import org.springframework.integration.history.MessageHistory; import org.springframework.integration.jms.JmsMessageDrivenEndpoint; import org.springframework.integration.test.util.TestUtils; import org.springframework.jms.listener.AbstractMessageListenerContainer; import org.springframework.jms.listener.DefaultMessageListenerContainer; import org.springframework.jms.support.destination.JmsDestinationAccessor; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.PollableChannel; /** * @author Mark Fisher * @author Michael Bannister + * @author Gary Russell */ public class JmsMessageDrivenChannelAdapterParserTests { @@ -58,6 +60,7 @@ public class JmsMessageDrivenChannelAdapterParserTests { assertNotNull("message should not be null", message); assertEquals("test [with selector: TestProperty = 'foo']", message.getPayload()); endpoint.stop(); + context.close(); } @Test @@ -68,6 +71,7 @@ public class JmsMessageDrivenChannelAdapterParserTests { JmsDestinationAccessor container = (JmsDestinationAccessor) new DirectFieldAccessor(endpoint).getPropertyValue("listenerContainer"); assertEquals(Boolean.TRUE, container.isPubSubDomain()); endpoint.stop(); + context.close(); } @Test @@ -81,66 +85,91 @@ public class JmsMessageDrivenChannelAdapterParserTests { assertEquals("testDurableSubscriptionName", container.getDurableSubscriptionName()); assertEquals("testClientId", container.getClientId()); endpoint.stop(); + context.close(); } @Test public void adapterWithTaskExecutor() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "jmsInboundWithTaskExecutor.xml", this.getClass()); - JmsMessageDrivenEndpoint endpoint = context.getBean("messageDrivenAdapter", JmsMessageDrivenEndpoint.class); + JmsMessageDrivenEndpoint endpoint = context.getBean("messageDrivenAdapter.adapter", JmsMessageDrivenEndpoint.class); DefaultMessageListenerContainer container = TestUtils.getPropertyValue(endpoint, "listenerContainer", DefaultMessageListenerContainer.class); assertSame(context.getBean("exec"), TestUtils.getPropertyValue(container, "taskExecutor")); endpoint.stop(); + context.close(); } @Test - public void testGatewayWithReceiveTimeout() { + public void testAdapterWithReceiveTimeout() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "jmsInboundWithContainerSettings.xml", this.getClass()); - JmsMessageDrivenEndpoint gateway = (JmsMessageDrivenEndpoint) context.getBean("adapterWithReceiveTimeout"); - gateway.start(); + JmsMessageDrivenEndpoint adapter = (JmsMessageDrivenEndpoint) context.getBean("adapterWithReceiveTimeout.adapter"); + adapter.start(); AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) - new DirectFieldAccessor(gateway).getPropertyValue("listenerContainer"); + new DirectFieldAccessor(adapter).getPropertyValue("listenerContainer"); assertEquals(1111L, new DirectFieldAccessor(container).getPropertyValue("receiveTimeout")); - gateway.stop(); + adapter.stop(); + context.close(); } @Test - public void testGatewayWithRecoveryInterval() { + public void testAdapterWithRecoveryInterval() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "jmsInboundWithContainerSettings.xml", this.getClass()); - JmsMessageDrivenEndpoint gateway = (JmsMessageDrivenEndpoint) context.getBean("adapterWithRecoveryInterval"); - gateway.start(); + JmsMessageDrivenEndpoint adapter = (JmsMessageDrivenEndpoint) context.getBean("adapterWithRecoveryInterval.adapter"); + adapter.start(); AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) - new DirectFieldAccessor(gateway).getPropertyValue("listenerContainer"); + new DirectFieldAccessor(adapter).getPropertyValue("listenerContainer"); assertEquals(2222L, new DirectFieldAccessor(container).getPropertyValue("recoveryInterval")); - gateway.stop(); + adapter.stop(); + context.close(); } @Test - public void testGatewayWithIdleTaskExecutionLimit() { + public void testAdapterWithIdleTaskExecutionLimit() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "jmsInboundWithContainerSettings.xml", this.getClass()); - JmsMessageDrivenEndpoint gateway = (JmsMessageDrivenEndpoint) context.getBean("adapterWithIdleTaskExecutionLimit"); - gateway.start(); + JmsMessageDrivenEndpoint adapter = (JmsMessageDrivenEndpoint) context.getBean("adapterWithIdleTaskExecutionLimit.adapter"); + adapter.start(); AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) - new DirectFieldAccessor(gateway).getPropertyValue("listenerContainer"); + new DirectFieldAccessor(adapter).getPropertyValue("listenerContainer"); assertEquals(7, new DirectFieldAccessor(container).getPropertyValue("idleTaskExecutionLimit")); - gateway.stop(); + adapter.stop(); + context.close(); } @Test - public void testGatewayWithIdleConsumerLimit() { + public void testAdapterWithIdleConsumerLimit() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "jmsInboundWithContainerSettings.xml", this.getClass()); - JmsMessageDrivenEndpoint gateway = (JmsMessageDrivenEndpoint) context.getBean("adapterWithIdleConsumerLimit"); - gateway.start(); + JmsMessageDrivenEndpoint adapter = (JmsMessageDrivenEndpoint) context.getBean("adapterWithIdleConsumerLimit.adapter"); + adapter.start(); AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) - new DirectFieldAccessor(gateway).getPropertyValue("listenerContainer"); + new DirectFieldAccessor(adapter).getPropertyValue("listenerContainer"); assertEquals(33, new DirectFieldAccessor(container).getPropertyValue("idleConsumerLimit")); assertEquals(3, new DirectFieldAccessor(container).getPropertyValue("cacheLevel")); - gateway.stop(); + adapter.stop(); + context.close(); + } + + @Test + public void testAdapterWithContainerClass() { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "jmsInboundWithContainerClass.xml", this.getClass()); + JmsMessageDrivenEndpoint adapter = context.getBean("adapterWithIdleConsumerLimit.adapter", JmsMessageDrivenEndpoint.class); + MessageChannel channel = context.getBean("adapterWithIdleConsumerLimit", MessageChannel.class); + assertSame(channel, TestUtils.getPropertyValue(adapter, "listener.gatewayDelegate.requestChannel")); + adapter.start(); + FooContainer container = TestUtils.getPropertyValue(adapter, "listenerContainer", FooContainer.class); + assertEquals(33, new DirectFieldAccessor(container).getPropertyValue("idleConsumerLimit")); + assertEquals(3, new DirectFieldAccessor(container).getPropertyValue("cacheLevel")); + adapter.stop(); + context.close(); + } + + public static final class FooContainer extends DefaultMessageListenerContainer { + } } diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsInboundWithContainerClass.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsInboundWithContainerClass.xml new file mode 100644 index 0000000000..34bba45821 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsInboundWithContainerClass.xml @@ -0,0 +1,29 @@ + + + + + + + + + + + + + + diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParser.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParser.java index 13aab72337..2a22976734 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParser.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParser.java @@ -45,6 +45,7 @@ public class RedisInboundChannelAdapterParser extends AbstractChannelAdapterPars builder.addConstructorArgReference(connectionFactory); builder.addPropertyReference("outputChannel", channelName); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "topics"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "topic-patterns"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "serializer", true); diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParser.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParser.java index 18996384da..401b413dea 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParser.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParser.java @@ -50,6 +50,7 @@ public class RedisQueueInboundChannelAdapterParser extends AbstractChannelAdapte IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "expect-message"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "receive-timeout"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "recovery-interval"); builder.addPropertyReference("outputChannel", channelName); return builder.getBeanDefinition(); diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java index 737a8c6ca4..e58b4228fd 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java @@ -21,7 +21,9 @@ import java.util.List; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.listener.ChannelTopic; +import org.springframework.data.redis.listener.PatternTopic; import org.springframework.data.redis.listener.RedisMessageListenerContainer; +import org.springframework.data.redis.listener.Topic; import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; import org.springframework.data.redis.serializer.RedisSerializer; import org.springframework.data.redis.serializer.StringRedisSerializer; @@ -35,6 +37,7 @@ import org.springframework.util.Assert; * @author Mark Fisher * @author Oleg Zhurakousky * @author Gary Russell + * @author Artem Bilan * @since 2.1 */ public class RedisInboundChannelAdapter extends MessageProducerSupport { @@ -45,6 +48,8 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport { private volatile String[] topics; + private volatile String[] topicPatterns; + private volatile RedisSerializer serializer = new StringRedisSerializer(); public RedisInboundChannelAdapter(RedisConnectionFactory connectionFactory) { @@ -60,6 +65,10 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport { this.topics = topics; } + public void setTopicPatterns(String... topicPatterns) { + this.topicPatterns = topicPatterns; + } + public void setMessageConverter(MessageConverter messageConverter) { Assert.notNull(messageConverter, "messageConverter must not be null"); this.messageConverter = messageConverter; @@ -73,13 +82,32 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport { @Override protected void onInit() { super.onInit(); - Assert.notEmpty(this.topics, "at least one topis is required for subscription"); + boolean hasTopics = false; + if (this.topics != null) { + Assert.noNullElements(this.topics, "'topics' may not contain null elements."); + hasTopics = true; + } + boolean hasPatterns = false; + if (this.topicPatterns != null) { + Assert.noNullElements(this.topicPatterns, "'topicPatterns' may not contain null elements."); + hasPatterns = true; + + } + Assert.state(hasTopics || hasPatterns, "at least one topic or topic pattern is required for subscription."); + MessageListenerDelegate delegate = new MessageListenerDelegate(); MessageListenerAdapter adapter = new MessageListenerAdapter(delegate); adapter.setSerializer(this.serializer); - List topicList = new ArrayList(); - for (String topic : this.topics) { - topicList.add(new ChannelTopic(topic)); + List topicList = new ArrayList(); + if (hasTopics) { + for (String topic : this.topics) { + topicList.add(new ChannelTopic(topic)); + } + } + if (hasPatterns) { + for (String pattern : this.topicPatterns) { + topicList.add(new PatternTopic(pattern)); + } } adapter.afterPropertiesSet(); this.container.addMessageListener(adapter, topicList); diff --git a/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-3.0.xsd b/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-3.0.xsd index d34cfb6e0e..141e6eaac2 100644 --- a/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-3.0.xsd +++ b/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-3.0.xsd @@ -147,6 +147,13 @@ + + + + Redis topic patterns as a comma-delimited list of Strings. + + + + + + + Specify the time in milliseconds for which the listener task should sleep after catching + an Exception on a Redis operation, before restarting the listener task. + Default is 5 seconds. + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests-context.xml index ac2fa84b26..658fb11b95 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests-context.xml +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests-context.xml @@ -7,7 +7,7 @@ http://www.springframework.org/schema/integration/redis http://www.springframework.org/schema/integration/redis/spring-integration-redis.xsd"> @@ -25,14 +25,14 @@ class="org.springframework.integration.redis.config.RedisInboundChannelAdapterParserTests$TestMessageConverter" /> + id="autoChannel" topics="foo1, bar1" error-channel="testErrorChannel" + message-converter="testConverter" auto-startup="false"/> - + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java index f34cef85a7..dc1b29efe1 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java @@ -20,7 +20,9 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; +import org.hamcrest.Matchers; import org.junit.Test; import org.junit.runner.RunWith; @@ -29,12 +31,14 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.ApplicationContext; import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.listener.RedisMessageListenerContainer; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.redis.inbound.RedisInboundChannelAdapter; import org.springframework.integration.redis.rules.RedisAvailable; import org.springframework.integration.redis.rules.RedisAvailableTests; import org.springframework.integration.support.converter.SimpleMessageConverter; import org.springframework.integration.test.util.TestUtils; +import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -59,7 +63,6 @@ public class RedisInboundChannelAdapterParserTests extends RedisAvailableTests { private RedisInboundChannelAdapter autoChannelAdapter; @Test - @RedisAvailable public void validateConfiguration() { RedisInboundChannelAdapter adapter = context.getBean("adapter", RedisInboundChannelAdapter.class); assertEquals("adapter", adapter.getComponentName()); @@ -79,18 +82,24 @@ public class RedisInboundChannelAdapterParserTests extends RedisAvailableTests { @Test @RedisAvailable public void testInboundChannelAdapterMessaging() throws Exception { + RedisInboundChannelAdapter adapter = context.getBean("adapter", RedisInboundChannelAdapter.class); + this.awaitContainerSubscribedWithPatterns(TestUtils.getPropertyValue(adapter, "container", RedisMessageListenerContainer.class)); + RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest(); connectionFactory.getConnection().publish("foo".getBytes(), "Hello Redis from foo".getBytes()); + connectionFactory.getConnection().publish("bar".getBytes(), "Hello Redis from bar".getBytes()); QueueChannel receiveChannel = context.getBean("receiveChannel", QueueChannel.class); - assertEquals("Hello Redis from foo", receiveChannel.receive(2000).getPayload()); - connectionFactory.getConnection().publish("bar".getBytes(), "Hello Redis from bar".getBytes()); - assertEquals("Hello Redis from bar", receiveChannel.receive(2000).getPayload()); + for (int i = 0; i < 3; i++) { + Message receive = receiveChannel.receive(2000); + assertNotNull(receive); + assertThat(receive.getPayload(), Matchers. isOneOf("Hello Redis from foo", "Hello Redis from bar")); + } + } @Test - @RedisAvailable public void testAutoChannel() { assertSame(autoChannel, TestUtils.getPropertyValue(autoChannelAdapter, "outputChannel")); } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests-context.xml similarity index 98% rename from spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests-context.xml rename to spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests-context.xml index 6c7558adf3..5700eee8ef 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests-context.xml +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests-context.xml @@ -28,6 +28,7 @@ serializer="serializer" error-channel="errorChannel" receive-timeout="2000" + recovery-interval="3000" task-executor="executor" auto-startup="false" phase="100"/> diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests.java similarity index 94% rename from spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests.java rename to spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests.java index 5d4750989b..c8e8358c16 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParserTests.java @@ -46,7 +46,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; */ @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) -public class RedisMessageDrivenEndpointParserTests { +public class RedisQueueInboundChannelAdapterParserTests { @Autowired @Qualifier("redisConnectionFactory") @@ -90,6 +90,7 @@ public class RedisMessageDrivenEndpointParserTests { assertEquals("si.test.Int3017.Inbound1", TestUtils.getPropertyValue(this.defaultAdapter, "boundListOperations.key")); assertFalse(TestUtils.getPropertyValue(this.defaultAdapter, "expectMessage", Boolean.class)); assertEquals(new Long(1000), TestUtils.getPropertyValue(this.defaultAdapter, "receiveTimeout", Long.class)); + assertEquals(new Long(5000), TestUtils.getPropertyValue(this.defaultAdapter, "recoveryInterval", Long.class)); assertNull(TestUtils.getPropertyValue(this.defaultAdapter, "errorChannel")); assertThat(TestUtils.getPropertyValue(this.defaultAdapter, "taskExecutor"), Matchers.instanceOf(ErrorHandlingTaskExecutor.class)); assertThat(TestUtils.getPropertyValue(this.defaultAdapter, "serializer"), Matchers.instanceOf(JdkSerializationRedisSerializer.class)); @@ -103,6 +104,7 @@ public class RedisMessageDrivenEndpointParserTests { assertEquals("si.test.Int3017.Inbound2", TestUtils.getPropertyValue(this.customAdapter, "boundListOperations.key")); assertTrue(TestUtils.getPropertyValue(this.customAdapter, "expectMessage", Boolean.class)); assertEquals(new Long(2000), TestUtils.getPropertyValue(this.customAdapter, "receiveTimeout", Long.class)); + assertEquals(new Long(3000), TestUtils.getPropertyValue(this.customAdapter, "recoveryInterval", Long.class)); assertSame(this.errorChannel, TestUtils.getPropertyValue(this.customAdapter, "errorChannel")); assertSame(this.taskExecutor, TestUtils.getPropertyValue(this.customAdapter, "taskExecutor")); assertSame(this.serializer, TestUtils.getPropertyValue(this.customAdapter, "serializer")); diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java index 2709daca91..d701b3ace7 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java @@ -20,12 +20,15 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; import java.util.ArrayList; import java.util.Arrays; import java.util.Date; import java.util.List; import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import org.hamcrest.Matchers; import org.junit.Test; @@ -213,6 +216,8 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { final List exceptionEvents = new ArrayList(); + final CountDownLatch exceptionsLatch = new CountDownLatch(2); + RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, this.connectionFactory); endpoint.setBeanFactory(Mockito.mock(BeanFactory.class)); endpoint.setApplicationEventPublisher(new ApplicationEventPublisher() { @@ -220,6 +225,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { @Override public void publishEvent(ApplicationEvent event) { exceptionEvents.add(event); + exceptionsLatch.countDown(); } }); endpoint.setOutputChannel(channel); @@ -228,16 +234,26 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests { endpoint.afterPropertiesSet(); endpoint.start(); + int n = 0; + do { + n++; + if (n == 100) { + break; + } + Thread.sleep(100); + } while (!endpoint.isListening()); + + assertTrue(n < 100); + ((DisposableBean) this.connectionFactory).destroy(); - Thread.sleep(300); + assertTrue(exceptionsLatch.await(10, TimeUnit.SECONDS)); - assertThat(exceptionEvents.size(), Matchers.greaterThan(0)); for (ApplicationEvent exceptionEvent : exceptionEvents) { assertThat(exceptionEvent, Matchers.instanceOf(RedisExceptionEvent.class)); assertSame(endpoint, exceptionEvent.getSource()); assertThat(((IntegrationEvent) exceptionEvent).getCause().getClass(), - Matchers.isIn(Arrays.> asList(RedisSystemException.class, RedisConnectionFailureException.class))); + Matchers.isIn(Arrays.>asList(RedisSystemException.class, RedisConnectionFailureException.class))); } ((InitializingBean) this.connectionFactory).afterPropertiesSet(); diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java index 4adb288a48..b1d089a889 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java @@ -68,9 +68,25 @@ public class RedisAvailableTests { while (n++ < 100 && !connection.isSubscribed()) { Thread.sleep(100); } + // TODO: remove this additional delay when/if https://jira.springsource.org/browse/DATAREDIS-242 is resolved + Thread.sleep(250); assertTrue("RedisMessageListenerContainer Failed to Subscribe", n < 100); } + protected void awaitContainerSubscribedWithPatterns(RedisMessageListenerContainer container) throws Exception { + this.awaitContainerSubscribed(container); + RedisConnection connection = TestUtils.getPropertyValue(container, "subscriptionTask.connection", + RedisConnection.class); + + int n = 0; + while (n++ < 100 && connection.getSubscription().getPatterns().size() == 0) { + Thread.sleep(100); + } + // TODO: remove this additional delay when/if https://jira.springsource.org/browse/DATAREDIS-242 is resolved + Thread.sleep(250); + assertTrue("RedisMessageListenerContainer Failed to Subscribe with patterns", n < 100); + } + protected void prepareList(RedisConnectionFactory connectionFactory){ StringRedisTemplate redisTemplate = new StringRedisTemplate(); @@ -119,4 +135,5 @@ public class RedisAvailableTests { ops.add("Abraham Lincoln", 19); ops.add("George Washington", 18); } + } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java index 3f33c12e32..9bad196505 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,11 +17,14 @@ package org.springframework.integration.sftp.session; import java.util.Properties; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.ReentrantReadWriteLock; import org.springframework.beans.factory.BeanCreationException; import org.springframework.core.io.Resource; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.file.remote.session.SharedSessionCapable; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -39,10 +42,11 @@ import com.jcraft.jsch.UserInfo; * @author Mario Gray * @author Oleg Zhurakousky * @author Gunnar Hillert + * @author Gary Russell * * @since 2.0 */ -public class DefaultSftpSessionFactory implements SessionFactory { +public class DefaultSftpSessionFactory implements SessionFactory, SharedSessionCapable { private volatile String host; @@ -76,9 +80,35 @@ public class DefaultSftpSessionFactory implements SessionFactory { private volatile Boolean enableDaemonThread; + private final JSch jsch; - private final JSch jsch = new JSch(); + private final boolean isSharedSession; + private volatile JSchSessionWrapper sharedJschSession; + + private final ReentrantReadWriteLock sharedSessionLock = new ReentrantReadWriteLock(); + + + public DefaultSftpSessionFactory() { + this(false); + } + + /** + * @param isSharedSession + */ + public DefaultSftpSessionFactory(boolean isSharedSession) { + this(new JSch(), isSharedSession); + } + + /** + * Intended for use in tests so the jsch can be mocked. + * @param jsch + * @param isSharedSession + */ + public DefaultSftpSessionFactory(JSch jsch, boolean isSharedSession) { + this.jsch = jsch; + this.isSharedSession = isSharedSession; + } /** * The url of the host you want connect to. This is a mandatory property. @@ -257,9 +287,35 @@ public class DefaultSftpSessionFactory implements SessionFactory { Assert.isTrue(StringUtils.hasText(this.password) || this.privateKey != null, "either a password or a private key is required"); try { - com.jcraft.jsch.Session jschSession = this.initJschSession(); + JSchSessionWrapper jschSession; + if (this.isSharedSession) { + this.sharedSessionLock.readLock().lock(); + try { + if (this.sharedJschSession == null || !this.sharedJschSession.isConnected()) { + this.sharedSessionLock.readLock().unlock(); + this.sharedSessionLock.writeLock().lock(); + try { + if (this.sharedJschSession == null || !this.sharedJschSession.isConnected()) { + this.sharedJschSession = new JSchSessionWrapper(initJschSession()); + } + } + finally { + this.sharedSessionLock.readLock().lock(); + this.sharedSessionLock.writeLock().unlock(); + } + } + } + finally { + this.sharedSessionLock.readLock().unlock(); + } + jschSession = this.sharedJschSession; + } + else { + jschSession = new JSchSessionWrapper(initJschSession()); + } SftpSession sftpSession = new SftpSession(jschSession); sftpSession.connect(); + jschSession.addChannel(); return sftpSession; } catch (Exception e) { @@ -327,6 +383,16 @@ public class DefaultSftpSessionFactory implements SessionFactory { return jschSession; } + @Override + public final boolean isSharedSession() { + return this.isSharedSession; + } + + @Override + public void resetSharedSession() { + Assert.state(this.isSharedSession, "Shared sessions are not being used"); + this.sharedJschSession = null; + } /** * this is a simple, optimistic implementation of the UserInfo interface. @@ -370,4 +436,38 @@ public class DefaultSftpSessionFactory implements SessionFactory { } } + /** + * A wrapper for a JSch session that maintains a channel count and + * physically disconnects when the last channel is closed. + * + */ + public class JSchSessionWrapper { + + private final com.jcraft.jsch.Session session; + + private final AtomicInteger channels = new AtomicInteger(); + + JSchSessionWrapper(com.jcraft.jsch.Session session) { + this.session = session; + } + + public void addChannel() { + this.channels.incrementAndGet(); + } + + public void close() { + if (channels.decrementAndGet() <= 0) { + this.session.disconnect(); + } + } + + public final com.jcraft.jsch.Session getSession() { + return session; + } + + public boolean isConnected() { + return session.isConnected(); + } + + } } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java index 96cdb9bf12..aadd108ae6 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.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. @@ -25,8 +25,10 @@ import java.util.Vector; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + import org.springframework.core.NestedIOException; import org.springframework.integration.file.remote.session.Session; +import org.springframework.integration.sftp.session.DefaultSftpSessionFactory.JSchSessionWrapper; import org.springframework.util.Assert; import org.springframework.util.FileCopyUtils; @@ -52,15 +54,26 @@ class SftpSession implements Session { private final com.jcraft.jsch.Session jschSession; + private final JSchSessionWrapper wrapper; + private volatile ChannelSftp channel; + private volatile boolean closed; + public SftpSession(com.jcraft.jsch.Session jschSession) { Assert.notNull(jschSession, "jschSession must not be null"); this.jschSession = jschSession; + this.wrapper = null; } + public SftpSession(DefaultSftpSessionFactory.JSchSessionWrapper wrapper) { + Assert.notNull(wrapper, "wrapper must not be null"); + this.jschSession = wrapper.getSession(); + this.wrapper = wrapper; + } + @Override public boolean remove(String path) throws IOException { Assert.state(this.channel != null, "session is not connected"); try { @@ -72,6 +85,7 @@ class SftpSession implements Session { } } + @Override public LsEntry[] list(String path) throws IOException { Assert.state(this.channel != null, "session is not connected"); try { @@ -92,6 +106,7 @@ class SftpSession implements Session { return new LsEntry[0]; } + @Override public String[] listNames(String path) throws IOException { LsEntry[] entries = this.list(path); List names = new ArrayList(); @@ -107,6 +122,7 @@ class SftpSession implements Session { } + @Override public void read(String source, OutputStream os) throws IOException { Assert.state(this.channel != null, "session is not connected"); try { @@ -114,10 +130,26 @@ class SftpSession implements Session { FileCopyUtils.copy(is, os); } catch (SftpException e) { - throw new NestedIOException("failed to read file", e); + throw new NestedIOException("failed to read file " + source, e); } } + @Override + public InputStream readRaw(String source) throws IOException { + try { + return this.channel.get(source); + } + catch (SftpException e) { + throw new NestedIOException("failed to read file " + source, e); + } + } + + @Override + public boolean finalizeRaw() throws IOException { + return true; + } + + @Override public void write(InputStream inputStream, String destination) throws IOException { Assert.state(this.channel != null, "session is not connected"); try { @@ -128,38 +160,48 @@ class SftpSession implements Session { } } + @Override public void close() { - if (this.jschSession.isConnected()) { - this.jschSession.disconnect(); + this.closed = true; + if (this.wrapper != null) { + this.channel.disconnect(); + this.wrapper.close(); + } + else { + if (this.jschSession.isConnected()) { + this.jschSession.disconnect(); + } } } + @Override public boolean isOpen() { - return this.jschSession.isConnected(); + return !this.closed && this.jschSession.isConnected(); } + @Override public void rename(String pathFrom, String pathTo) throws IOException { - try { + try { this.channel.rename(pathFrom, pathTo); - } + } catch (SftpException sftpex) { if (logger.isDebugEnabled()){ - logger.debug("Initial File rename failed, possibly because file already exists. Will attempt to delete file: " + logger.debug("Initial File rename failed, possibly because file already exists. Will attempt to delete file: " + pathTo + " and execute rename again."); } - try { + try { this.remove(pathTo); if (logger.isDebugEnabled()) { logger.debug("Delete file: " + pathTo + " succeeded. Will attempt rename again"); - } - } + } + } catch (IOException ioex) { throw new NestedIOException("Failed to delete file " + pathTo, ioex); } try { // attempt to rename again this.channel.rename(pathFrom, pathTo); - } + } catch (SftpException sftpex2) { throw new NestedIOException("failed to rename from " + pathFrom + " to " + pathTo, sftpex2); } @@ -169,8 +211,9 @@ class SftpSession implements Session { } } + @Override public boolean mkdir(String remoteDirectory) throws IOException { - try { + try { this.channel.mkdir(remoteDirectory); } catch (SftpException e) { @@ -179,6 +222,7 @@ class SftpSession implements Session { return true; } + @Override public boolean exists(String path) { try { this.channel.lstat(path); @@ -189,13 +233,13 @@ class SftpSession implements Session { } return false; } - + void connect() { try { if (!this.jschSession.isConnected()) { this.jschSession.connect(); - this.channel = (ChannelSftp) this.jschSession.openChannel("sftp"); } + this.channel = (ChannelSftp) this.jschSession.openChannel("sftp"); if (this.channel != null && !this.channel.isConnected()) { this.channel.connect(); } @@ -204,4 +248,5 @@ class SftpSession implements Session { throw new IllegalStateException("failed to connect", e); } } + } diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpOutboundTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpOutboundTests.java index 686f7971ac..6b920320d2 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpOutboundTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpOutboundTests.java @@ -17,15 +17,21 @@ package org.springframework.integration.sftp.outbound; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.mockito.Matchers.anyString; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.io.File; import java.io.FileOutputStream; import java.io.InputStream; +import java.lang.reflect.Constructor; import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -36,6 +42,7 @@ import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanFactory; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; @@ -46,16 +53,20 @@ import org.springframework.messaging.PollableChannel; import org.springframework.integration.file.DefaultFileNameGenerator; import org.springframework.integration.file.remote.FileInfo; import org.springframework.integration.file.remote.handler.FileTransferringMessageHandler; +import org.springframework.integration.file.remote.session.CachingSessionFactory; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.messaging.support.GenericMessage; import org.springframework.integration.sftp.session.DefaultSftpSessionFactory; import org.springframework.integration.sftp.session.SftpTestSessionFactory; import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.test.util.TestUtils; import org.springframework.util.FileCopyUtils; import com.jcraft.jsch.ChannelSftp; import com.jcraft.jsch.ChannelSftp.LsEntry; +import com.jcraft.jsch.JSch; +import com.jcraft.jsch.JSchException; import com.jcraft.jsch.SftpATTRS; /** @@ -194,6 +205,7 @@ public class SftpOutboundTests { handler.afterPropertiesSet(); final List madeDirs = new ArrayList(); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { madeDirs.add((String) invocation.getArguments()[0]); return null; @@ -206,6 +218,121 @@ public class SftpOutboundTests { assertEquals("/foo/bar/baz", madeDirs.get(2)); } + @Test + public void testSharedSession() throws Exception { + JSch jsch = spy(new JSch()); + Constructor ctor = com.jcraft.jsch.Session.class.getDeclaredConstructor(JSch.class); + ctor.setAccessible(true); + com.jcraft.jsch.Session jschSession1 = spy(ctor.newInstance(jsch)); + com.jcraft.jsch.Session jschSession2 = spy(ctor.newInstance(jsch)); + new DirectFieldAccessor(jschSession1).setPropertyValue("isConnected", true); + new DirectFieldAccessor(jschSession2).setPropertyValue("isConnected", true); + when(jsch.getSession("foo", "host", 22)).thenReturn(jschSession1, jschSession2); + ChannelSftp channel1 = spy(new ChannelSftp()); + ChannelSftp channel2 = spy(new ChannelSftp()); + new DirectFieldAccessor(channel1).setPropertyValue("session", jschSession1); + new DirectFieldAccessor(channel2).setPropertyValue("session", jschSession1); + when(jschSession1.openChannel("sftp")).thenReturn(channel1, channel2); + DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(jsch, true); + factory.setHost("host"); + factory.setUser("foo"); + factory.setPassword("bar"); + noopConnect(channel1); + noopConnect(channel2); + Session s1 = factory.getSession(); + Session s2 = factory.getSession(); + assertSame(TestUtils.getPropertyValue(s1, "jschSession"), TestUtils.getPropertyValue(s2, "jschSession")); + } + + @Test + public void testNotSharedSession() throws Exception { + JSch jsch = spy(new JSch()); + Constructor ctor = com.jcraft.jsch.Session.class.getDeclaredConstructor(JSch.class); + ctor.setAccessible(true); + com.jcraft.jsch.Session jschSession1 = spy(ctor.newInstance(jsch)); + com.jcraft.jsch.Session jschSession2 = spy(ctor.newInstance(jsch)); + new DirectFieldAccessor(jschSession1).setPropertyValue("isConnected", true); + new DirectFieldAccessor(jschSession2).setPropertyValue("isConnected", true); + when(jsch.getSession("foo", "host", 22)).thenReturn(jschSession1, jschSession2); + ChannelSftp channel1 = spy(new ChannelSftp()); + ChannelSftp channel2 = spy(new ChannelSftp()); + new DirectFieldAccessor(channel1).setPropertyValue("session", jschSession1); + new DirectFieldAccessor(channel2).setPropertyValue("session", jschSession1); + when(jschSession1.openChannel("sftp")).thenReturn(channel1); + when(jschSession2.openChannel("sftp")).thenReturn(channel2); + DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(jsch, false); + factory.setHost("host"); + factory.setUser("foo"); + factory.setPassword("bar"); + noopConnect(channel1); + noopConnect(channel2); + Session s1 = factory.getSession(); + Session s2 = factory.getSession(); + assertNotSame(TestUtils.getPropertyValue(s1, "jschSession"), TestUtils.getPropertyValue(s2, "jschSession")); + } + + @Test + public void testSharedSessionCachedReset() throws Exception { + JSch jsch = spy(new JSch()); + Constructor ctor = com.jcraft.jsch.Session.class.getDeclaredConstructor(JSch.class); + ctor.setAccessible(true); + com.jcraft.jsch.Session jschSession1 = spy(ctor.newInstance(jsch)); + com.jcraft.jsch.Session jschSession2 = spy(ctor.newInstance(jsch)); + new DirectFieldAccessor(jschSession1).setPropertyValue("isConnected", true); + new DirectFieldAccessor(jschSession2).setPropertyValue("isConnected", true); + when(jsch.getSession("foo", "host", 22)).thenReturn(jschSession1, jschSession2); + ChannelSftp channel1 = spy(new ChannelSftp()); + ChannelSftp channel2 = spy(new ChannelSftp()); + ChannelSftp channel3 = spy(new ChannelSftp()); + ChannelSftp channel4 = spy(new ChannelSftp()); + new DirectFieldAccessor(channel1).setPropertyValue("session", jschSession1); + new DirectFieldAccessor(channel2).setPropertyValue("session", jschSession1); + when(jschSession1.openChannel("sftp")).thenReturn(channel1, channel2); + when(jschSession2.openChannel("sftp")).thenReturn(channel3, channel4); + DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(jsch, true); + factory.setHost("host"); + factory.setUser("foo"); + factory.setPassword("bar"); + CachingSessionFactory cachedFactory = new CachingSessionFactory(factory); + noopConnect(channel1); + noopConnect(channel2); + noopConnect(channel3); + noopConnect(channel4); + Session s1 = cachedFactory.getSession(); + Session s2 = cachedFactory.getSession(); + assertSame(jschSession1, TestUtils.getPropertyValue(s2, "targetSession.jschSession")); + assertSame(TestUtils.getPropertyValue(s1, "targetSession.jschSession"), TestUtils.getPropertyValue(s2, "targetSession.jschSession")); + s1.close(); + Session s3 = cachedFactory.getSession(); + assertSame(TestUtils.getPropertyValue(s1, "targetSession"), TestUtils.getPropertyValue(s3, "targetSession")); + s3.close(); + cachedFactory.resetCache(); + verify(jschSession1, never()).disconnect(); + s3 = cachedFactory.getSession(); + assertSame(jschSession2, TestUtils.getPropertyValue(s3, "targetSession.jschSession")); + assertNotSame(TestUtils.getPropertyValue(s1, "targetSession"), TestUtils.getPropertyValue(s3, "targetSession")); + s2.close(); + verify(jschSession1).disconnect(); + s2 = cachedFactory.getSession(); + assertSame(jschSession2, TestUtils.getPropertyValue(s2, "targetSession.jschSession")); + assertNotSame(TestUtils.getPropertyValue(s3, "targetSession"), TestUtils.getPropertyValue(s2, "targetSession")); + s2.close(); + s3.close(); + verify(jschSession2, never()).disconnect(); + cachedFactory.resetCache(); + verify(jschSession2).disconnect(); + } + + private void noopConnect(ChannelSftp channel1) throws JSchException { + doAnswer(new Answer() { + + @Override + public Object answer(InvocationOnMock invocation) throws Throwable { + return null; + } + }).when(channel1).connect(); + } + public static class TestSftpSessionFactory extends DefaultSftpSessionFactory { @Override @@ -214,6 +341,7 @@ public class SftpOutboundTests { ChannelSftp channel = mock(ChannelSftp.class); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { File file = new File((String)invocation.getArguments()[1]); @@ -225,6 +353,7 @@ public class SftpOutboundTests { }).when(channel).put(Mockito.any(InputStream.class), Mockito.anyString()); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { File file = new File((String) invocation.getArguments()[0]); diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests-context.xml b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests-context.xml index da0b9d7d55..cf054cdccd 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests-context.xml +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests-context.xml @@ -77,4 +77,19 @@ + + + + + + + + + + + + + diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java index d807208c27..8c78d5cec4 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java @@ -18,14 +18,22 @@ package org.springframework.integration.sftp.outbound; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; +import java.io.ByteArrayOutputStream; import java.io.File; import java.io.IOException; +import java.io.PipedInputStream; +import java.io.PipedOutputStream; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import org.hamcrest.Matchers; import org.junit.After; @@ -38,11 +46,13 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.util.FileCopyUtils; import com.jcraft.jsch.ChannelSftp.LsEntry; import com.jcraft.jsch.SftpATTRS; @@ -55,10 +65,10 @@ import com.jcraft.jsch.SftpATTRS; * 
  *  $ tree sftpSource/
  *  sftpSource/
- *  ├── sftpSource1.txt
- *  ├── sftpSource2.txt
+ *  ├── sftpSource1.txt - contains 'source1'
+ *  ├── sftpSource2.txt - contains 'source2'
  *  └── subSftpSource
- *      └── subSftpSource1.txt
+ *      └── subSftpSource1.txt - contains 'subSource1'
  *
* * @author Artem Bilan @@ -156,6 +166,11 @@ public class SftpServerOutboundTests { @Test public void testInt2866LocalDirectoryExpressionGET() { + Session session = null; + boolean sharedSession = "realSSHSharedSession".equals(System.getProperty("spring.profiles.active")); + if (sharedSession) { + session = this.sessionFactory.getSession(); + } String dir = "sftpSource/"; this.inboundGet.send(new GenericMessage(dir + "sftpSource1.txt")); Message result = this.output.receive(1000); @@ -171,6 +186,11 @@ public class SftpServerOutboundTests { localFile = (File) result.getPayload(); assertThat(localFile.getPath().replaceAll(java.util.regex.Matcher.quoteReplacement(File.separator), "/"), Matchers.containsString(dir.toUpperCase())); + if (sharedSession) { + Session session2 = this.sessionFactory.getSession(); + assertSame(TestUtils.getPropertyValue(session, "targetSession.jschSession"), + TestUtils.getPropertyValue(session2, "targetSession.jschSession")); + } } @Test @@ -251,4 +271,86 @@ public class SftpServerOutboundTests { } + /** + * Only runs with a real server (see class javadocs). + */ + @Test + public void testInt3100RawGET() throws Exception { + if (!sessionFactory.toString().startsWith("Mock for")) { + Session session = this.sessionFactory.getSession(); + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + FileCopyUtils.copy(session.readRaw("sftpSource/sftpSource1.txt"), baos); + assertTrue(session.finalizeRaw()); + assertEquals("source1", new String(baos.toByteArray())); + + baos = new ByteArrayOutputStream(); + FileCopyUtils.copy(session.readRaw("sftpSource/sftpSource2.txt"), baos); + assertTrue(session.finalizeRaw()); + assertEquals("source2", new String(baos.toByteArray())); + + session.close(); + } + } + + @Test + public void testInt3047ConcurrentSharedSession() throws Exception { + if ("realSSHSharedSession".equals(System.getProperty("spring.profiles.active"))) { + final Session session1 = this.sessionFactory.getSession(); + final Session session2 = this.sessionFactory.getSession(); + final PipedInputStream pipe1 = new PipedInputStream(); + PipedOutputStream out1 = new PipedOutputStream(pipe1); + final PipedInputStream pipe2 = new PipedInputStream(); + PipedOutputStream out2 = new PipedOutputStream(pipe2); + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + Executors.newSingleThreadExecutor().execute(new Runnable() { + + @Override + public void run() { + try { + session1.write(pipe1, "foo.txt"); + } + catch (IOException e) { + e.printStackTrace(); + } + latch1.countDown(); + } + }); + Executors.newSingleThreadExecutor().execute(new Runnable() { + + @Override + public void run() { + try { + session2.write(pipe2, "bar.txt"); + } + catch (IOException e) { + e.printStackTrace(); + } + latch2.countDown(); + } + }); + + out1.write('a'); + out2.write('b'); + out1.write('c'); + out2.write('d'); + out1.write('e'); + out2.write('f'); + out1.close(); + out2.close(); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + assertTrue(latch2.await(10, TimeUnit.SECONDS)); + ByteArrayOutputStream bos1 = new ByteArrayOutputStream(); + ByteArrayOutputStream bos2 = new ByteArrayOutputStream(); + session1.read("foo.txt", bos1); + session2.read("bar.txt", bos2); + assertEquals("ace", new String(bos1.toByteArray())); + assertEquals("bdf", new String(bos2.toByteArray())); + session1.remove("foo.txt"); + session2.remove("bar.txt"); + session1.close(); + session2.close(); + } + } + } diff --git a/src/reference/docbook/ftp.xml b/src/reference/docbook/ftp.xml index 1e04a9960f..42747dd15d 100644 --- a/src/reference/docbook/ftp.xml +++ b/src/reference/docbook/ftp.xml @@ -154,6 +154,7 @@ protected void postProcessClientBeforeConnect(T client) throws IOException { filename-pattern="*.txt" remote-directory="some/remote/path" remote-file-separator="/" + preserve-timestamp="true" local-filename-generator-expression="#this.toUpperCase() + '.a'" local-filter="myFilter" local-directory="."> @@ -172,7 +173,11 @@ protected void postProcessClientBeforeConnect(T client) throws IOException { that's what it ultimately generates with the transferred file as its payload. So, the root object of the SpEL Evaluation Context is the original name of the remote file (String). - + + Starting with Spring Integration 3.0, you can specify the preserve-timestamp + attribute (default false); when true, the local file's modified timestamp will be set to the value + retrieved from the server; otherwise it will be set to the current time. + Sometimes file filtering based on the simple pattern specified via filename-pattern attribute might not be sufficient. If this is the case, you can use the filename-regex attribute to specify a Regular Expression @@ -500,8 +505,8 @@ protected void postProcessClientBeforeConnect(T client) throws IOException {
FTP Session Caching - Starting with version 3.0, sessions are no longer cached by default; the cache-sessions attribute - is no longer supported on endpoints. You must now use a CachingSessionFactory (see below) if you + Starting with Spring Integration version 3.0, sessions are no longer cached by default; the cache-sessions attribute + is no longer supported on endpoints. You must use a CachingSessionFactory (see below) if you wish to cache sessions. @@ -532,5 +537,10 @@ protected void postProcessClientBeforeConnect(T client) throws IOException { sessionCacheSize set to 10 and the sessionWaitTimeout set to 1 second (its value is in millliseconds). + + Starting with Spring Integration version 3.0, the CachingConnectionFactory + provides a resetCache() method. When invoked, all idle sessions are immediately closed and in-use + sessions are closed when they are returned to the cache. New requests for sessions will establish new sessions as necessary. +
diff --git a/src/reference/docbook/jms.xml b/src/reference/docbook/jms.xml index 9e482ffe38..f8b8c5169d 100644 --- a/src/reference/docbook/jms.xml +++ b/src/reference/docbook/jms.xml @@ -87,12 +87,21 @@ message-driven Channel Adapter with a Destination reference. ]]> + The Message-Driven adapter also accepts several properties that pertain to the MessageListener container. - These values are only considered if you do not provide an actual 'container' reference. In that case, + These values are only considered if you do not provide a container reference. In that case, an instance of DefaultMessageListenerContainer will be created and configured based on these properties. For example, you can specify the "transaction-manager" reference, the "concurrent-consumers" value, and several other property references and values. Refer to the JavaDoc and Spring Integration's JMS Schema - (spring-integration-jms.xsd) for more detail. + (spring-integration-jms.xsd) for more details. + + + If you have a custom listener container implementation (usually a subclass of + DefaultMessageListenerContainer), you can either provide a reference to an instance + of it using the container attribute, or simply provide its fully qualified class name using + the container-class attribute. In that case, the attributes on the adapter + are transferred to an instance of your custom container. + diff --git a/src/reference/docbook/redis.xml b/src/reference/docbook/redis.xml index 1bd7c2023a..e3c71c2be7 100644 --- a/src/reference/docbook/redis.xml +++ b/src/reference/docbook/redis.xml @@ -150,8 +150,15 @@ rt.setConnectionFactory(redisConnectionFactory);]]> Redis Messages and the Spring Integration Message payloads. The default is a SimpleMessageConverter. - Inbound adapters can subscribe to multiple topic names hence the comma-delimited set of values in the - topics attribute. + + Inbound adapters can subscribe to multiple topic names hence the comma-delimited set of values in the + topics attribute. + + + Since Spring Integration 3.0, the Inbound Adapter, in addition to the existing topics attribute, + now has the topic-patterns attribute. This attribute contains a comma-delimited set of Redis topic patterns. + For more information regarding Redis publish/subscribe, see Redis Pub/Sub. + Inbound adapters can use a RedisSerializer to deserialize the body of Redis Messages. The serializer attribute of the <int-redis:inbound-channel-adapter> can be set to an @@ -205,6 +212,7 @@ rt.setConnectionFactory(redisConnectionFactory);]]> error-channel="" ]]> ]]> @@ -247,7 +255,9 @@ rt.setConnectionFactory(redisConnectionFactory);]]> The MessageChannel to which to send ErrorMessages with - Exceptions from the listening task of the Endpoint. + Exceptions from the listening task of the Endpoint. By default + the underlying MessagePublishingErrorHandler uses the + default errorChannel from the application context. @@ -262,6 +272,12 @@ rt.setConnectionFactory(redisConnectionFactory);]]> The timeout in milliseconds for 'right pop' operation to wait for a Redis message from the queue. Default is 1 second. + + + The time in milliseconds for which the listener task should sleep after exceptions on the 'right pop' operation, + before restarting the listener task. + + Specify if this Endpoint expects data from the Redis queue to contain entire Messages. @@ -344,6 +360,22 @@ rt.setConnectionFactory(redisConnectionFactory);]]> +
+ Redis Application Events + + Since Spring Integration 3.0, the Redis module provides an implementation + of IntegrationEvent - which, in turn, is a + org.springframework.context.ApplicationEvent. The RedisExceptionEvent + encapsulates an Exceptions from Redis operations (with the Endpoint being the source + of the event). For example, the <int-redis:queue-inbound-channel-adapter/> + emits those events after catching Exceptions from the BoundListOperations.rightPop + operation. + The exception may be any generic org.springframework.data.redis.RedisSystemException or + a org.springframework.data.redis.RedisConnectionFailureException. + Handling these events using an <int-event:inbound-channel-adapter/> can be useful to determine + problems with background Redis tasks and to take administrative actions. + +
diff --git a/src/reference/docbook/sftp.xml b/src/reference/docbook/sftp.xml index a73cf9c6d6..5111c75d5f 100644 --- a/src/reference/docbook/sftp.xml +++ b/src/reference/docbook/sftp.xml @@ -59,6 +59,27 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp However, Spring Integration also supports the caching of SFTP sessions, please see for more information. + + + JSch supports multiple channels (operations) over a connection to the server. By default, + the Spring Integration session factory uses a separate physical connection for each channel. + Since Spring Integration 3.0, you can configure the session factory + (using a boolean constructor arg - default false) to use a single connection + to the server and create multiple JSch channels on that single connection. + + + When using this feature, you must wrap the session factory in a caching session + factory, as described below, so that the connection is not physically closed when + an operation completes. + + + If the cache is reset, the session is disconnected only when the last channel is closed. + + + The connection will be refreshed if it is found to be disconnected when a new operation + obtains a session. + + If you experience connectivity problems and would like to trace Session creation as well as see which Sessions are polled you may enable it by @@ -79,6 +100,11 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp Below you will find all properties that are exposed by the DefaultSftpSessionFactory. + isSharedSession (constructor argument) + + When true, a single connection will be used and JSch Channels will be multiplexed. + Defaults to false. + clientVersion Allows you to set the client version property. It's default @@ -183,8 +209,8 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp
SFTP Session Caching - Starting with version 3.0, sessions are no longer cached by default; the cache-sessions attribute - is no longer supported on endpoints. You must now use a CachingSessionFactory (see below) if you + Starting with Spring Integration version 3.0, sessions are no longer cached by default; the cache-sessions attribute + is no longer supported on endpoints. You must use a CachingSessionFactory (see below) if you wish to cache sessions. @@ -217,6 +243,13 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp sessionCacheSize set to 10 and the sessionWaitTimeout set to 1 second (its value is in millliseconds). + + Starting with Spring Integration version 3.0, the CachingConnectionFactory + provides a resetCache() method. When invoked, all idle sessions are immediately closed and in-use + sessions are closed when they are returned to the cache. When using isSharedSession=true, the channel is + closed, and the shared session is closed only when the last channel is closed. + New requests for sessions will establish new sessions as necessary. +
@@ -230,6 +263,7 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp channel="requestChannel" filename-pattern="*.txt" remote-directory="/foo/bar" + preserve-timestamp="true" local-directory="file:target/foo" auto-create-local-directory="true" local-filename-generator-expression="#this.toUpperCase() + '.a'" @@ -252,6 +286,11 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp that's what it ultimately generates with the transferred file as its payload. So, the root object of the SpEL Evaluation Context is the original name of the remote file (String). + + Starting with Spring Integration 3.0, you can specify the preserve-timestamp + attribute (default false); when true, the local file's modified timestamp will be set to the value + retrieved from the server; otherwise it will be set to the current time. + Sometimes file filtering based on the simple pattern specified via filename-pattern attribute might not be sufficient. If this is the case, you can use the filename-regex attribute to specify a Regular Expression diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 3a6935d225..c1271ca497 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -243,6 +243,16 @@ For more information, see and . + + The CachingConnectionFactory now provides a new method + resetCache(). This immediately closes idle sessions and causes in-use + sessions to be closed as and when they are returned to the cache. + + + The DefaultSftpSessionFactory (in conjunction with a + CachingSessionFactory) now supports multiplexing channels over + a single SSH connection (SFTP Only). +
FTP, SFTP and FTPS Inbound Adapters @@ -584,6 +594,10 @@ The Redis Outbound Channel Adapter now has the topic-expression property to determine the Redis topic against the Message at runtime. + + The Redis Inbound Channel Adapter, in addition to the existing topics attribute, + now has the topic-patterns attribute. +