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..77c1363c8f 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 @@ -65,14 +65,17 @@ 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(); } @@ -100,6 +103,7 @@ 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()); } @@ -107,6 +111,7 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe /** * Remove (close) any unused sessions in the pool. */ + @Override public void destroy() { this.pool.removeAllIdleItems(); } @@ -122,6 +127,7 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe this.targetSession = targetSession; } + @Override public synchronized void close() { if (released) { if (logger.isDebugEnabled()){ @@ -137,41 +143,61 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe } } + @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/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 16a6171622..e460b71359 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.integration.support.MessageBuilder; /** * @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); @@ -922,4 +1108,4 @@ class TestPatternFilter extends AbstractSimplePatternFileListFilter return file.getFilename(); } -} \ No newline at end of file +} 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 a5c9275949..04d1f849e0 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; @@ -33,10 +35,12 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.Message; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.core.PollableChannel; +import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.ftp.TesFtpServer; import org.springframework.integration.message.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-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..a3d47250af 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,6 +25,7 @@ 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.util.Assert; @@ -61,6 +62,7 @@ class SftpSession implements Session { } + @Override public boolean remove(String path) throws IOException { Assert.state(this.channel != null, "session is not connected"); try { @@ -72,6 +74,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 +95,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 +111,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 +119,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 +149,41 @@ class SftpSession implements Session { } } + @Override public void close() { if (this.jschSession.isConnected()) { this.jschSession.disconnect(); } } + @Override public boolean isOpen() { return 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 +193,9 @@ class SftpSession implements Session { } } + @Override public boolean mkdir(String remoteDirectory) throws IOException { - try { + try { this.channel.mkdir(remoteDirectory); } catch (SftpException e) { @@ -179,6 +204,7 @@ class SftpSession implements Session { return true; } + @Override public boolean exists(String path) { try { this.channel.lstat(path); @@ -189,7 +215,7 @@ class SftpSession implements Session { } return false; } - + void connect() { try { if (!this.jschSession.isConnected()) { 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 dcccb049c5..3e5d978499 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 @@ -19,10 +19,12 @@ package org.springframework.integration.sftp.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 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.util.List; @@ -44,6 +46,7 @@ import org.springframework.integration.message.GenericMessage; import org.springframework.integration.sftp.session.SftpFileInfo; 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; @@ -56,10 +59,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 @@ -252,4 +255,26 @@ 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(); + } + } + + }