INT-1614 naming changes for synchronizers and the interface's strategy method itself
This commit is contained in:
@@ -23,7 +23,6 @@ import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.core.io.Resource;
|
||||
import org.springframework.integration.file.filters.FileListFilter;
|
||||
|
||||
/**
|
||||
@@ -36,7 +35,7 @@ import org.springframework.integration.file.filters.FileListFilter;
|
||||
*
|
||||
* @author Josh Long
|
||||
*/
|
||||
public abstract class AbstractInboundRemoteFileSystemSychronizer<F> implements InitializingBean {
|
||||
public abstract class AbstractInboundFileSynchronizer<F> implements InboundFileSynchronizer, InitializingBean {
|
||||
|
||||
protected final Log logger = LogFactory.getLog(this.getClass());
|
||||
|
||||
@@ -94,11 +93,6 @@ public abstract class AbstractInboundRemoteFileSystemSychronizer<F> implements I
|
||||
return (this.filter != null) ? this.filter.filterFiles(files) : Arrays.asList(files);
|
||||
}
|
||||
|
||||
/**
|
||||
* This is the callback where the subclasses must synchronize.
|
||||
*/
|
||||
protected abstract void syncRemoteToLocalFileSystem(Resource localDirectory);
|
||||
|
||||
|
||||
/**
|
||||
* Strategy interface to expose a hook for dispatching, moving, or deleting
|
||||
@@ -47,14 +47,13 @@ import org.springframework.util.Assert;
|
||||
* specific).
|
||||
* <p/>
|
||||
* This class is to be used as a pair with an implementation of
|
||||
* {@link AbstractInboundRemoteFileSystemSychronizer}. The synchronizer must
|
||||
* {@link AbstractInboundFileSynchronizer}. The synchronizer must
|
||||
* handle the work of actually connecting to the remote file system and
|
||||
* delivering new {@link File}s.
|
||||
*
|
||||
* @author Josh Long
|
||||
*/
|
||||
public abstract class AbstractInboundRemoteFileSystemSynchronizingMessageSource<F, S extends AbstractInboundRemoteFileSystemSychronizer<F>>
|
||||
extends MessageProducerSupport implements MessageSource<File> {
|
||||
public abstract class AbstractInboundFileSynchronizingMessageSource<F> extends MessageProducerSupport implements MessageSource<File> {
|
||||
|
||||
/**
|
||||
* Extension used when downloading files. We change it right after we know it's downloaded.
|
||||
@@ -71,7 +70,7 @@ public abstract class AbstractInboundRemoteFileSystemSynchronizingMessageSource<
|
||||
* An implementation that will handle the chores of actually connecting to and synching up
|
||||
* the remote file system with the local one, in an inbound direction.
|
||||
*/
|
||||
protected volatile S synchronizer;
|
||||
protected volatile AbstractInboundFileSynchronizer<F> synchronizer;
|
||||
|
||||
/**
|
||||
* Directory to which things should be synched locally.
|
||||
@@ -93,7 +92,7 @@ public abstract class AbstractInboundRemoteFileSystemSynchronizingMessageSource<
|
||||
this.autoCreateDirectories = autoCreateDirectories;
|
||||
}
|
||||
|
||||
public void setSynchronizer(S synchronizer) {
|
||||
public void setSynchronizer(AbstractInboundFileSynchronizer<F> synchronizer) {
|
||||
this.synchronizer = synchronizer;
|
||||
}
|
||||
|
||||
@@ -151,7 +150,7 @@ public abstract class AbstractInboundRemoteFileSystemSynchronizingMessageSource<
|
||||
Assert.state(this.synchronizer != null, "synchronizer must not be null");
|
||||
Message<File> message = this.fileSource.receive();
|
||||
if (message == null) {
|
||||
this.synchronizer.syncRemoteToLocalFileSystem(this.localDirectory);
|
||||
this.synchronizer.synchronizeToLocalDirectory(this.localDirectory);
|
||||
message = this.fileSource.receive();
|
||||
}
|
||||
return message;
|
||||
@@ -0,0 +1,31 @@
|
||||
/*
|
||||
* Copyright 2002-2010 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.synchronizer;
|
||||
|
||||
import org.springframework.core.io.Resource;
|
||||
|
||||
/**
|
||||
* Strategy for synchronizing from a remote File system to a local directory.
|
||||
*
|
||||
* @author Mark Fisher
|
||||
* @since 2.0
|
||||
*/
|
||||
public interface InboundFileSynchronizer {
|
||||
|
||||
void synchronizeToLocalDirectory(Resource localDirectory);
|
||||
|
||||
}
|
||||
@@ -29,8 +29,8 @@ import org.apache.commons.net.ftp.FTPFile;
|
||||
|
||||
import org.springframework.core.io.Resource;
|
||||
import org.springframework.integration.MessagingException;
|
||||
import org.springframework.integration.file.synchronizer.AbstractInboundRemoteFileSystemSychronizer;
|
||||
import org.springframework.integration.file.synchronizer.AbstractInboundRemoteFileSystemSynchronizingMessageSource;
|
||||
import org.springframework.integration.file.synchronizer.AbstractInboundFileSynchronizer;
|
||||
import org.springframework.integration.file.synchronizer.AbstractInboundFileSynchronizingMessageSource;
|
||||
import org.springframework.integration.ftp.client.FtpClientPool;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
@@ -40,7 +40,7 @@ import org.springframework.util.Assert;
|
||||
* @author Iwein Fuld
|
||||
* @author Josh Long
|
||||
*/
|
||||
public class FtpInboundRemoteFileSystemSynchronizer extends AbstractInboundRemoteFileSystemSychronizer<FTPFile> {
|
||||
public class FtpInboundRemoteFileSystemSynchronizer extends AbstractInboundFileSynchronizer<FTPFile> {
|
||||
|
||||
private volatile FtpClientPool clientPool;
|
||||
|
||||
@@ -54,15 +54,14 @@ public class FtpInboundRemoteFileSystemSynchronizer extends AbstractInboundRemot
|
||||
this.clientPool = clientPool;
|
||||
}
|
||||
|
||||
public void afterPropertiesSet() throws Exception {
|
||||
public void afterPropertiesSet() {
|
||||
Assert.notNull(this.clientPool, "clientPool must not be null");
|
||||
if (this.shouldDeleteSourceFile) {
|
||||
this.setEntryAcknowledgmentStrategy(new DeletionEntryAcknowledgmentStrategy());
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void syncRemoteToLocalFileSystem(Resource localDirectory) {
|
||||
public void synchronizeToLocalDirectory(Resource localDirectory) {
|
||||
try {
|
||||
FTPClient client = this.clientPool.getClient();
|
||||
Assert.state(client != null,
|
||||
@@ -93,7 +92,7 @@ public class FtpInboundRemoteFileSystemSynchronizer extends AbstractInboundRemot
|
||||
String localFileName = localDirectory.getFile().getPath() + "/" + remoteFileName;
|
||||
File localFile = new File(localFileName);
|
||||
if (!localFile.exists()) {
|
||||
String tempFileName = localFileName + AbstractInboundRemoteFileSystemSynchronizingMessageSource.INCOMPLETE_EXTENSION;
|
||||
String tempFileName = localFileName + AbstractInboundFileSynchronizingMessageSource.INCOMPLETE_EXTENSION;
|
||||
File file = new File(tempFileName);
|
||||
FileOutputStream fos = new FileOutputStream(file);
|
||||
try {
|
||||
@@ -122,7 +121,7 @@ public class FtpInboundRemoteFileSystemSynchronizer extends AbstractInboundRemot
|
||||
/**
|
||||
* An acknowledgment strategy that deletes the file.
|
||||
*/
|
||||
private static class DeletionEntryAcknowledgmentStrategy implements AbstractInboundRemoteFileSystemSychronizer.EntryAcknowledgmentStrategy<FTPFile> {
|
||||
private static class DeletionEntryAcknowledgmentStrategy implements EntryAcknowledgmentStrategy<FTPFile> {
|
||||
|
||||
private final Log logger = LogFactory.getLog(this.getClass());
|
||||
|
||||
|
||||
@@ -18,7 +18,8 @@ package org.springframework.integration.ftp.inbound;
|
||||
|
||||
import org.apache.commons.net.ftp.FTPFile;
|
||||
|
||||
import org.springframework.integration.file.synchronizer.AbstractInboundRemoteFileSystemSynchronizingMessageSource;
|
||||
import org.springframework.integration.file.synchronizer.AbstractInboundFileSynchronizer;
|
||||
import org.springframework.integration.file.synchronizer.AbstractInboundFileSynchronizingMessageSource;
|
||||
import org.springframework.integration.ftp.client.FtpClientPool;
|
||||
|
||||
/**
|
||||
@@ -27,8 +28,7 @@ import org.springframework.integration.ftp.client.FtpClientPool;
|
||||
* @author Iwein Fuld
|
||||
* @author Josh Long
|
||||
*/
|
||||
public class FtpInboundRemoteFileSystemSynchronizingMessageSource
|
||||
extends AbstractInboundRemoteFileSystemSynchronizingMessageSource<FTPFile, FtpInboundRemoteFileSystemSynchronizer> {
|
||||
public class FtpInboundRemoteFileSystemSynchronizingMessageSource extends AbstractInboundFileSynchronizingMessageSource<FTPFile> {
|
||||
|
||||
private volatile FtpClientPool clientPool;
|
||||
|
||||
@@ -41,10 +41,17 @@ public class FtpInboundRemoteFileSystemSynchronizingMessageSource
|
||||
return "ftp:inbound-channel-adapter";
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setSynchronizer(AbstractInboundFileSynchronizer<FTPFile> synchronizer) {
|
||||
super.setSynchronizer(synchronizer);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void onInit() {
|
||||
super.onInit();
|
||||
this.synchronizer.setClientPool(this.clientPool);
|
||||
if (this.synchronizer instanceof FtpInboundRemoteFileSystemSynchronizer) {
|
||||
((FtpInboundRemoteFileSystemSynchronizer) this.synchronizer).setClientPool(this.clientPool);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -13,6 +13,7 @@
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.integration.ftp.inbound;
|
||||
|
||||
import static org.mockito.Mockito.mock;
|
||||
@@ -71,7 +72,7 @@ public class FtpInboundRemoteFileSystemSynchronizerTest {
|
||||
syncronizer.afterPropertiesSet();
|
||||
|
||||
Resource localDirectory = new FileSystemResource(System.getProperty("java.io.tmpdir"));
|
||||
syncronizer.syncRemoteToLocalFileSystem(localDirectory);
|
||||
syncronizer.synchronizeToLocalDirectory(localDirectory);
|
||||
|
||||
verify(ftpClient, times(1)).retrieveFile(Mockito.anyString(), Mockito.any(OutputStream.class));
|
||||
verify(ftpClient, times(1)).deleteFile(Mockito.anyString());
|
||||
|
||||
@@ -25,8 +25,8 @@ import org.apache.commons.io.IOUtils;
|
||||
|
||||
import org.springframework.core.io.Resource;
|
||||
import org.springframework.integration.MessagingException;
|
||||
import org.springframework.integration.file.synchronizer.AbstractInboundRemoteFileSystemSychronizer;
|
||||
import org.springframework.integration.file.synchronizer.AbstractInboundRemoteFileSystemSynchronizingMessageSource;
|
||||
import org.springframework.integration.file.synchronizer.AbstractInboundFileSynchronizer;
|
||||
import org.springframework.integration.file.synchronizer.AbstractInboundFileSynchronizingMessageSource;
|
||||
import org.springframework.integration.sftp.session.SftpSession;
|
||||
import org.springframework.integration.sftp.session.SftpSessionFactory;
|
||||
import org.springframework.util.Assert;
|
||||
@@ -40,7 +40,7 @@ import com.jcraft.jsch.ChannelSftp;
|
||||
* @author Oleg Zhurakousky
|
||||
* @since 2.0
|
||||
*/
|
||||
public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSychronizer<ChannelSftp.LsEntry> {
|
||||
public class SftpInboundSynchronizer extends AbstractInboundFileSynchronizer<ChannelSftp.LsEntry> {
|
||||
|
||||
/**
|
||||
* the path on the remote mount
|
||||
@@ -107,9 +107,8 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
@SuppressWarnings("unchecked")
|
||||
protected void syncRemoteToLocalFileSystem(Resource localDirectory) {
|
||||
public void synchronizeToLocalDirectory(Resource localDirectory) {
|
||||
SftpSession session = null;
|
||||
try {
|
||||
session = this.sessionFactory.getSession();
|
||||
@@ -147,7 +146,7 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych
|
||||
FileOutputStream fileOutputStream = null;
|
||||
try {
|
||||
File tmpLocalTarget = new File(localFile.getAbsolutePath() +
|
||||
AbstractInboundRemoteFileSystemSynchronizingMessageSource.INCOMPLETE_EXTENSION);
|
||||
AbstractInboundFileSynchronizingMessageSource.INCOMPLETE_EXTENSION);
|
||||
fileOutputStream = new FileOutputStream(tmpLocalTarget);
|
||||
String remoteFqPath = this.remotePath + "/" + entry.getFilename();
|
||||
in = sftpSession.get(remoteFqPath);
|
||||
@@ -178,7 +177,7 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych
|
||||
}
|
||||
|
||||
|
||||
private class DeletionEntryAcknowledgmentStrategy implements AbstractInboundRemoteFileSystemSychronizer.EntryAcknowledgmentStrategy<ChannelSftp.LsEntry> {
|
||||
private class DeletionEntryAcknowledgmentStrategy implements AbstractInboundFileSynchronizer.EntryAcknowledgmentStrategy<ChannelSftp.LsEntry> {
|
||||
|
||||
public void acknowledge(Object useful, ChannelSftp.LsEntry msg) throws Exception {
|
||||
SftpSession sftpSession = (SftpSession) useful;
|
||||
|
||||
@@ -21,7 +21,7 @@ import java.util.regex.Pattern;
|
||||
|
||||
import org.springframework.integration.MessagingException;
|
||||
import org.springframework.integration.file.FileReadingMessageSource;
|
||||
import org.springframework.integration.file.synchronizer.AbstractInboundRemoteFileSystemSynchronizingMessageSource;
|
||||
import org.springframework.integration.file.synchronizer.AbstractInboundFileSynchronizingMessageSource;
|
||||
import org.springframework.integration.sftp.filters.SftpPatternMatchingFileListFilter;
|
||||
|
||||
import com.jcraft.jsch.ChannelSftp;
|
||||
@@ -33,8 +33,7 @@ import com.jcraft.jsch.ChannelSftp;
|
||||
* @author Oleg Zhurakousky
|
||||
* @since 2.0
|
||||
*/
|
||||
public class SftpInboundSynchronizingMessageSource
|
||||
extends AbstractInboundRemoteFileSystemSynchronizingMessageSource<ChannelSftp.LsEntry, SftpInboundSynchronizer> {
|
||||
public class SftpInboundSynchronizingMessageSource extends AbstractInboundFileSynchronizingMessageSource<ChannelSftp.LsEntry> {
|
||||
|
||||
private volatile Pattern filenamePattern;
|
||||
|
||||
@@ -68,7 +67,9 @@ public class SftpInboundSynchronizingMessageSource
|
||||
if (this.filenamePattern != null) {
|
||||
SftpPatternMatchingFileListFilter filter = new SftpPatternMatchingFileListFilter(this.filenamePattern);
|
||||
this.synchronizer.setFilter(filter);
|
||||
this.synchronizer.setAutoCreateDirectories(this.autoCreateDirectories);
|
||||
if (this.synchronizer instanceof SftpInboundSynchronizer) {
|
||||
((SftpInboundSynchronizer) this.synchronizer).setAutoCreateDirectories(this.autoCreateDirectories);
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (RuntimeException e) {
|
||||
|
||||
@@ -27,7 +27,7 @@ import org.junit.Test;
|
||||
|
||||
import org.springframework.core.io.FileSystemResource;
|
||||
import org.springframework.core.io.Resource;
|
||||
import org.springframework.integration.file.synchronizer.AbstractInboundRemoteFileSystemSychronizer.EntryAcknowledgmentStrategy;
|
||||
import org.springframework.integration.file.synchronizer.AbstractInboundFileSynchronizer.EntryAcknowledgmentStrategy;
|
||||
import org.springframework.integration.sftp.inbound.SftpInboundSynchronizer;
|
||||
import org.springframework.integration.sftp.session.SftpSession;
|
||||
import org.springframework.integration.sftp.session.SftpSessionFactory;
|
||||
|
||||
@@ -16,6 +16,7 @@
|
||||
|
||||
package org.springframework.integration.sftp.inbound;
|
||||
|
||||
import static org.mockito.Mockito.atLeast;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.times;
|
||||
import static org.mockito.Mockito.verify;
|
||||
@@ -23,11 +24,13 @@ import static org.mockito.Mockito.when;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.FileInputStream;
|
||||
import java.io.InputStream;
|
||||
import java.util.Vector;
|
||||
|
||||
import org.junit.Ignore;
|
||||
import org.junit.Test;
|
||||
import org.mockito.Mockito;
|
||||
import org.mockito.invocation.InvocationOnMock;
|
||||
import org.mockito.stubbing.Answer;
|
||||
|
||||
import org.springframework.core.io.FileSystemResource;
|
||||
import org.springframework.core.io.Resource;
|
||||
@@ -47,7 +50,6 @@ public class SftpInboundRemoteFileSystemSynchronizerTests {
|
||||
|
||||
@SuppressWarnings({ "rawtypes", "unchecked" })
|
||||
@Test
|
||||
@Ignore
|
||||
public void testCopyFileToLocalDir() throws Exception {
|
||||
File file = new File(System.getProperty("java.io.tmpdir") + "/foo.txt");
|
||||
if (file.exists()){
|
||||
@@ -65,7 +67,7 @@ public class SftpInboundRemoteFileSystemSynchronizerTests {
|
||||
SftpSession sftpSession = mock(SftpSession.class);
|
||||
|
||||
when(sessionFactory.getSession()).thenReturn(sftpSession);
|
||||
ChannelSftp channel = mock(ChannelSftp.class);
|
||||
final ChannelSftp channel = mock(ChannelSftp.class);
|
||||
when(channel.get((String) Mockito.any())).thenReturn(new FileInputStream(new File("template.mf")));
|
||||
Vector<LsEntry> entries = new Vector<ChannelSftp.LsEntry>();
|
||||
LsEntry entry = mock(LsEntry.class);
|
||||
@@ -77,14 +79,24 @@ public class SftpInboundRemoteFileSystemSynchronizerTests {
|
||||
entries.add(entry);
|
||||
when(channel.ls("foo/bar")).thenReturn(entries);
|
||||
when(filter.filterFiles((Object[]) Mockito.any())).thenReturn(entries);
|
||||
|
||||
when(sftpSession.get(Mockito.anyString())).thenAnswer(new Answer<InputStream>() {
|
||||
public InputStream answer(InvocationOnMock invocation)
|
||||
throws Throwable {
|
||||
String filePath = (String) invocation.getArguments()[0];
|
||||
return channel.get(filePath);
|
||||
}
|
||||
});
|
||||
|
||||
syncronizer.setShouldDeleteSourceFile(true);
|
||||
syncronizer.afterPropertiesSet();
|
||||
|
||||
Resource localDirectory = new FileSystemResource(System.getProperty("java.io.tmpdir"));
|
||||
syncronizer.syncRemoteToLocalFileSystem(localDirectory);
|
||||
syncronizer.synchronizeToLocalDirectory(localDirectory);
|
||||
|
||||
verify(sessionFactory, times(1)).getSession();
|
||||
verify(attr, atLeast(1)).isDir();
|
||||
// will add more validation, but for now this test is mainly to get the test coverage up
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user