diff --git a/clearing-parent/lim-exporter/pom.xml b/clearing-parent/lim-exporter/pom.xml index 879ff7543..1a9f7225b 100644 --- a/clearing-parent/lim-exporter/pom.xml +++ b/clearing-parent/lim-exporter/pom.xml @@ -23,12 +23,15 @@ org.springframework.boot spring-boot-starter + + com.fasterxml.jackson.core + jackson-databind + org.springframework.integration spring-integration-sftp - ru.spcex.clearing classes diff --git a/clearing-parent/lim-exporter/src/main/java/ru/clearing/lim/exporter/config/settings/Store.java b/clearing-parent/lim-exporter/src/main/java/ru/clearing/lim/exporter/config/settings/Store.java deleted file mode 100644 index 4adebe4d9..000000000 --- a/clearing-parent/lim-exporter/src/main/java/ru/clearing/lim/exporter/config/settings/Store.java +++ /dev/null @@ -1,14 +0,0 @@ -package ru.clearing.lim.exporter.config.settings; - -public class Store { - - private String outDir; - - public String getOutDir() { - return outDir; - } - - public void setOutDir(String outDir) { - this.outDir = outDir; - } -} diff --git a/clearing-parent/lim-exporter/src/main/java/ru/clearing/lim/exporter/LimExportApplication.java b/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/LimExportApplication.java similarity index 94% rename from clearing-parent/lim-exporter/src/main/java/ru/clearing/lim/exporter/LimExportApplication.java rename to clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/LimExportApplication.java index 9a6c8f8a2..f2d29d17a 100644 --- a/clearing-parent/lim-exporter/src/main/java/ru/clearing/lim/exporter/LimExportApplication.java +++ b/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/LimExportApplication.java @@ -1,4 +1,4 @@ -package ru.clearing.lim.exporter; +package ru.spcex.clearing.lim.exporter; import org.slf4j.LoggerFactory; import org.springframework.boot.autoconfigure.SpringBootApplication; diff --git a/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/config/KafkaConfig.java b/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/config/KafkaConfig.java new file mode 100644 index 000000000..45fc6642c --- /dev/null +++ b/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/config/KafkaConfig.java @@ -0,0 +1,28 @@ +package ru.spcex.clearing.lim.exporter.config; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.producer.Producer; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.config.ConfigurableBeanFactory; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Scope; +import ru.spcex.clearing.lim.exporter.config.settings.ExportLimServiceSettings; +import ru.spcex.clearing.platform.messaging.config.KafkaConsumerFactory; +import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory; + +@Configuration +public class KafkaConfig { + @Autowired + @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE) + @Bean + public Consumer createConsumer(ExportLimServiceSettings settings) { + return KafkaConsumerFactory.consumer(settings.getKafkaConsumer()); + } + + @Autowired + @Bean + public Producer createProducer(ExportLimServiceSettings settings) { + return KafkaProducerFactory.producer(settings.getKafkaProducer()); + } +} diff --git a/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/config/LimExporterConfig.java b/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/config/LimExporterConfig.java new file mode 100644 index 000000000..b81b8820f --- /dev/null +++ b/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/config/LimExporterConfig.java @@ -0,0 +1,63 @@ +package ru.spcex.clearing.lim.exporter.config; + +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.ComponentScan; +import org.springframework.context.annotation.Configuration; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import ru.spcex.clearing.lim.exporter.config.settings.ExportLimServiceSettings; +import ru.spcex.platform.imdg.api.ImdgProvider; +import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams; +import ru.spcex.platform.imdg.iml.hazelcast.service.HazelcastService; + +@Configuration +@EnableConfigurationProperties +@ComponentScan(basePackages = {"ru.spcex.clearing.lim.exporter"}) +public class LimExporterConfig { + + @Bean("taskExecutorHazelcastClientInitializer") + public ThreadPoolTaskExecutor taskExecutorHazelcastClientInitializer() { + return createThreadPoolTaskExecutor(1, true); + } + + @Bean("taskExecutorIdGeneratorAwaiter") + public ThreadPoolTaskExecutor taskExecutorIdGeneratorAwaiter() { + return createThreadPoolTaskExecutor(1, false); + } + + @Bean("imdgProvider") + public ImdgProvider imdgProvider(@Qualifier("taskExecutorHazelcastClientInitializer") ThreadPoolTaskExecutor taskExecutorHazelcastClientInitializer, + @Qualifier("taskExecutorIdGeneratorAwaiter") ThreadPoolTaskExecutor taskExecutorIdGeneratorAwaiter, + ExportLimServiceSettings settings) { + HazelcastClientParams params = new HazelcastClientParams(); + params.setClusterMembers(settings.getHazelcast().getClusterMembers()); + params.setLogin(settings.getHazelcast().getLogin()); + params.setPassword(settings.getHazelcast().getPassword()); + return new HazelcastService(taskExecutorHazelcastClientInitializer, taskExecutorIdGeneratorAwaiter, params); + } + + @Bean("executor") + public ThreadPoolTaskExecutor executor(ExportLimServiceSettings settings) { + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + executor.setMaxPoolSize(settings.getCommon().getThreadsCount()); + executor.setCorePoolSize(settings.getCommon().getThreadsCount()); + executor.setThreadNamePrefix("dbf-exporter"); + executor.setWaitForTasksToCompleteOnShutdown(true); + executor.setAwaitTerminationSeconds(300); + executor.initialize(); + return executor; + } + + private static ThreadPoolTaskExecutor createThreadPoolTaskExecutor(int maxPoolSz, boolean waitForCompletion) { + ThreadPoolTaskExecutor pool = new ThreadPoolTaskExecutor(); + if (maxPoolSz > 2) { + pool.setKeepAliveSeconds(60); + pool.setAllowCoreThreadTimeOut(true); + } + pool.setCorePoolSize(maxPoolSz); + pool.setWaitForTasksToCompleteOnShutdown(waitForCompletion); + return pool; + } + +} diff --git a/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/config/SFTPConfig.java b/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/config/SFTPConfig.java new file mode 100644 index 000000000..c6f9c1670 --- /dev/null +++ b/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/config/SFTPConfig.java @@ -0,0 +1,54 @@ +package ru.spcex.clearing.lim.exporter.config; + +import com.jcraft.jsch.ChannelSftp; +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.file.remote.session.CachingSessionFactory; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.sftp.outbound.SftpMessageHandler; +import org.springframework.integration.sftp.session.DefaultSftpSessionFactory; +import org.springframework.messaging.MessageHandler; +import ru.spcex.clearing.lim.exporter.config.settings.ExportLimServiceSettings; + +import java.io.File; + +@Configuration +public class SFTPConfig { + + @Bean + public SessionFactory sftpSessionFactory(ExportLimServiceSettings settings) { + 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 + @ServiceActivator(inputChannel = "toSftpChannel") + public MessageHandler handler(SessionFactory sessionFactory, ExportLimServiceSettings settings) { + SftpMessageHandler handler = new SftpMessageHandler(sessionFactory); + handler.setRemoteDirectoryExpression(new LiteralExpression(settings.getStore().getOutDir())); + handler.setAutoCreateDirectory(true); + handler.setFileNameGenerator(message -> { + if (message.getPayload() instanceof File) { + return ((File) message.getPayload()).getName(); + }else { + throw new IllegalArgumentException("File expected as payload."); + } + }); + return handler; + } + + @MessagingGateway + public interface LimGateway { + @Gateway(requestChannel = "toSftpChannel") + void sendToSftp(File file); + } +} diff --git a/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/config/settings/Common.java b/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/config/settings/Common.java new file mode 100644 index 000000000..28dab5aa9 --- /dev/null +++ b/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/config/settings/Common.java @@ -0,0 +1,32 @@ +package ru.spcex.clearing.lim.exporter.config.settings; + +public class Common { + + private String encoding; + private int insertBatchSize; + private int threadsCount; + + public String getEncoding() { + return encoding; + } + + public void setEncoding(String encoding) { + this.encoding = encoding; + } + + public int getInsertBatchSize() { + return insertBatchSize; + } + + public void setInsertBatchSize(int insertBatchSize) { + this.insertBatchSize = insertBatchSize; + } + + public int getThreadsCount() { + return threadsCount; + } + + public void setThreadsCount(int threadsCount) { + this.threadsCount = threadsCount; + } +} diff --git a/clearing-parent/lim-exporter/src/main/java/ru/clearing/lim/exporter/config/settings/ExportLimServiceSettings.java b/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/config/settings/ExportLimServiceSettings.java similarity index 77% rename from clearing-parent/lim-exporter/src/main/java/ru/clearing/lim/exporter/config/settings/ExportLimServiceSettings.java rename to clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/config/settings/ExportLimServiceSettings.java index ff437e161..9bb6de24a 100644 --- a/clearing-parent/lim-exporter/src/main/java/ru/clearing/lim/exporter/config/settings/ExportLimServiceSettings.java +++ b/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/config/settings/ExportLimServiceSettings.java @@ -1,4 +1,4 @@ -package ru.clearing.lim.exporter.config.settings; +package ru.spcex.clearing.lim.exporter.config.settings; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.PropertySource; @@ -9,12 +9,13 @@ import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams; @Component @PropertySource("file:${spring.config.location}/application.properties") -@ConfigurationProperties("export-dbf-service") +@ConfigurationProperties("export-lim-service") public class ExportLimServiceSettings { private HazelcastClientParams hazelcast; private KafkaConsumerSettings kafkaConsumer; private KafkaProducerSettings kafkaProducer; - private Store store; + private Store sFTPStore; + private Common common; public HazelcastClientParams getHazelcast() { return hazelcast; @@ -33,11 +34,11 @@ public class ExportLimServiceSettings { } public Store getStore() { - return store; + return sFTPStore; } - public void setStore(Store store) { - this.store = store; + public void setStore(Store Store) { + this.sFTPStore = Store; } public KafkaProducerSettings getKafkaProducer() { @@ -47,4 +48,12 @@ public class ExportLimServiceSettings { public void setKafkaProducer(KafkaProducerSettings kafkaProducer) { this.kafkaProducer = kafkaProducer; } + + public Common getCommon() { + return common; + } + + public void setCommon(Common common) { + this.common = common; + } } diff --git a/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/config/settings/Store.java b/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/config/settings/Store.java new file mode 100644 index 000000000..ec4c6012f --- /dev/null +++ b/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/config/settings/Store.java @@ -0,0 +1,50 @@ +package ru.spcex.clearing.lim.exporter.config.settings; + +public class Store { + + private String outDir; + private String user; + private String password; + private String serverIp; + private int serverPort; + + 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; + } +} diff --git a/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/services/AbstractExporterService.java b/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/services/AbstractExporterService.java new file mode 100644 index 000000000..009c28cfc --- /dev/null +++ b/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/services/AbstractExporterService.java @@ -0,0 +1,48 @@ +package ru.spcex.clearing.lim.exporter.services; + +import ru.spcex.clearing.lim.exporter.config.SFTPConfig; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.time.LocalDateTime; +import java.time.format.DateTimeFormatter; +import java.util.Collection; + +public abstract class AbstractExporterService { + + private static final DateTimeFormatter dtFormatter = DateTimeFormatter.ofPattern("yyyyMMddHHmmss"); + private final SFTPConfig.LimGateway gateway; + + protected AbstractExporterService(SFTPConfig.LimGateway gateway) { + this.gateway = gateway; + } + + public abstract Collection getLimFileRows(); + public void process() { + String fileName = prepareFileName("security"); + File limFile = new File(fileName); + + Path limFilePath = limFile.toPath(); + try { + Files.write(limFilePath, getLimFileRows()); + gateway.sendToSftp(limFile); + Files.deleteIfExists(limFilePath); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + private String prepareFileName(String target) { + StringBuilder result = new StringBuilder(); + String dt = dtFormatter.format(LocalDateTime.now()); + + result.append("limits_"); + result.append(target); + result.append('_'); + result.append(dt); + result.append(".lim"); + return result.toString(); + } +} diff --git a/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/services/LauncherCommandReceiver.java b/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/services/LauncherCommandReceiver.java new file mode 100644 index 000000000..660a4c287 --- /dev/null +++ b/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/services/LauncherCommandReceiver.java @@ -0,0 +1,36 @@ +package ru.spcex.clearing.lim.exporter.services; + +import org.apache.kafka.clients.consumer.Consumer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.stereotype.Service; +import ru.spcex.clearing.platform.messaging.domain.cud.schedule.LauncherCommandRequest; +import ru.spcex.clearing.platform.messaging.service.QueueConsumer; +import ru.spcex.platform.enumeration.Task; + +@Service +public class LauncherCommandReceiver extends QueueConsumer implements InitializingBean { + private final Logger log = LoggerFactory.getLogger(getClass()); + private final MoneyExporterService moneyExporterService; + private final SecurityExporterService securityExporterService; + + public LauncherCommandReceiver(Consumer kafkaQueue, + MoneyExporterService moneyExporterService, + SecurityExporterService securityExporterService) { + super(kafkaQueue); + this.moneyExporterService = moneyExporterService; + this.securityExporterService = securityExporterService; + } + + @Override + public void afterPropertiesSet() { + callback(LauncherCommandRequest.class) + .setConsumer(action -> moneyExporterService.process()) + .forDestination(Task.unloadingSession_LIMM.topic(), callbacks::put); // LIMM + callback(LauncherCommandRequest.class) + .setConsumer(action -> securityExporterService.process()) + .forDestination(Task.unloadingSession_LIMS.topic(), callbacks::put); // LIMS + init(); + } +} diff --git a/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/services/MoneyExporterService.java b/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/services/MoneyExporterService.java new file mode 100644 index 000000000..2e6272f7e --- /dev/null +++ b/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/services/MoneyExporterService.java @@ -0,0 +1,21 @@ +package ru.spcex.clearing.lim.exporter.services; + +import org.springframework.stereotype.Service; + +import java.util.Arrays; +import java.util.Collection; +import java.util.List; + +@Service +public class MoneyExporterService extends AbstractExporterService { + + public MoneyExporterService() { + super(gateway); + } + + @Override + public Collection getLimFileRows(){ + List limFileRows = Arrays.asList("a", "b", "c"); + return limFileRows; + } +} diff --git a/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/services/SecurityExporterService.java b/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/services/SecurityExporterService.java new file mode 100644 index 000000000..59a8dd69a --- /dev/null +++ b/clearing-parent/lim-exporter/src/main/java/ru/spcex/clearing/lim/exporter/services/SecurityExporterService.java @@ -0,0 +1,20 @@ +package ru.spcex.clearing.lim.exporter.services; + +import org.springframework.stereotype.Service; + +import java.util.Arrays; +import java.util.Collection; +import java.util.List; + +@Service +public class SecurityExporterService extends AbstractExporterService { + public SecurityExporterService() { + super(gateway); + } + + @Override + public Collection getLimFileRows(){ + List limFileRows = Arrays.asList("a", "b", "c"); + return limFileRows; + } +} diff --git a/clearing-parent/lim-exporter/src/main/resources/application.properties b/clearing-parent/lim-exporter/src/main/resources/application.properties index 471634fbb..114f0954c 100644 --- a/clearing-parent/lim-exporter/src/main/resources/application.properties +++ b/clearing-parent/lim-exporter/src/main/resources/application.properties @@ -7,7 +7,15 @@ export-lim-service.hazelcast.password=dev-pass export-lim-service.common.encoding=cp866 export-lim-service.common.threads-count=10 -export-lim-service.store.out-dir=d:\\trash\\clearing\\exporter\\out\\ +export-lim-service.store.out-dir=DocOut +export-lim-service.store.user:tester +export-lim-service.store.password=password +export-lim-service.store.server-ip=10.230.238.53 +export-lim-service.store.server-port=2222 + +export-lim-service.hazelcast.cluster-members=127.0.0.1:5701 +export-lim-service.hazelcast.login=dev +export-lim-service.hazelcast.password=dev-pass export-lim-service.kafka-consumer.bootstrap-servers=localhost:9092 export-lim-service.kafka-consumer.group-id=dev-group-balance-service @@ -16,3 +24,10 @@ export-lim-service.kafka-consumer.session-timeout-ms=30000 export-lim-service.kafka-consumer.auto-offset-reset=latest export-lim-service.kafka-consumer.linger-ms=1 export-lim-service.kafka-consumer.buffer-memory=33554432 + +export-lim-service.kafka-producer.bootstrap-servers=localhost:9092 +export-lim-service.kafka-producer.acks=all +export-lim-service.kafka-producer.retries=0 +export-lim-service.kafka-producer.batch-size=16384 +export-lim-service.kafka-producer.linger-ms=1 +export-lim-service.kafka-producer.buffer-memory=33554432 \ No newline at end of file diff --git a/platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/Task.java b/platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/Task.java index 678b0acf5..e6b512d58 100644 --- a/platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/Task.java +++ b/platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/Task.java @@ -16,7 +16,8 @@ public enum Task implements IEnumKey { createOrderConfirm("CORC"), getAllBalance("GALB"), createReport_GREP("GREP"), // Создание отчёта (report-service) RPRT нескольких видов, этот GREP - unloadingSession_LIMM("LIMM"),//Выгрузка в торговую систему остатков секции МКР + unloadingSession_LIMM("LIMM"),//Выгрузка в торговую систему остатков по деньгам + unloadingSession_LIMS("LIMS"),//Выгрузка в торговую систему остатков по бумагам unloadingSession_LIMF("LIMF"),//Выгрузка в торговую систему остатков Фондовой секции liquidationSession_LIQU("LIQU"),//Ликвидационная сессия по обязательтсвам участника startSession_STRM("STRM"),//Начало торговой сессии секции МКР