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

http://jira.mfd.msk:8088/browse/CLS-358 sftp

Browse source
This commit is contained in:
akulikov 2023-07-31 15:22:10 +03:00
parent cec22d06f8
commit 626d8d3206
14 changed files with 481 additions and 48 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
clearing-parent/reports-service/pom.xml
View file

@ -35,6 +35,10 @@
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-autoconfigure</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-sftp</artifactId>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.dataformat</groupId>
<artifactId>jackson-dataformat-xml</artifactId>

2
clearing-parent/reports-service/src/main/java/ru/spcex/clearing/reports/config/ClearingImdgConfig.java
View file

@ -5,7 +5,7 @@ import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import ru.spcex.clearing.reports.config.element.ReportsServiceSettings;
import ru.spcex.clearing.reports.config.settings.ReportsServiceSettings;
import ru.spcex.platform.imdg.api.ImdgProvider;
import ru.spcex.platform.imdg.iml.hazelcast.service.HazelcastService;

2
clearing-parent/reports-service/src/main/java/ru/spcex/clearing/reports/config/KafkaConfig.java
View file

@ -15,7 +15,7 @@ import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory;
import ru.spcex.clearing.platform.messaging.config.element.KafkaProducerSettings;
import ru.spcex.clearing.platform.messaging.service.RequestInfo;
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
import ru.spcex.clearing.reports.config.element.ReportsServiceSettings;
import ru.spcex.clearing.reports.config.settings.ReportsServiceSettings;
import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.ImdgId;
import ru.spcex.platform.imdg.api.ImdgProvider;

115
clearing-parent/reports-service/src/main/java/ru/spcex/clearing/reports/config/SFTPNotificationsConfig.java Normal file
View file

@ -0,0 +1,115 @@
package ru.spcex.clearing.reports.config;
import com.jcraft.jsch.ChannelSftp;
import org.springframework.beans.factory.annotation.Qualifier;
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.reports.config.settings.ReportsServiceSettings;
import java.io.File;
import java.util.List;
import static org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway.Command.LS;
@Configuration
public class SFTPNotificationsConfig {
@Bean("notificationsSftpSessionFactory")
public SessionFactory<ChannelSftp.LsEntry> sftpSessionFactory(ReportsServiceSettings settings) {
DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
factory.setHost(settings.getNotificationsStore().getServerIp());
factory.setPort(settings.getNotificationsStore().getServerPort());
factory.setUser(settings.getNotificationsStore().getUser());
factory.setPassword(settings.getNotificationsStore().getPassword());
factory.setAllowUnknownKeys(true);
return new CachingSessionFactory<>(factory);
}
@Bean("notificationsSftpHandler")
@ServiceActivator(inputChannel = "notificationsToSftpChannel")
public MessageHandler handler(
@Qualifier("notificationsSftpSessionFactory")
SessionFactory<ChannelSftp.LsEntry> sessionFactory,
ReportsServiceSettings settings
) {
SftpMessageHandler handler = new SftpMessageHandler(sessionFactory);
handler.setRemoteDirectoryExpression(new LiteralExpression(settings.getNotificationsStore().getOutDir()));
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("notificationsListSftpChannel")
public MessageChannel listSftpChannel(
@Qualifier("notificationsSftpSessionFactory")
SessionFactory<ChannelSftp.LsEntry> sessionFactory,
ReportsServiceSettings settings
) {
DirectChannel dc = new DirectChannel();
dc.subscribe(handlerList(sessionFactory, settings));
return dc;
}
@Bean("notificationsToSftpChannel")
public MessageChannel toSftpChannel(
@Qualifier("notificationsSftpSessionFactory")
SessionFactory<ChannelSftp.LsEntry> sessionFactory,
ReportsServiceSettings settings
) {
DirectChannel dc = new DirectChannel();
dc.subscribe(handler(sessionFactory, settings));
return dc;
}
@Bean("notificationsHandlerList")
@ServiceActivator(inputChannel = "notificationsListSftpChannel")
public MessageHandler handlerList(
@Qualifier("notificationsSftpSessionFactory")
SessionFactory<ChannelSftp.LsEntry> sessionFactory,
ReportsServiceSettings settings
) {
String expression = "'/%s'".formatted(settings.getNotificationsStore().getOutDir());
SftpOutboundGateway sftpOutboundGateway = new SftpOutboundGateway(sessionFactory, LS.getCommand(), expression);
return sftpOutboundGateway;
}
@Bean("notificationsSftpOutboundListFlow")
public IntegrationFlow sftpOutboundListFlow(
@Qualifier("notificationsSftpSessionFactory")
SessionFactory<ChannelSftp.LsEntry> sessionFactory
) {
return IntegrationFlows.from("notificationsListSftpChannel")
.handle(new SftpOutboundGateway(sessionFactory, "ls", "payload"))
.get();
}
@MessagingGateway(name = "notificationsSftpGateway")
public interface NotificationsGateway {
@Gateway(requestChannel = "notificationsToSftpChannel")
void sendToSftp(File file);
@Gateway(requestChannel = "notificationsListSftpChannel")
List<SftpFileInfo> listFiles(String dir);
}
}

115
clearing-parent/reports-service/src/main/java/ru/spcex/clearing/reports/config/SFTPReportsConfig.java Normal file
View file

@ -0,0 +1,115 @@
package ru.spcex.clearing.reports.config;
import com.jcraft.jsch.ChannelSftp;
import org.springframework.beans.factory.annotation.Qualifier;
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.reports.config.settings.ReportsServiceSettings;
import java.io.File;
import java.util.List;
import static org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway.Command.LS;
@Configuration
public class SFTPReportsConfig {
@Bean("reportsSftpSessionFactory")
public SessionFactory<ChannelSftp.LsEntry> sftpSessionFactory(ReportsServiceSettings settings) {
DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
factory.setHost(settings.getReportsStore().getServerIp());
factory.setPort(settings.getReportsStore().getServerPort());
factory.setUser(settings.getReportsStore().getUser());
factory.setPassword(settings.getReportsStore().getPassword());
factory.setAllowUnknownKeys(true);
return new CachingSessionFactory<>(factory);
}
@Bean("reportsSftpHandler")
@ServiceActivator(inputChannel = "reportsToSftpChannel")
public MessageHandler handler(
@Qualifier("reportsSftpSessionFactory")
SessionFactory<ChannelSftp.LsEntry> sessionFactory,
ReportsServiceSettings settings
) {
SftpMessageHandler handler = new SftpMessageHandler(sessionFactory);
handler.setRemoteDirectoryExpression(new LiteralExpression(settings.getReportsStore().getOutDir()));
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("reportsListSftpChannel")
public MessageChannel listSftpChannel(
@Qualifier("reportsSftpSessionFactory")
SessionFactory<ChannelSftp.LsEntry> sessionFactory,
ReportsServiceSettings settings
) {
DirectChannel dc = new DirectChannel();
dc.subscribe(handlerList(sessionFactory, settings));
return dc;
}
@Bean("reportsToSftpChannel")
public MessageChannel toSftpChannel(
@Qualifier("reportsSftpSessionFactory")
SessionFactory<ChannelSftp.LsEntry> sessionFactory,
ReportsServiceSettings settings
) {
DirectChannel dc = new DirectChannel();
dc.subscribe(handler(sessionFactory, settings));
return dc;
}
@Bean("reportsHandlerList")
@ServiceActivator(inputChannel = "reportsListSftpChannel")
public MessageHandler handlerList(
@Qualifier("reportsSftpSessionFactory")
SessionFactory<ChannelSftp.LsEntry> sessionFactory,
ReportsServiceSettings settings
) {
String expression = "'/%s'".formatted(settings.getReportsStore().getOutDir());
SftpOutboundGateway sftpOutboundGateway = new SftpOutboundGateway(sessionFactory, LS.getCommand(), expression);
return sftpOutboundGateway;
}
@Bean("reportsSftpOutboundListFlow")
public IntegrationFlow sftpOutboundListFlow(
@Qualifier("reportsSftpSessionFactory")
SessionFactory<ChannelSftp.LsEntry> sessionFactory
) {
return IntegrationFlows.from("reportsListSftpChannel")
.handle(new SftpOutboundGateway(sessionFactory, "ls", "payload"))
.get();
}
@MessagingGateway(name = "reportsSftpGateway")
public interface ReportsGateway {
@Gateway(requestChannel = "reportsToSftpChannel")
void sendToSftp(File file);
@Gateway(requestChannel = "reportsListSftpChannel")
List<SftpFileInfo> listFiles(String dir);
}
}

22
clearing-parent/reports-service/src/main/java/ru/spcex/clearing/reports/config/element/ReportsServiceSettings.java → clearing-parent/reports-service/src/main/java/ru/spcex/clearing/reports/config/settings/ReportsServiceSettings.java
View file

@ -1,4 +1,4 @@
package ru.spcex.clearing.reports.config.element;
package ru.spcex.clearing.reports.config.settings;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.PropertySource;
@ -14,8 +14,8 @@ public class ReportsServiceSettings {
private HazelcastClientParams hazelcast;
private KafkaConsumerSettings kafkaConsumer;
private KafkaProducerSettings kafkaProducer;
private String reportModuleOut;
private String notificationModuleOut;
private Store reportsStore;
private Store notificationsStore;
public HazelcastClientParams getHazelcast() {
return hazelcast;
@ -41,19 +41,19 @@ public class ReportsServiceSettings {
this.kafkaProducer = kafkaProducer;
}
public String getReportModuleOut() {
return reportModuleOut;
public Store getReportsStore() {
return reportsStore;
}
public void setReportModuleOut(String reportModuleOut) {
this.reportModuleOut = reportModuleOut;
public void setReportsStore(Store reportsStore) {
this.reportsStore = reportsStore;
}
public String getNotificationModuleOut() {
return notificationModuleOut;
public Store getNotificationsStore() {
return notificationsStore;
}
public void setNotificationModuleOut(String notificationModuleOut) {
this.notificationModuleOut = notificationModuleOut;
public void setNotificationsStore(Store notificationsStore) {
this.notificationsStore = notificationsStore;
}
}

69
clearing-parent/reports-service/src/main/java/ru/spcex/clearing/reports/config/settings/Store.java Normal file
View file

@ -0,0 +1,69 @@
package ru.spcex.clearing.reports.config.settings;
public class Store {
private String outDir;
private String localTempDir;
private String user;
private String password;
private String serverIp;
private int serverPort;
public boolean isDeleteAfterSend() {
return deleteAfterSend;
}
public void setDeleteAfterSend(boolean deleteAfterSend) {
this.deleteAfterSend = deleteAfterSend;
}
private boolean deleteAfterSend;
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;
}
public String getOutDir() {
return outDir;
}
public void setOutDir(String outDir) {
this.outDir = outDir;
}
public String getLocalTempDir() {
return localTempDir;
}
public void setLocalTempDir(String localTempDir) {
this.localTempDir = localTempDir;
}
}

46
clearing-parent/reports-service/src/main/java/ru/spcex/clearing/reports/services/NotificationService.java
View file

@ -11,12 +11,15 @@ import ru.spcex.clearing.platform.messaging.domain.Consts;
import ru.spcex.clearing.platform.messaging.domain.cud.reports.NotificationRequest;
import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
import ru.spcex.clearing.platform.messaging.service.RequestInfoUpdate;
import ru.spcex.clearing.reports.config.element.ReportsServiceSettings;
import ru.spcex.clearing.reports.config.SFTPNotificationsConfig;
import ru.spcex.clearing.reports.config.settings.ReportsServiceSettings;
import ru.spcex.clearing.reports.notifications.NCMPNotificationBuilder;
import ru.spcex.clearing.reports.notifications.NTCRNotificationBuilder;
import ru.spcex.clearing.util.security.UserRoleVerification;
import java.io.File;
import java.util.Collection;
import java.util.List;
import java.util.Map;
@Service
@ -27,19 +30,30 @@ public class NotificationService extends QueueConsumer implements InitializingBe
private final NTCRNotificationBuilder ntcrNotificationBuilder;
private final File outFolder;
private final boolean deleteAfterSend;
private final SFTPNotificationsConfig.NotificationsGateway notificationsSftpGateway;
public NotificationService(Consumer<String, Object> kafkaQueue,
Producer<String, Object> kafkaResponseQueue,
UserRoleVerification userRoleVerification,
NCMPNotificationBuilder ncmpNotificationBuilder,
NTCRNotificationBuilder ntcrNotificationBuilder,
ReportsServiceSettings reportsServiceSettings) {
ReportsServiceSettings reportsServiceSettings,
SFTPNotificationsConfig.NotificationsGateway notificationsSftpGateway) {
super(kafkaQueue, kafkaResponseQueue);
this.userRoleVerification = userRoleVerification;
this.ncmpNotificationBuilder = ncmpNotificationBuilder;
this.ntcrNotificationBuilder = ntcrNotificationBuilder;
this.outFolder = new File(reportsServiceSettings.getNotificationModuleOut());
this.notificationsSftpGateway = notificationsSftpGateway;
this.outFolder = new File(reportsServiceSettings.getNotificationsStore().getLocalTempDir());
if (!outFolder.exists()) {
boolean mkDirOk = outFolder.mkdirs();
if (!mkDirOk) throw new IllegalStateException("Can't create output directory for reports, reports-service was terminated");
} else if (!outFolder.isDirectory()) {
throw new IllegalStateException("Output reports path from settings is not directory!");
}
assert outFolder.isDirectory();
this.deleteAfterSend = reportsServiceSettings.getNotificationsStore().isDeleteAfterSend();
}
@Override
@ -63,10 +77,11 @@ public class NotificationService extends QueueConsumer implements InitializingBe
logUnknownProperties(userRequest);
File outFile = ncmpNotificationBuilder.buildNotification(req.getConsumerId(), outFolder);
if (outFile == null) {
log.error("Create notification NCMP for id={} failed, see log", req.getConsumerId());
} else {
if (outFile != null) {
log.debug("Notification NCMP for id={} successfully created. Output files: {}", req.getConsumerId(), outFile);
sendFilesToSftp(List.of(outFile));
} else {
log.error("Create notification NCMP for id={} failed, see log", req.getConsumerId());
}
return null;
}
@ -81,10 +96,11 @@ public class NotificationService extends QueueConsumer implements InitializingBe
logUnknownProperties(userRequest);
File outFile = ntcrNotificationBuilder.buildNotification(req.getConsumerId(), outFolder);
if (outFile == null) {
log.error("Create notification NTCR for id={} failed, see log", req.getConsumerId());
} else {
if (outFile != null) {
log.debug("Notification NTCR for id={} successfully created. Output files: {}", req.getConsumerId(), outFile);
sendFilesToSftp(List.of(outFile));
} else {
log.error("Create notification NTCR for id={} failed, see log", req.getConsumerId());
}
return null;
}
@ -95,4 +111,16 @@ public class NotificationService extends QueueConsumer implements InitializingBe
log.warn("Unknown property in request {} = {}", unknownProperty.getKey(), unknownProperty.getValue());
}
}
private void sendFilesToSftp(Collection<File> files) {
for (File file : files) {
notificationsSftpGateway.sendToSftp(file);
if (deleteAfterSend) {
boolean delete = file.delete();
if (!delete) {
log.warn("Can't delete temp file {}", file.getAbsolutePath());
}
}
}
}
}

76
clearing-parent/reports-service/src/main/java/ru/spcex/clearing/reports/services/ReportService.java
View file

@ -19,7 +19,8 @@ import ru.spcex.clearing.platform.messaging.domain.cud.schedule.LauncherCommandR
import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
import ru.spcex.clearing.platform.messaging.service.RequestInfoUpdate;
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
import ru.spcex.clearing.reports.config.element.ReportsServiceSettings;
import ru.spcex.clearing.reports.config.SFTPReportsConfig;
import ru.spcex.clearing.reports.config.settings.ReportsServiceSettings;
import ru.spcex.clearing.reports.reports.CSVReportBuilder;
import ru.spcex.clearing.reports.reports.EmptyParams;
import ru.spcex.clearing.reports.reports.ReportInfo;
@ -62,6 +63,8 @@ public class ReportService extends QueueConsumer implements InitializingBean {
private final Imdg<Session> sessionImdg;
private final File outFolder;
private final SFTPReportsConfig.ReportsGateway reportsSftpGateway;
private final boolean deleteAfterSend;
public ReportService(Consumer<String, Object> kafkaQueue,
Producer<String, Object> kafkaResponseQueue,
@ -94,8 +97,8 @@ public class ReportService extends QueueConsumer implements InitializingBean {
@Qualifier("reportRequestGREFValidator")
Function<LauncherCommandRequest, IValidator> reportRequestGREFValidator,
ImdgProvider imdgProvider
ImdgProvider imdgProvider,
SFTPReportsConfig.ReportsGateway reportsSftpGateway
) {
super(kafkaQueue, kafkaResponseQueue);
this.userRoleVerification = userRoleVerification;
@ -111,8 +114,9 @@ public class ReportService extends QueueConsumer implements InitializingBean {
this.reportBuildersForGREP = reportBuildersForGREP;
this.reportBuildersForGREF = reportBuildersForGREF;
this.reportBuildersForGRET = reportBuildersForGRET;
this.reportsSftpGateway = reportsSftpGateway;
this.sessionImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Session, Session.class);
this.outFolder = new File(reportsServiceSettings.getReportModuleOut());
this.outFolder = new File(reportsServiceSettings.getReportsStore().getLocalTempDir());
if (!outFolder.exists()) {
boolean mkDirOk = outFolder.mkdirs();
if (!mkDirOk) throw new IllegalStateException("Can't create output directory for reports, reports-service was terminated");
@ -120,6 +124,7 @@ public class ReportService extends QueueConsumer implements InitializingBean {
throw new IllegalStateException("Output reports path from settings is not directory!");
}
assert outFolder.isDirectory();
this.deleteAfterSend = reportsServiceSettings.getReportsStore().isDeleteAfterSend();
}
@Override
@ -160,10 +165,12 @@ public class ReportService extends QueueConsumer implements InitializingBean {
IEnumKey.getEnumByKey(ReportBuilderType.class, req.getReportId())
);
List<String> outFilenames = new ArrayList<>();
List<File> outFiles = new ArrayList<>();
for (CSVReportBuilder<EmptyParams, ?> builder : builders) {
File outFile = builder.createReport(new EmptyParams(), outFolder);
if (outFile == null) continue;
outFilenames.add(outFile.getAbsolutePath());
outFiles.add(outFile);
}
int outFilesCnt = outFilenames.size();
@ -174,6 +181,8 @@ public class ReportService extends QueueConsumer implements InitializingBean {
} else {
log.debug("Reports {} successfully created. Output files: {}", req.getReportId(), String.join(", ", outFilenames));
}
sendFilesToSftp(outFiles);
return null;
}
@ -195,6 +204,7 @@ public class ReportService extends QueueConsumer implements InitializingBean {
IEnumKey.getEnumByKey(ReportBuilderType.class, req.getReportId())
);
List<String> outFilenames = new ArrayList<>();
List<File> outFiles = new ArrayList<>();
for (CSVReportBuilder<SessionIdParam, ?> builder : builders) {
SessionIdParam sessionIdParam = new SessionIdParam();
sessionIdParam.setSessionId(List.of(req.getSessionId()));
@ -211,6 +221,8 @@ public class ReportService extends QueueConsumer implements InitializingBean {
} else {
log.debug("Reports {} successfully created. Output files: {}", req.getReportId(), String.join(", ", outFilenames));
}
sendFilesToSftp(outFiles);
return null;
}
@ -229,9 +241,10 @@ public class ReportService extends QueueConsumer implements InitializingBean {
} else {
sessionIds = getSessionIdsForDate(LocalDate.now());
}
Map<String, ReportInfo> outFilenames = createReportsForTask(sessionIds, reportBuildersForGREP.values());
Map<File, ReportInfo> outFiles = createReportsForTask(sessionIds, reportBuildersForGREP.values());
sendReportsToGateway(sessionIds, outFilenames, ReportType.REPORT_KS_TMP.getKey());
sendReportsToGateway(sessionIds, outFiles, ReportType.REPORT_KS_TMP.getKey());
sendFilesToSftp(outFiles.keySet());
return null;
}
@ -250,9 +263,10 @@ public class ReportService extends QueueConsumer implements InitializingBean {
} else {
sessionIds = getSessionIdsForDate(LocalDate.now());
}
Map<String, ReportInfo> outFilenames = createReportsForTask(sessionIds, reportBuildersForGRET.values());
Map<File, ReportInfo> outFiles = createReportsForTask(sessionIds, reportBuildersForGRET.values());
sendReportsToGateway(sessionIds, outFilenames, null);
sendReportsToGateway(sessionIds, outFiles, null);
sendFilesToSftp(outFiles.keySet());
return null;
}
@ -271,14 +285,15 @@ public class ReportService extends QueueConsumer implements InitializingBean {
} else {
sessionIds = getSessionIdsForDate(LocalDate.now());
}
Map<String, ReportInfo> outFilenames = createReportsForTask(sessionIds, reportBuildersForGREF.values());
Map<File, ReportInfo> outFiles = createReportsForTask(sessionIds, reportBuildersForGREF.values());
sendReportsToGateway(sessionIds, outFilenames, ReportType.REPORT_KS_FINAL.getKey());
sendReportsToGateway(sessionIds, outFiles, ReportType.REPORT_KS_FINAL.getKey());
sendFilesToSftp(outFiles.keySet());
return null;
}
private Map<String, ReportInfo> createReportsForTask(List<Long> sessionIds, Collection<List<CSVReportBuilder<?, ?>>> buildersForTask) {
Map<String, ReportInfo> outFilenames = new HashMap<>();
private Map<File, ReportInfo> createReportsForTask(List<Long> sessionIds, Collection<List<CSVReportBuilder<?, ?>>> buildersForTask) {
Map<File, ReportInfo> outFiles = new HashMap<>();
int cntErrors = 0;
for (List<CSVReportBuilder<?, ?>> builders : buildersForTask) {
for (CSVReportBuilder<?, ?> builder : builders) {
@ -290,7 +305,7 @@ public class ReportService extends QueueConsumer implements InitializingBean {
cntErrors++;
continue;
}
outFilenames.put(outFile.getName(), new ReportInfo(builder.getReportKey().getKey(), builder.getSection()));
outFiles.put(outFile, new ReportInfo(builder.getReportKey().getKey(), builder.getSection()));
} else if (builder.getParamsClass() == SessionIdParam.class) {
SessionIdParam sessionIdParam = new SessionIdParam();
sessionIdParam.setSessionId(sessionIds);
@ -300,28 +315,38 @@ public class ReportService extends QueueConsumer implements InitializingBean {
cntErrors++;
continue;
}
outFilenames.put(outFile.getName(), reportInfo);
outFiles.put(outFile, reportInfo);
}
}
}
if (cntErrors == 0) log.debug("All reports successfully created. Output files: {}", String.join(", ", outFilenames.keySet()));
else log.debug("Some reports ({}) create failed. Output files: {}", cntErrors, String.join(", ", outFilenames.keySet()));
if (cntErrors == 0) {
log.debug(
"All reports successfully created. Output files: {}",
outFiles.keySet().stream().map(File::getName).collect(Collectors.joining(", "))
);
} else {
log.debug(
"Some reports ({}) create failed. Output files: {}",
cntErrors,
outFiles.keySet().stream().map(File::getName).collect(Collectors.joining(", "))
);
}
return outFilenames;
return outFiles;
}
private void sendReportsToGateway(List<Long> sessionIds, Map<String, ReportInfo> reportsWithReportType, String type) {
private void sendReportsToGateway(List<Long> sessionIds, Map<File, ReportInfo> reportsWithReportType, String type) {
SendReportRequest sendReportRequest = new SendReportRequest();
sendReportRequest.setSessionId(sessionIds);
sendReportRequest.setType(type);
List<ReportPart> reports = new ArrayList<>(reportsWithReportType.size());
for (Map.Entry<String, ReportInfo> entry : reportsWithReportType.entrySet()) {
for (Map.Entry<File, ReportInfo> entry : reportsWithReportType.entrySet()) {
ReportInfo reportInfo = entry.getValue();
ReportPart reportPart = new ReportPart();
reportPart.setReportType(reportInfo.reportKey());
reportPart.setSection(reportInfo.section());
reportPart.setFileName(entry.getKey());
reportPart.setFileName(entry.getKey().getName());
reports.add(reportPart);
}
@ -346,4 +371,15 @@ public class ReportService extends QueueConsumer implements InitializingBean {
return sessionIds.stream().toList();
}
private void sendFilesToSftp(Collection<File> files) {
for (File file : files) {
reportsSftpGateway.sendToSftp(file);
if (deleteAfterSend) {
boolean delete = file.delete();
if (!delete) {
log.warn("Can't delete temp file {}", file.getAbsolutePath());
}
}
}
}
}

15
clearing-parent/reports-service/src/main/resources/application.properties
View file

@ -17,5 +17,16 @@ reports-service.kafka-producer.batch-size=16384
reports-service.kafka-producer.linger-ms=1
reports-service.kafka-producer.buffer-memory=33554432
reports-service.ReportModuleOut=./report_out
reports-service.NotificationModuleOut=./notification_out
reports-service.reports-store.local-temp-dir=./reports_out_temp
reports-service.reports-store.out-dir=reports
reports-service.reports-store.user=teste
reports-service.reports-store.password=password
reports-service.reports-store.server-ip=10.230.238.53
reports-service.reports-store.server-port=2222
reports-service.notifications-store.local-temp-dir=./notifications_out_temp
reports-service.notifications-store.out-dir=notifications
reports-service.notifications-store.user=tester
reports-service.notifications-store.password=password
reports-service.notifications-store.server-ip=10.230.238.53
reports-service.notifications-store.server-port=2222

48
clearing-parent/reports-service/src/test/java/ru/spcex/clearing/reports/config/SFTPTestConfig.java Normal file
View file

@ -0,0 +1,48 @@
package ru.spcex.clearing.reports.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.sftp.session.SftpFileInfo;
import java.io.File;
import java.util.ArrayList;
import java.util.List;
@Configuration
public class SFTPTestConfig {
@Bean
public SFTPReportsConfig.ReportsGateway reportsGateway(){
return new ReportsGateway();
}
public static class ReportsGateway implements SFTPReportsConfig.ReportsGateway {
@Override
public void sendToSftp(File file) {
}
@Override
public List<SftpFileInfo> listFiles(String dir) {
return new ArrayList<>();
}
}
@Bean
public SFTPNotificationsConfig.NotificationsGateway notificationsGateway(){
return new NotificationsGateway();
}
public static class NotificationsGateway implements SFTPNotificationsConfig.NotificationsGateway {
@Override
public void sendToSftp(File file) {
}
@Override
public List<SftpFileInfo> listFiles(String dir) {
return new ArrayList<>();
}
}
}

7
clearing-parent/reports-service/src/test/java/ru/spcex/clearing/reports/config/TestConfig.java
View file

@ -2,7 +2,8 @@ package ru.spcex.clearing.reports.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import ru.spcex.clearing.reports.config.element.ReportsServiceSettings;
import ru.spcex.clearing.reports.config.settings.ReportsServiceSettings;
import ru.spcex.clearing.reports.config.settings.Store;
@Configuration
public class TestConfig {
@ -10,7 +11,9 @@ public class TestConfig {
@Bean
public ReportsServiceSettings reportsServiceSettings() {
ReportsServiceSettings reportsServiceSettings = new ReportsServiceSettings();
reportsServiceSettings.setReportModuleOut("out");
Store store = new Store();
store.setLocalTempDir("out");
reportsServiceSettings.setReportsStore(store);
return reportsServiceSettings;
}

4
clearing-parent/reports-service/src/test/java/ru/spcex/clearing/reports/services/ReportServiceTest_BP.java
View file

@ -29,6 +29,7 @@ import ru.spcex.clearing.platform.messaging.domain.Consts;
import ru.spcex.clearing.platform.messaging.domain.cud.reports.ReportRequestWithSessionId;
import ru.spcex.clearing.reports.config.BeanConfiguration;
import ru.spcex.clearing.reports.config.ReportBuildersConfig;
import ru.spcex.clearing.reports.config.SFTPTestConfig;
import ru.spcex.clearing.reports.config.TestConfig;
import ru.spcex.clearing.reports.config.validation.ValidationConfig;
import ru.spcex.clearing.test.TestUtils;
@ -58,7 +59,8 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
ImdgTestConfig.class,
KafkaTestConfig.class,
ReportBuildersConfig.class,
TestConfig.class
TestConfig.class,
SFTPTestConfig.class
})
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
public class ReportServiceTest_BP {

4
clearing-parent/reports-service/src/test/java/ru/spcex/clearing/reports/services/ReportServiceTest_KS.java
View file

@ -31,6 +31,7 @@ import ru.spcex.clearing.platform.messaging.domain.cud.reports.ReportRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.reports.ReportRequestWithSessionId;
import ru.spcex.clearing.reports.config.BeanConfiguration;
import ru.spcex.clearing.reports.config.ReportBuildersConfig;
import ru.spcex.clearing.reports.config.SFTPTestConfig;
import ru.spcex.clearing.reports.config.TestConfig;
import ru.spcex.clearing.reports.config.validation.ValidationConfig;
import ru.spcex.clearing.test.TestUtils;
@ -62,7 +63,8 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
ImdgTestConfig.class,
KafkaTestConfig.class,
ReportBuildersConfig.class,
TestConfig.class
TestConfig.class,
SFTPTestConfig.class
})
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
public class ReportServiceTest_KS {