swt-exporter SFTP

This commit is contained in:
ialbert 2023-07-31 20:19:17 +03:00
parent 5343be3b58
commit 67a80c7475
5 changed files with 181 additions and 1 deletions

View file

@ -0,0 +1,102 @@
package ru.spcex.clearing.swt.exporter.config;
import com.jcraft.jsch.ChannelSftp;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.integration.annotation.Gateway;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.file.remote.session.CachingSessionFactory;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.sftp.gateway.SftpOutboundGateway;
import org.springframework.integration.sftp.outbound.SftpMessageHandler;
import org.springframework.integration.sftp.session.DefaultSftpSessionFactory;
import org.springframework.integration.sftp.session.SftpFileInfo;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import ru.spcex.clearing.swt.exporter.config.settings.ExportSwtServiceSettings;
import java.io.File;
import java.util.List;
import static org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway.Command.LS;
@Configuration
@ConditionalOnProperty(value="export-swt-service.sftp-out.sftpOutDir")
public class SFTPConfig {
private final ExportSwtServiceSettings settings;
public SFTPConfig(ExportSwtServiceSettings settings) {
this.settings = settings;
}
@Bean("sftpSessionFactory")
public SessionFactory<ChannelSftp.LsEntry> sftpSessionFactory() {
DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
factory.setHost(settings.getSftpOut().getServerIp());
factory.setPort(settings.getSftpOut().getServerPort());
factory.setUser(settings.getSftpOut().getUser());
factory.setPassword(settings.getSftpOut().getPassword());
factory.setAllowUnknownKeys(true);
return new CachingSessionFactory<>(factory);
}
@Bean("sftpHandler")
@ServiceActivator(inputChannel = "toSftpChannel")
public MessageHandler handler(@Qualifier("sftpSessionFactory") SessionFactory<ChannelSftp.LsEntry> sessionFactory) {
SftpMessageHandler handler = new SftpMessageHandler(sessionFactory);
handler.setRemoteDirectoryExpression(new LiteralExpression(settings.getSftpOut().getSftpOutDir()));
handler.setAutoCreateDirectory(true);
handler.setFileNameGenerator(message -> {
if (message.getPayload() instanceof File) {
return ((File) message.getPayload()).getName();
} else {
throw new IllegalArgumentException("File must expected as payload.");
}
});
return handler;
}
@Bean("sftpHandlerList")
@ServiceActivator(inputChannel = "listSftpChannel")
public MessageHandler handlerList(@Qualifier("sftpSessionFactory") SessionFactory<ChannelSftp.LsEntry> sessionFactory) {
String expression = "'/%s'".formatted(settings.getSftpOut().getSftpOutDir());
return new SftpOutboundGateway(sessionFactory, LS.getCommand(), expression);
}
@Bean("toSftpChannel")
public MessageChannel toSftpChannel(@Qualifier("sftpHandler") MessageHandler handler) {
DirectChannel dc = new DirectChannel();
dc.subscribe(handler);
return dc;
}
@Bean("listSftpChannel")
public MessageChannel listSftpChannel(@Qualifier("sftpHandlerList") MessageHandler handler) {
DirectChannel dc = new DirectChannel();
dc.subscribe(handler);
return dc;
}
@Bean("notificationsSftpOutboundListFlow")
public IntegrationFlow sftpOutboundListFlow(@Qualifier("sftpSessionFactory") SessionFactory<ChannelSftp.LsEntry> sessionFactory) {
return IntegrationFlows.from("listSftpChannel")
.handle(new SftpOutboundGateway(sessionFactory, "ls", "payload"))
.get();
}
@MessagingGateway(name = "notificationsSftpGateway")
public interface SftpGateway {
@Gateway(requestChannel = "toSftpChannel")
void sendToSftp(File file);
@Gateway(requestChannel = "listSftpChannel")
List<SftpFileInfo> listFiles(String dir);
}
}

View file

@ -6,6 +6,7 @@ import org.springframework.stereotype.Component;
import ru.spcex.clearing.platform.messaging.config.element.KafkaConsumerSettings;
import ru.spcex.clearing.platform.messaging.config.element.KafkaProducerSettings;
import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams;
import ru.spcex.platform.utils.config.SftpOutboundFolderSetting;
@Component
@PropertySource("file:${spring.config.location}/application.properties")
@ -15,10 +16,19 @@ public class ExportSwtServiceSettings {
private KafkaConsumerSettings kafkaConsumer;
private KafkaProducerSettings kafkaProducer;
private SftpOutboundFolderSetting sftpOut;
private String docOut;
//todo ??? private Long interval
public SftpOutboundFolderSetting getSftpOut() {
return sftpOut;
}
public void setSftpOut(SftpOutboundFolderSetting sftpOut) {
this.sftpOut = sftpOut;
}
public HazelcastClientParams getHazelcast() {
return hazelcast;
}

View file

@ -5,6 +5,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import ru.spcex.clearing.swt.exporter.config.SFTPConfig;
import ru.spcex.clearing.swt.exporter.config.settings.ExportSwtServiceSettings;
import java.io.File;
@ -14,6 +15,7 @@ import java.io.IOException;
public class FileStorage {
protected final Logger log = LoggerFactory.getLogger(getClass());
protected File outPath;
private SFTPConfig.SftpGateway gateway;
@Autowired
public FileStorage(ExportSwtServiceSettings config) {
@ -30,8 +32,19 @@ public class FileStorage {
log.info("Output directory \"{}\"", outPath);
}
@Autowired(required = false)
public void setGateway(SFTPConfig.SftpGateway gateway) {
this.gateway = gateway;
}
public void saveFile(String fileName, byte[] data) throws IOException {
File toFile = new File(outPath, fileName);
FileUtils.writeByteArrayToFile(toFile, data);
if (gateway == null) {
log.debug("sftp is disabled. not sending {}", toFile.getAbsolutePath());
} else {
log.debug("sftp is enabled. sending {}", toFile.getAbsolutePath());
gateway.sendToSftp(toFile);
}
}
}

View file

@ -22,4 +22,10 @@ export-swt-service.kafka-producer.acks=all
export-swt-service.kafka-producer.retries=0
export-swt-service.kafka-producer.batch-size=16384
export-swt-service.kafka-producer.linger-ms=1
export-swt-service.kafka-producer.buffer-memory=33554432
export-swt-service.kafka-producer.buffer-memory=33554432
export-swt-service.sftp-out.sftpOutDir=swt_out_sftp_directory
export-swt-service.sftp-out.user=user
export-swt-service.sftp-out.password=********
export-swt-service.sftp-out.server-ip=127.0.0.1
export-swt-service.sftp-out.server-port=22

View file

@ -0,0 +1,49 @@
package ru.spcex.platform.utils.config;
public class SftpOutboundFolderSetting {
private String sftpOutDir;
private String user;
private String password;
private String serverIp;
private int serverPort;
public String getSftpOutDir() {
return sftpOutDir;
}
public void setSftpOutDir(String sftpOutDir) {
this.sftpOutDir = sftpOutDir;
}
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;
}
}