diff --git a/clearing-parent/dbf-exporter/pom.xml b/clearing-parent/dbf-exporter/pom.xml index 5fd93f86d..fc2108169 100644 --- a/clearing-parent/dbf-exporter/pom.xml +++ b/clearing-parent/dbf-exporter/pom.xml @@ -41,6 +41,10 @@ ru.spcex.platform platform-imdg-api-hazelcast-impl + + ru.spcex.platform + platform-messaging + ru.spcex.clearing classes diff --git a/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/DBFExporterConfig.java b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/DBFExporterConfig.java index ba9acb1ef..10261ad81 100644 --- a/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/DBFExporterConfig.java +++ b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/DBFExporterConfig.java @@ -7,10 +7,10 @@ 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.dbf.exporter.config.settings.ExportDBFServiceSettings; import ru.spcex.clearing.dbf.exporter.logic.stages.ExportFromHazelcast; import ru.spcex.clearing.dbf.exporter.logic.stages.PrepareDBFFile; import ru.spcex.clearing.dbf.exporter.logic.stages.Stage; -import ru.spcex.clearing.dbf.exporter.properties.AProperties; 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; @@ -22,13 +22,6 @@ import java.util.List; @EnableConfigurationProperties @ComponentScan(basePackages = {"ru.spcex.clearing.dbf.exporter"}) public class DBFExporterConfig { - private final AProperties properties; - private final ApplicationContext context; - - public DBFExporterConfig(@Qualifier("dbfExporterProperties") AProperties properties, ApplicationContext context) { - this.properties = properties; - this.context = context; - } @Bean("taskExecutorHazelcastClientInitializer") public ThreadPoolTaskExecutor taskExecutorHazelcastClientInitializer() { @@ -43,17 +36,16 @@ public class DBFExporterConfig { @Bean("imdgProvider") public ImdgProvider imdgProvider(@Qualifier("taskExecutorHazelcastClientInitializer") ThreadPoolTaskExecutor taskExecutorHazelcastClientInitializer, @Qualifier("taskExecutorIdGeneratorAwaiter") ThreadPoolTaskExecutor taskExecutorIdGeneratorAwaiter, - AProperties properties) { + ExportDBFServiceSettings settings) { HazelcastClientParams params = new HazelcastClientParams(); - params.setClusterMembers(properties.getHazelcastClusterMembers()); - params.setLogin(properties.getHazelcastLogin()); - params.setPassword(properties.getHazelcastPassword()); -// params.setInstanceName("dbf-exporter"); + params.setClusterMembers(settings.getHazelcast().getClusterMembers()); + params.setLogin(settings.getHazelcast().getLogin()); + params.setPassword(settings.getHazelcast().getPassword()); return new HazelcastService(taskExecutorHazelcastClientInitializer, taskExecutorIdGeneratorAwaiter, params); } @Bean("pipeline") - public List pipeline() { + public List pipeline(ApplicationContext context) { List pipeline = new LinkedList<>(); pipeline.add(context.getBean(PrepareDBFFile.class)); @@ -63,11 +55,11 @@ public class DBFExporterConfig { } @Bean("executor") - public ThreadPoolTaskExecutor executor() { + public ThreadPoolTaskExecutor executor(ExportDBFServiceSettings settings) { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); - executor.setMaxPoolSize(properties.getThreadsCount()); - executor.setCorePoolSize(properties.getThreadsCount()); - executor.setThreadNamePrefix("dbf-exporter-thread-"); + executor.setMaxPoolSize(settings.getCommon().getThreadsCount()); + executor.setCorePoolSize(settings.getCommon().getThreadsCount()); + executor.setThreadNamePrefix("dbf-exporter"); executor.setWaitForTasksToCompleteOnShutdown(true); executor.setAwaitTerminationSeconds(300); executor.initialize(); diff --git a/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/KafkaConfig.java b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/KafkaConfig.java new file mode 100644 index 000000000..73840d3e5 --- /dev/null +++ b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/KafkaConfig.java @@ -0,0 +1,21 @@ +package ru.spcex.clearing.dbf.exporter.config; + +import org.apache.kafka.clients.consumer.Consumer; +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.dbf.exporter.config.settings.ExportDBFServiceSettings; +import ru.spcex.clearing.platform.messaging.config.KafkaConsumerFactory; + +@Configuration +public class KafkaConfig { + + @Autowired + @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE) + @Bean + public Consumer createConsumer(ExportDBFServiceSettings settings) { + return KafkaConsumerFactory.consumer(settings.getKafkaConsumer()); + } +} diff --git a/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/settings/Common.java b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/settings/Common.java new file mode 100644 index 000000000..1d2fbf854 --- /dev/null +++ b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/settings/Common.java @@ -0,0 +1,32 @@ +package ru.spcex.clearing.dbf.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/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/settings/Cron.java b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/settings/Cron.java new file mode 100644 index 000000000..ff7f8bda8 --- /dev/null +++ b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/settings/Cron.java @@ -0,0 +1,14 @@ +package ru.spcex.clearing.dbf.exporter.config.settings; + +public class Cron { + + private String checkSrcDirCron; + + public String getCheckSrcDirCron() { + return checkSrcDirCron; + } + + public void setCheckSrcDirCron(String checkSrcDirCron) { + this.checkSrcDirCron = checkSrcDirCron; + } +} diff --git a/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/settings/ExportDBFServiceSettings.java b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/settings/ExportDBFServiceSettings.java new file mode 100644 index 000000000..8166f7931 --- /dev/null +++ b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/settings/ExportDBFServiceSettings.java @@ -0,0 +1,58 @@ +package ru.spcex.clearing.dbf.exporter.config.settings; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.context.annotation.PropertySource; +import org.springframework.stereotype.Component; +import ru.spcex.clearing.platform.messaging.config.element.KafkaConsumerSettings; +import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams; + +@Component +@PropertySource("file:${spring.config.location}/application.properties") +@ConfigurationProperties("export-dbf-service") +public class ExportDBFServiceSettings { + private HazelcastClientParams hazelcast; + private KafkaConsumerSettings kafkaConsumer; + private Common common; + private Store store; + private Cron cron; + + public HazelcastClientParams getHazelcast() { + return hazelcast; + } + + public void setHazelcast(HazelcastClientParams hazelcast) { + this.hazelcast = hazelcast; + } + + public KafkaConsumerSettings getKafkaConsumer() { + return kafkaConsumer; + } + + public void setKafkaConsumer(KafkaConsumerSettings kafkaConsumer) { + this.kafkaConsumer = kafkaConsumer; + } + + public Common getCommon() { + return common; + } + + public void setCommon(Common common) { + this.common = common; + } + + public Store getStore() { + return store; + } + + public void setStore(Store store) { + this.store = store; + } + + public Cron getCron() { + return cron; + } + + public void setCron(Cron cron) { + this.cron = cron; + } +} diff --git a/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/settings/Store.java b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/settings/Store.java new file mode 100644 index 000000000..e1430418a --- /dev/null +++ b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/config/settings/Store.java @@ -0,0 +1,14 @@ +package ru.spcex.clearing.dbf.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/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/stages/ExportFromHazelcast.java b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/stages/ExportFromHazelcast.java index 01be79565..f6318249d 100644 --- a/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/stages/ExportFromHazelcast.java +++ b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/stages/ExportFromHazelcast.java @@ -6,11 +6,11 @@ import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.stereotype.Component; import ru.clearing.classes.statics.data.sdf.*; +import ru.spcex.clearing.dbf.exporter.config.settings.ExportDBFServiceSettings; import ru.spcex.clearing.dbf.exporter.exceptions.ConfigException; import ru.spcex.clearing.dbf.exporter.logic.data.ResultContainer; import ru.spcex.clearing.dbf.exporter.logic.data.enums.StageResult; import ru.spcex.clearing.dbf.exporter.logic.data.enums.Table; -import ru.spcex.clearing.dbf.exporter.properties.AProperties; import ru.spcex.clearing.dbf.exporter.services.converters.*; import ru.spcex.platform.classes.base.SpcexObjectBase; import ru.spcex.platform.imdg.api.Imdg; @@ -28,7 +28,7 @@ import java.util.Objects; */ @Component public class ExportFromHazelcast extends Stage implements InitializingBean { - private final AProperties properties; + private final ExportDBFServiceSettings settings; private final ImdgProvider imdgProvider; private final S_DF02_Converter s_df02_converter; @@ -40,14 +40,14 @@ public class ExportFromHazelcast extends Stage implements InitializingBean { private final Map dbfFieldsForTable = new HashMap<>(); private Charset dbfCharset; - public ExportFromHazelcast(@Qualifier("dbfExporterProperties") AProperties properties, + public ExportFromHazelcast(ExportDBFServiceSettings settings, @Qualifier("imdgProvider") ImdgProvider imdgProvider, S_DF02_Converter s_df02_converter, S_DF08_Converter s_df08_converter, S_DF18_Converter s_df18_converter, S_DF10_Converter s_df10_converter, S_DF17_Converter s_df17_converter) { - this.properties = properties; + this.settings = settings; this.imdgProvider = imdgProvider; this.s_df02_converter = s_df02_converter; this.s_df08_converter = s_df08_converter; @@ -123,9 +123,9 @@ public class ExportFromHazelcast extends Stage implements InitializingBean { private void initDBFCharset() { try { - dbfCharset = Charset.forName(properties.getDbfEncoding()); + dbfCharset = Charset.forName(settings.getCommon().getEncoding()); } catch (Exception e) { - throw new ConfigException("В properties файле содержится неизвестная кодировка: " + properties.getDbfEncoding(), e); + throw new ConfigException("В properties файле содержится неизвестная кодировка: " + settings.getCommon().getEncoding(), e); } } diff --git a/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/stages/PrepareDBFFile.java b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/stages/PrepareDBFFile.java index a1a099489..a0743322c 100644 --- a/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/stages/PrepareDBFFile.java +++ b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/logic/stages/PrepareDBFFile.java @@ -1,12 +1,11 @@ package ru.spcex.clearing.dbf.exporter.logic.stages; import org.springframework.beans.factory.InitializingBean; -import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.stereotype.Component; +import ru.spcex.clearing.dbf.exporter.config.settings.ExportDBFServiceSettings; import ru.spcex.clearing.dbf.exporter.logic.data.ResultContainer; import ru.spcex.clearing.dbf.exporter.logic.data.enums.StageResult; import ru.spcex.clearing.dbf.exporter.logic.data.enums.Table; -import ru.spcex.clearing.dbf.exporter.properties.AProperties; import java.io.File; import java.io.IOException; @@ -26,11 +25,11 @@ public class PrepareDBFFile extends Stage implements InitializingBean { private static final String CODE_OF_MEMBER = null; private static final DateTimeFormatter tsFormatter = DateTimeFormatter.ofPattern("yyMMddHHmm"); private static final DateTimeFormatter utilFormatter = DateTimeFormatter.ofPattern("yyMMdd"); - private final AProperties properties; + private final ExportDBFServiceSettings settings; private String outDir; - public PrepareDBFFile(@Qualifier("dbfExporterProperties") AProperties properties) { - this.properties = properties; + public PrepareDBFFile(ExportDBFServiceSettings settings) { + this.settings = settings; } @Override @@ -54,7 +53,7 @@ public class PrepareDBFFile extends Stage implements InitializingBean { @Override public void afterPropertiesSet() throws Exception { - String outDirPath = properties.getOutDir(); + String outDirPath = settings.getStore().getOutDir(); File outDirFile = new File(outDirPath); if (outDirFile.exists() && !outDirFile.isDirectory()) throw new IOException("Output directory " + outDirPath + " is file."); diff --git a/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/properties/AProperties.java b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/properties/AProperties.java deleted file mode 100644 index 46ee82324..000000000 --- a/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/properties/AProperties.java +++ /dev/null @@ -1,74 +0,0 @@ -package ru.spcex.clearing.dbf.exporter.properties; - -import org.springframework.beans.factory.annotation.Value; -import org.springframework.stereotype.Component; - -@Component("dbfExporterProperties") -public class AProperties { - - @Value("${hazelcast.cluster-members}") - private String hazelcastClusterMembers; - - @Value("${hazelcast.login}") - private String hazelcastLogin; - - @Value("${hazelcast.password}") - private String hazelcastPassword; - - @Value("${dbf.out-dir}") - private String outDir; - - @Value("${dbf.encoding}") - private String dbfEncoding; - - @Value("${dbf.threads-count}") - private int threadsCount; - - public String getHazelcastClusterMembers() { - return hazelcastClusterMembers; - } - - public void setHazelcastClusterMembers(String hazelcastClusterMembers) { - this.hazelcastClusterMembers = hazelcastClusterMembers; - } - - public String getHazelcastLogin() { - return hazelcastLogin; - } - - public void setHazelcastLogin(String hazelcastLogin) { - this.hazelcastLogin = hazelcastLogin; - } - - public String getHazelcastPassword() { - return hazelcastPassword; - } - - public void setHazelcastPassword(String hazelcastPassword) { - this.hazelcastPassword = hazelcastPassword; - } - - public String getOutDir() { - return outDir; - } - - public void setOutDir(String outDir) { - this.outDir = outDir; - } - - public String getDbfEncoding() { - return dbfEncoding; - } - - public void setDbfEncoding(String dbfEncoding) { - this.dbfEncoding = dbfEncoding; - } - - public int getThreadsCount() { - return threadsCount; - } - - public void setThreadsCount(int threadsCount) { - this.threadsCount = threadsCount; - } -} diff --git a/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/services/CommandService.java b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/services/CommandService.java new file mode 100644 index 000000000..87fdc1fa1 --- /dev/null +++ b/clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/services/CommandService.java @@ -0,0 +1,33 @@ +package ru.spcex.clearing.dbf.exporter.services; + +import org.apache.kafka.clients.consumer.Consumer; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.stereotype.Service; +import ru.spcex.clearing.platform.messaging.domain.BaseRequest; +import ru.spcex.clearing.platform.messaging.domain.Consts; +import ru.spcex.clearing.platform.messaging.domain.cud.balance.ExportToFileRequest; +import ru.spcex.clearing.platform.messaging.service.QueueConsumer; +import ru.spcex.platform.imdg.api.ImdgProvider; + +@Service +public class CommandService extends QueueConsumer implements InitializingBean { + + private final ImdgProvider imdgProvider; + + public CommandService(Consumer kafkaQueue, ImdgProvider imdgProvider) { + super(kafkaQueue); + this.imdgProvider = imdgProvider; + } + + @Override + public void afterPropertiesSet() throws Exception { + callback(ExportToFileRequest.class) + .setConsumer(this::process) + .forDestination(Consts.STATEMENT_PROCESS, callbacks::put); + init(); + } + + private void process(BaseRequest systemRequest) { + + } +} diff --git a/clearing-parent/dbf-exporter/src/main/resources/application.properties b/clearing-parent/dbf-exporter/src/main/resources/application.properties index 0a57267ef..deda8d69c 100644 --- a/clearing-parent/dbf-exporter/src/main/resources/application.properties +++ b/clearing-parent/dbf-exporter/src/main/resources/application.properties @@ -2,11 +2,18 @@ server.port=8080 server.servlet.context-path=/exporter spring.main.web-application-type=servlet -hazelcast.cluster-members=127.0.0.1:5071 -hazelcast.login=dev -hazelcast.password=dev-pass +export-dbf-service.hazelcast.cluster-members=10.200.200.181:5071 +export-dbf-service.hazelcast.login=dev +export-dbf-service.hazelcast.password=dev-pass -dbf.encoding=cp866 -dbf.threads-count=10 +export-dbf-service.common.encoding=cp866 +export-dbf-service.common.threads-count=10 -dbf.out-dir=/opt/clearing/file/exporter/ +export-dbf-service.store.out-dir=/opt/clearing/file/exporter/ + +import-dbf-service.kafka-consumer.bootstrap-servers=localhost:9092 +import-dbf-service.kafka-consumer.acks=all +import-dbf-service.kafka-consumer.retries=0 +import-dbf-service.kafka-consumer.batch-size=16384 +import-dbf-service.kafka-consumer.linger-ms=1 +import-dbf-service.kafka-consumer.buffer-memory=33554432 \ No newline at end of file diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/balance/ExportToFileRequest.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/balance/ExportToFileRequest.java new file mode 100644 index 000000000..5b196e20d --- /dev/null +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/balance/ExportToFileRequest.java @@ -0,0 +1,26 @@ +package ru.spcex.clearing.platform.messaging.domain.cud.balance; + +import com.fasterxml.jackson.annotation.JsonProperty; + +public class ExportToFileRequest { + @JsonProperty + private Long sdfGroupId; + @JsonProperty + private String nameOfTable; + + public Long getSdfGroupId() { + return sdfGroupId; + } + + public void setSdfGroupId(Long sdfGroupId) { + this.sdfGroupId = sdfGroupId; + } + + public String getNameOfTable() { + return nameOfTable; + } + + public void setNameOfTable(String nameOfTable) { + this.nameOfTable = nameOfTable; + } +} \ No newline at end of file