add consumer to export service

clearing-parent/dbf-exporter/pom.xml

@@ -41,6 +41,10 @@
             <groupId>ru.spcex.platform</groupId>
             <artifactId>platform-imdg-api-hazelcast-impl</artifactId>
         </dependency>
+        <dependency>
+            <groupId>ru.spcex.platform</groupId>
+            <artifactId>platform-messaging</artifactId>
+        </dependency>
         <dependency>
             <groupId>ru.spcex.clearing</groupId>
             <artifactId>classes</artifactId>

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.setClusterMembers(settings.getHazelcast().getClusterMembers());
-        params.setLogin(properties.getHazelcastLogin());
+        params.setLogin(settings.getHazelcast().getLogin());
-        params.setPassword(properties.getHazelcastPassword());
+        params.setPassword(settings.getHazelcast().getPassword());
-//        params.setInstanceName("dbf-exporter");
         return new HazelcastService(taskExecutorHazelcastClientInitializer, taskExecutorIdGeneratorAwaiter, params);
     }
 
     @Bean("pipeline")
-    public List<Stage> pipeline() {
+    public List<Stage> pipeline(ApplicationContext context) {
         List<Stage> 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.setMaxPoolSize(settings.getCommon().getThreadsCount());
-        executor.setCorePoolSize(properties.getThreadsCount());
+        executor.setCorePoolSize(settings.getCommon().getThreadsCount());
-        executor.setThreadNamePrefix("dbf-exporter-thread-");
+        executor.setThreadNamePrefix("dbf-exporter");
         executor.setWaitForTasksToCompleteOnShutdown(true);
         executor.setAwaitTerminationSeconds(300);
         executor.initialize();

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<String, Object> createConsumer(ExportDBFServiceSettings settings) {
+        return KafkaConsumerFactory.consumer(settings.getKafkaConsumer());
+    }
+}

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;
+    }
+}

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;
+    }
+}

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;
+    }
+}

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;
+    }
+}

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<Table, DBFField[]> 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);
         }
     }
 

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) {
+    public PrepareDBFFile(ExportDBFServiceSettings settings) {
-        this.properties = properties;
+        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.");

clearing-parent/dbf-exporter/src/main/java/ru/spcex/clearing/dbf/exporter/properties/AProperties.java

@@ -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;
-    }
-}

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<String, Object> 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<ExportToFileRequest> systemRequest) {
+
+    }
+}

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
+export-dbf-service.hazelcast.cluster-members=10.200.200.181:5071
-hazelcast.login=dev
+export-dbf-service.hazelcast.login=dev
-hazelcast.password=dev-pass
+export-dbf-service.hazelcast.password=dev-pass
 
-dbf.encoding=cp866
+export-dbf-service.common.encoding=cp866
-dbf.threads-count=10
+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

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;
+    }
+}