treshenkov/clearing
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

http://jira.mfd.msk:8088/browse/CLS-619

Browse source 
Изменил импорт файлов по SFTP в dbf-importer.
This commit is contained in:
psemenkov 2024-04-15 17:09:58 +03:00
parent eb79ccf8d0
commit 4da055edad
7 changed files with 125 additions and 60 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
clearing-parent/dbf-importer/pom.xml
View file

@ -87,11 +87,11 @@
</dependency>
<!-- TEST -->
<!-- <dependency>-->
<!-- <groupId>org.springframework</groupId>-->
<!-- <artifactId>spring-test</artifactId>-->
<!-- <scope>test</scope>-->
<!-- </dependency>-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter</artifactId>

64
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/config/SFTPConfig.java
View file

@ -4,29 +4,27 @@ import com.jcraft.jsch.ChannelSftp;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.InboundChannelAdapter;
import org.springframework.integration.annotation.Gateway;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.file.filters.AcceptAllFileListFilter;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway;
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.gateway.SftpOutboundGateway;
import org.springframework.integration.sftp.session.DefaultSftpSessionFactory;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.scheduling.support.PeriodicTrigger;
import ru.spcex.clearing.dbf.importer.config.settings.ImportDBFServiceSettings;
import java.io.File;
import java.util.Comparator;
import java.util.concurrent.TimeUnit;
import java.util.List;
import static org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway.Command.MGET;
@Configuration
@ConditionalOnProperty(value="import-dbf-service.store.sftp-in.sftp-src-dir")
public class SFTPConfig {
private final Logger log = LoggerFactory.getLogger(getClass());
private final ImportDBFServiceSettings settings;
@ -47,40 +45,26 @@ public class SFTPConfig {
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;
@MessagingGateway
public interface DbfGateway {
@Gateway(requestChannel = "listSftpChannel")
List<File> listFiles(String dir);
}
@Bean
SftpInboundFileSynchronizer sftpInboundFileSynchronizer() {
SftpInboundFileSynchronizer fileSync = new SftpInboundFileSynchronizer(sftpSessionFactory());
fileSync.setDeleteRemoteFiles(true);
fileSync.setTemporaryFileSuffix(".tmp");
fileSync.setRemoteDirectory(settings.getStore().getSftpIn().getSftpSrcDir());
fileSync.setFilter(new AcceptAllFileListFilter<>());
fileSync.setPreserveTimestamp(true);
fileSync.setComparator(Comparator.comparingInt(f -> f.getAttrs().getMTime()));
return fileSync;
public MessageChannel listSftpChannel(SessionFactory<ChannelSftp.LsEntry> sessionFactory, ImportDBFServiceSettings settings) {
DirectChannel dc = new DirectChannel();
dc.subscribe(handlerList(sessionFactory, settings));
return dc;
}
@Bean
@InboundChannelAdapter("sftpChannel")
public MessageSource<File> 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());
};
@ServiceActivator(inputChannel = "listSftpChannel")
public MessageHandler handlerList(SessionFactory<ChannelSftp.LsEntry> sessionFactory, ImportDBFServiceSettings settings) {
SftpOutboundGateway sftpOutboundGateway = new SftpOutboundGateway(sessionFactory, MGET.getCommand(), null);
sftpOutboundGateway.setLocalDirectory(new File(settings.getStore().getSrcDir()));
sftpOutboundGateway.setAutoCreateLocalDirectory(true);
sftpOutboundGateway.setOption(AbstractRemoteFileOutboundGateway.Option.DELETE);
return sftpOutboundGateway;
}
}

1
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/DBFImporterService.java
View file

@ -48,6 +48,7 @@ public class DBFImporterService {
} else {
log.info("adding import task {}", specificTable);
}
fileChecker.checkAndLoadSFTP();
executorService.execute(() -> {
Map<ETable, List<File>> newFiles = null;
try {

51
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/FileChecker.java
View file

@ -1,24 +1,34 @@
package ru.spcex.clearing.dbf.importer.services;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import ru.spcex.clearing.dbf.importer.config.SFTPConfig;
import ru.spcex.clearing.dbf.importer.config.settings.ImportDBFServiceSettings;
import ru.spcex.clearing.dbf.importer.logic.data.enums.ETable;
import ru.spcex.platform.utils.log.ExceptionUtils;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@Service("fileChecker")
public class FileChecker {
private final ImportDBFServiceSettings settings;
private final Logger log = LoggerFactory.getLogger(this.getClass());
private final SFTPConfig.DbfGateway gateway;
public FileChecker(ImportDBFServiceSettings settings) {
public FileChecker(SFTPConfig.DbfGateway gateway,
ImportDBFServiceSettings settings) {
this.gateway = gateway;
this.settings = settings;
}
public Map<ETable, List<File>> checkNewFiles() {
return checkNewFiles(null);
}
public Map<ETable, List<File>> checkNewFiles(ETable specificTable) {
Map<ETable, List<File>> newFiles = new EnumMap<>(ETable.class);
@ -39,19 +49,30 @@ public class FileChecker {
}
private List<File> lsDBF(String dbfDirPath) {
File dbfDir = new File(dbfDirPath);
File[] dbfFiles = dbfDir.listFiles((dir, name) -> {
int formatPosition = name.lastIndexOf(".");
if (formatPosition == -1 || formatPosition == name.length() - 1) return false;
return "dbf".equalsIgnoreCase(name.substring(formatPosition + 1));
});
List<File> resultFiles = new LinkedList<>();
if (dbfFiles != null && dbfFiles.length >= 1) {
Arrays.sort(dbfFiles, Comparator.comparingLong(File::lastModified));
resultFiles.addAll(Arrays.asList(dbfFiles));
try (Stream<Path> stream = Files.walk(Paths.get(dbfDirPath))) {
resultFiles = stream
.filter(file -> !Files.isDirectory(file))
.map(Path::toFile)
.filter(file -> file.getName().toLowerCase(Locale.ROOT).endsWith(".dbf"))
.sorted(Comparator.comparingLong(File::lastModified))
.collect(Collectors.toList());
} catch (IOException e) {
log.error("Error reading 'service.store.src-dir' : {}", ExceptionUtils.getStackTrace(e));
}
return resultFiles;
}
public void checkAndLoadSFTP() {
if (settings.getStore().getSftpIn().getSftpSrcPayValDir() == null || settings.getStore().getSftpIn().getSftpSrcPayValDir().size() == 0) {
log.error("No SFTP scanning directories, need will be adding setting like 'import-dbf-service.store.sftp-in.sftp-src-pay-val-dir.VAL=/VAL' and restart app");
return;
}
List<File> files = new ArrayList<>();
for (String path : settings.getStore().getSftpIn().getSftpSrcPayValDir().values()) {
files.addAll(gateway.listFiles(path));
}
if (files.size() > 0) log.info("Loaded from SFTP files count={}", files.size());
}
}

2
clearing-parent/dbf-importer/src/main/resources/application.properties
View file

@ -14,6 +14,8 @@ import-dbf-service.common.insert-batch-size=100
import-dbf-service.common.threads-count=10
#sftpSrcDir
import-dbf-service.store.sftp-in.sftp-src-pay-val-dir.rub=clearing_importer_sftp/rub/
import-dbf-service.store.sftp-in.sftp-src-pay-val-dir.eur=clearing_importer_sftp/eur/
import-dbf-service.store.sftp-in.sftp-src-dir=clearing_importer_sftp
import-dbf-service.store.sftp-in.user=user
import-dbf-service.store.sftp-in.password=*********

46
clearing-parent/dbf-importer/src/test/java/ru/spcex/clearing/dbf/importer/logic/data/tables/FileWorkTest.java Normal file
View file

@ -0,0 +1,46 @@
package ru.spcex.clearing.dbf.importer.logic.data.tables;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit.jupiter.SpringExtension;
import ru.spcex.clearing.dbf.importer.config.settings.ImportDBFServiceSettings;
import ru.spcex.clearing.dbf.importer.logic.data.enums.ETable;
import ru.spcex.clearing.dbf.importer.logic.stages.ChangeDirOfFileStage;
import ru.spcex.clearing.dbf.importer.services.FileChecker;
import java.io.File;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.List;
import java.util.Map;
@ContextConfiguration(classes = {FileChecker.class, ChangeDirOfFileStage.class})
@EnableConfigurationProperties(value = ImportDBFServiceSettings.class)
@ExtendWith(SpringExtension.class)
class FileWorkTest {
@Autowired
FileChecker fileChecker;
@Autowired
ChangeDirOfFileStage changeDirOfFileStage;
@BeforeAll
static void setProperty() {
Path path = Paths.get("src", "main", "resources");
String currentPath = path.toAbsolutePath().toString();
System.setProperty("spring.config.location", currentPath);
}
@Disabled
@Test
void checkNewFiles() {
Map<ETable, List<File>> newFiles = fileChecker.checkNewFiles(null);
newFiles.size();
}
}

11
platform-parent/platform-utils/src/main/java/ru/spcex/platform/utils/config/SftpInboundFolderSetting.java
View file

@ -1,11 +1,14 @@
package ru.spcex.platform.utils.config;
import java.util.Map;
public class SftpInboundFolderSetting {
private String sftpSrcDir;
private String user;
private String password;
private String serverIp;
private int serverPort;
private Map<String, String> sftpSrcPayValDir;
public String getSftpSrcDir() {
return sftpSrcDir;
@ -46,4 +49,12 @@ public class SftpInboundFolderSetting {
public void setServerPort(int serverPort) {
this.serverPort = serverPort;
}
public Map<String, String> getSftpSrcPayValDir() {
return sftpSrcPayValDir;
}
public void setSftpSrcPayValDir(Map<String, String> sftpSrcPayValDir) {
this.sftpSrcPayValDir = sftpSrcPayValDir;
}
}