From ccb669ca750e0da7885b934b7dc8cc1f1e0cd575 Mon Sep 17 00:00:00 2001 From: ialbert Date: Tue, 25 Jul 2023 18:29:58 +0300 Subject: [PATCH] added SFTP to swt-importer --- clearing-parent/swt-importer/pom.xml | 4 + .../swt/importer/config/SFTPConfig.java | 81 +++++++++++++++++++ .../swt/importer/config/settings/Store.java | 46 +++++++++++ .../swt/importer/services/FileChecker.java | 9 ++- .../src/main/resources/application.properties | 7 ++ 5 files changed, 146 insertions(+), 1 deletion(-) create mode 100644 clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/SFTPConfig.java diff --git a/clearing-parent/swt-importer/pom.xml b/clearing-parent/swt-importer/pom.xml index ec4e4f967..a75ecaefa 100644 --- a/clearing-parent/swt-importer/pom.xml +++ b/clearing-parent/swt-importer/pom.xml @@ -30,6 +30,10 @@ org.springframework.boot spring-boot-autoconfigure + + org.springframework.integration + spring-integration-sftp + diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/SFTPConfig.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/SFTPConfig.java new file mode 100644 index 000000000..e0f46f154 --- /dev/null +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/SFTPConfig.java @@ -0,0 +1,81 @@ +package ru.spcex.clearing.swt.importer.config; + +import com.jcraft.jsch.ChannelSftp; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.annotation.InboundChannelAdapter; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.core.MessageSource; +import org.springframework.integration.file.filters.AcceptAllFileListFilter; +import org.springframework.integration.file.remote.session.CachingSessionFactory; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.scheduling.PollerMetadata; +import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizer; +import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizingMessageSource; +import org.springframework.integration.sftp.session.DefaultSftpSessionFactory; +import org.springframework.messaging.MessageHandler; +import org.springframework.scheduling.support.PeriodicTrigger; +import ru.spcex.clearing.swt.importer.config.settings.ImportSWTServiceSettings; + +import java.io.File; +import java.util.concurrent.TimeUnit; + +@Configuration +public class SFTPConfig { + private final Logger log = LoggerFactory.getLogger(getClass()); + private final ImportSWTServiceSettings settings; + + @Autowired + public SFTPConfig(ImportSWTServiceSettings settings) { + this.settings = settings; + } + + @Bean + public SessionFactory sftpSessionFactory() { + DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true); + factory.setHost(settings.getStore().getServerIp()); + factory.setPort(settings.getStore().getServerPort()); + factory.setUser(settings.getStore().getUser()); + factory.setPassword(settings.getStore().getPassword()); + factory.setAllowUnknownKeys(true); + return new CachingSessionFactory<>(factory); + } + + @Bean(name = PollerMetadata.DEFAULT_POLLER) + public PollerMetadata defaultPoller() { + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setTrigger(new PeriodicTrigger(5, TimeUnit.SECONDS)); + return pollerMetadata; + } + + @Bean + SftpInboundFileSynchronizer sftpInboundFileSynchronizer() { + SftpInboundFileSynchronizer fileSync = new SftpInboundFileSynchronizer(sftpSessionFactory()); + fileSync.setDeleteRemoteFiles(true); + fileSync.setTemporaryFileSuffix(".tmp"); + fileSync.setRemoteDirectory(settings.getStore().getSftpSrcDir()); + fileSync.setFilter(new AcceptAllFileListFilter<>()); + return fileSync; + } + + @Bean + @InboundChannelAdapter("sftpChannel") + public MessageSource sftpMessageSource() { + SftpInboundFileSynchronizingMessageSource source = new SftpInboundFileSynchronizingMessageSource(sftpInboundFileSynchronizer()); + source.setLocalDirectory(new File(settings.getStore().getSrcDir())); + source.setAutoCreateLocalDirectory(true); + return source; + } + + @Bean + @ServiceActivator(inputChannel="sftpChannel") + MessageHandler messageHandler() { + return arg0 -> { + File f = (File) arg0.getPayload(); + log.info("copy file '{}' from sftp event", f.getName()); + }; + } +} diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/settings/Store.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/settings/Store.java index f74a91998..af506b8e1 100644 --- a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/settings/Store.java +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/config/settings/Store.java @@ -6,6 +6,12 @@ public class Store { private String outDir; private String outDirError; private boolean deleteSrcFiles = true; + //sftp settings + private String sftpSrcDir; + private String user; + private String password; + private String serverIp; + private int serverPort; public String getSrcDir() { return srcDir; @@ -38,4 +44,44 @@ public class Store { public void setOutDirError(String outDirError) { this.outDirError = outDirError; } + + public String getSftpSrcDir() { + return sftpSrcDir; + } + + public void setSftpSrcDir(String sftpSrcDir) { + this.sftpSrcDir = sftpSrcDir; + } + + public String getUser() { + return user; + } + + public void setUser(String user) { + this.user = user; + } + + public String getPassword() { + return password; + } + + public void setPassword(String password) { + this.password = password; + } + + public String getServerIp() { + return serverIp; + } + + public void setServerIp(String serverIp) { + this.serverIp = serverIp; + } + + public int getServerPort() { + return serverPort; + } + + public void setServerPort(int serverPort) { + this.serverPort = serverPort; + } } diff --git a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/services/FileChecker.java b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/services/FileChecker.java index f0eb076ed..f8287590d 100644 --- a/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/services/FileChecker.java +++ b/clearing-parent/swt-importer/src/main/java/ru/spcex/clearing/swt/importer/services/FileChecker.java @@ -42,7 +42,14 @@ public class FileChecker { File swtDir = new File(swtDirPath); File[] swtFiles = swtDir.listFiles((dir, name) -> { String offsetName = name.length() > 10 ? name.substring(0, 9) : ""; - return offsetName.equalsIgnoreCase("RDC_KS_DF"); + String extension; + int formatPosition = name.lastIndexOf("."); + if (formatPosition == -1 || formatPosition == name.length() - 1) { + extension = null; + } else { + extension = name.substring(formatPosition + 1); + } + return offsetName.equalsIgnoreCase("RDC_KS_DF") && !("tmp".equalsIgnoreCase(extension)); }); List resultFiles = new LinkedList<>(); diff --git a/clearing-parent/swt-importer/src/main/resources/application.properties b/clearing-parent/swt-importer/src/main/resources/application.properties index debb4a56a..a12532494 100644 --- a/clearing-parent/swt-importer/src/main/resources/application.properties +++ b/clearing-parent/swt-importer/src/main/resources/application.properties @@ -9,6 +9,13 @@ import-swt-service.store.src-dir=clearing-parent/swt-importer/src/test/java/ru/s import-swt-service.store.out-dir=clearing-parent/swt-importer/src/test/java/ru/spcex/clearing/swt/importer/files/importer/loaded/ import-swt-service.store.out-dir-error=clearing-parent/swt-importer/src/test/java/ru/spcex/clearing/swt/importer/files/importer/error/ +#sftpSrcDir +import-swt-service.store.sftp-src-dir=/path/to/sftp/folder +import-swt-service.store.user=user +import-swt-service.store.password=********* +import-swt-service.store.server-ip=127.0.0.1 +import-swt-service.store.server-port=22 + import-swt-service.common.encoding-source=cp866 import-swt-service.common.insert-batch-size=100 import-swt-service.common.threads-count=10