http://jira.mfd.msk:8088/browse/CLS-758 kafka support added, new tasks created, added opencsv, controllers removed

clearing-parent/control-service/pom.xml

@@ -16,14 +16,15 @@
     </parent>
 
     <properties>
+        <opencsv.version>5.5.2</opencsv.version>
         <logstash-logback-encoder.version>7.0.1</logstash-logback-encoder.version>
     </properties>
 
     <dependencies>
-        <!-- Spring boot -->
+<!--         Spring boot -->
         <dependency>
             <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-web</artifactId>
+            <artifactId>spring-boot-starter</artifactId>
         </dependency>
 
         <!-- Database -->
@@ -40,6 +41,23 @@
             <groupId>com.mchange</groupId>
             <artifactId>c3p0</artifactId>
         </dependency>
+        <dependency>
+            <groupId>ru.spcex.platform</groupId>
+            <artifactId>platform-imdg-api-hazelcast-impl</artifactId>
+        </dependency>
+
+        <!-- CSV -->
+        <dependency>
+            <groupId>com.opencsv</groupId>
+            <artifactId>opencsv</artifactId>
+            <version>${opencsv.version}</version>
+        </dependency>
+
+        <!-- Kafka -->
+        <dependency>
+            <groupId>ru.spcex.platform</groupId>
+            <artifactId>platform-messaging</artifactId>
+        </dependency>
 
         <!-- special logging -->
         <dependency>
@@ -47,21 +65,24 @@
             <artifactId>logstash-logback-encoder</artifactId>
             <version>${logstash-logback-encoder.version}</version>
         </dependency>
-        <dependency>
-            <groupId>com.fasterxml.jackson.core</groupId>
-            <artifactId>jackson-databind</artifactId>
-        </dependency>
 
+        <!-- Miscellaneous -->
         <dependency>
             <groupId>ru.spcex.platform</groupId>
             <artifactId>platform-enum</artifactId>
         </dependency>
 
+        <!-- Test dependencies -->
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-test</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>ru.spcex.clearing</groupId>
+            <artifactId>test-clearing</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/config/ImdgConfig.java

@@ -0,0 +1,58 @@
+package ru.spcex.clearing.control.config;
+
+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.control.config.settings.ControlServiceSettings;
+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
+public class ImdgConfig {
+    @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,
+                                     ControlServiceSettings 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(ControlServiceSettings settings) {
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        executor.setMaxPoolSize(10);
+        executor.setCorePoolSize(10);
+        executor.setThreadNamePrefix("control-service");
+        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;
+    }
+
+}

clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/config/KafkaConfig.java

@@ -0,0 +1,30 @@
+package ru.spcex.clearing.control.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.control.config.settings.ControlServiceSettings;
+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<String, Object> createConsumer(ControlServiceSettings settings) {
+        return KafkaConsumerFactory.consumer(settings.getKafkaConsumer());
+    }
+
+    @Autowired
+    @Bean
+    public Producer<String, Object> createProducer(ControlServiceSettings settings) {
+        return settings.getKafkaProducer() != null ?
+            KafkaProducerFactory.producer(settings.getKafkaProducer()) :
+            null;
+    }
+}

clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/config/KafkaSenderConfig.java

@@ -0,0 +1,57 @@
+package ru.spcex.clearing.control.config;
+
+import java.util.function.Supplier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+import ru.spcex.clearing.control.config.settings.ControlServiceSettings;
+import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory;
+import ru.spcex.clearing.platform.messaging.config.element.KafkaProducerSettings;
+import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
+import ru.spcex.platform.imdg.api.ImdgId;
+import ru.spcex.platform.imdg.api.ImdgProvider;
+
+@Configuration
+public class KafkaSenderConfig {
+    Logger log = LoggerFactory.getLogger(getClass());
+
+    private final ImdgProvider imdgProvider;
+
+    @Autowired
+    public KafkaSenderConfig(ImdgProvider imdgProvider) {
+        this.imdgProvider = imdgProvider;
+    }
+
+    @Bean
+    public ProducerFactory<String, Object> pf(ControlServiceSettings settings) {
+        if (settings.getKafkaProducer() == null) {
+            return null;
+        }
+        KafkaProducerSettings kafkaSettings = settings.getKafkaProducer();
+        return KafkaProducerFactory.producerFactory(kafkaSettings);
+    }
+
+    @Bean("kafkaTemplate")
+    public KafkaTemplate<String, Object> kafkaTemplate(ProducerFactory<String, Object> pf) {
+        if (pf == null) {
+            return null;
+        }
+        return new KafkaTemplate<>(pf);
+    }
+
+    @Bean
+    public Supplier<KafkaSender> kafkaSenderSupplier(KafkaTemplate<String, Object> kafkaTemplate,
+                                                     ImdgProvider imdgProvider) {
+        ImdgId imdgIdGenerator = imdgProvider.getImdgIdGenerator();
+        return () -> KafkaSender
+            .setup()
+            .setKafkaTemplate(kafkaTemplate)
+            .idGenerator(imdgIdGenerator::nextId)
+            .saveRequestInfo(false)
+            .build();
+    }
+}

clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/config/settings/ControlServiceSettings.java

@@ -3,6 +3,9 @@ package ru.spcex.clearing.control.config.settings;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.NestedConfigurationProperty;
 import org.springframework.stereotype.Component;
+import ru.spcex.clearing.platform.messaging.config.element.KafkaConsumerSettings;
+import ru.spcex.clearing.platform.messaging.config.element.KafkaProducerSettings;
+import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams;
 
 @Component
 @ConfigurationProperties("control-service")
@@ -10,6 +13,14 @@ public class ControlServiceSettings {
     @NestedConfigurationProperty
     private Database database;
 
+    @NestedConfigurationProperty
+    private HazelcastClientParams hazelcast;
+
+    @NestedConfigurationProperty
+    private KafkaConsumerSettings kafkaConsumer;
+    @NestedConfigurationProperty
+    private KafkaProducerSettings kafkaProducer;
+
     private String resultDirectory;
 
     public Database getDatabase() {
@@ -20,6 +31,30 @@ public class ControlServiceSettings {
         this.database = database;
     }
 
+    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 KafkaProducerSettings getKafkaProducer() {
+        return kafkaProducer;
+    }
+
+    public void setKafkaProducer(KafkaProducerSettings kafkaProducer) {
+        this.kafkaProducer = kafkaProducer;
+    }
+
     public String getResultDirectory() {
         return resultDirectory;
     }

clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/controller/ControlController.java

@@ -1,33 +0,0 @@
-package ru.spcex.clearing.control.controller;
-
-import org.springframework.web.bind.annotation.GetMapping;
-import org.springframework.web.bind.annotation.PathVariable;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RestController;
-import ru.spcex.clearing.control.service.ControlService;
-import ru.spcex.platform.enumeration.SessionType;
-
-@RestController
-@RequestMapping("/api/v1/control")
-public class ControlController {
-    private final ControlService controlService;
-
-    public ControlController(ControlService controlService) {
-        this.controlService = controlService;
-    }
-
-    @GetMapping("/general-control/{sessionType}")
-    public String generalControl(@PathVariable("sessionType") SessionType sessionType) {
-        return controlService.generalControl(sessionType);
-    }
-
-    @GetMapping("/internal-control/{sessionId}")
-    public String internalControl(@PathVariable("sessionId") Long sessionId) {
-        return controlService.internalControl(sessionId);
-    }
-
-    @GetMapping("/internal-control-finl/{sessionId}")
-    public String internalControlFinl(@PathVariable("sessionId") Long sessionId) {
-        return controlService.internalControlFinl(sessionId);
-    }
-}

clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/model/ResultContainer.java

@@ -1,4 +1,142 @@
 package ru.spcex.clearing.control.model;
 
-public record ResultContainer(Long firstTable, Long secondTable, Long difference) {
+import com.opencsv.bean.CsvBindByName;
+import com.opencsv.bean.CsvBindByPosition;
+import java.util.Objects;
+import ru.spcex.platform.enumeration.SessionType;
+
+public class ResultContainer {
+    @CsvBindByName(column = "sessionId")
+    @CsvBindByPosition(position = 0)
+    private Long sessionId;
+
+    @CsvBindByName(column = "sessionType")
+    @CsvBindByPosition(position = 1)
+    private SessionType sessionType;
+
+    @CsvBindByName(column = "firstTableTitle")
+    @CsvBindByPosition(position = 2)
+    private String firstTableTitle;
+
+    @CsvBindByName(column = "firstTableRecordQuantity")
+    @CsvBindByPosition(position = 3)
+    private Long firstTableRecordQuantity;
+
+    @CsvBindByName(column = "secondTableRecordTitle")
+    @CsvBindByPosition(position = 4)
+    private String secondTableRecordTitle;
+
+    @CsvBindByName(column = "secondTableRecordQuantity")
+    @CsvBindByPosition(position = 5)
+    private Long secondTableRecordQuantity;
+
+    @CsvBindByName(column = "quantityDifference")
+    @CsvBindByPosition(position = 6)
+    private Long quantityDifference;
+
+    public ResultContainer(Long sessionId,
+                           SessionType sessionType,
+                           String firstTableTitle,
+                           Long firstTableRecordQuantity,
+                           String secondTableRecordTitle,
+                           Long secondTableRecordQuantity,
+                           Long quantityDifference) {
+        this.sessionId = sessionId;
+        this.sessionType = sessionType;
+        this.firstTableTitle = firstTableTitle;
+        this.firstTableRecordQuantity = firstTableRecordQuantity;
+        this.secondTableRecordTitle = secondTableRecordTitle;
+        this.secondTableRecordQuantity = secondTableRecordQuantity;
+        this.quantityDifference = quantityDifference;
+    }
+
+    public ResultContainer(Long firstTableRecordQuantity,
+                           Long secondTableRecordQuantity,
+                           Long quantityDifference) {
+        this.firstTableRecordQuantity = firstTableRecordQuantity;
+        this.secondTableRecordQuantity = secondTableRecordQuantity;
+        this.quantityDifference = quantityDifference;
+    }
+
+    public Long getSessionId() {
+        return sessionId;
+    }
+
+    public void setSessionId(Long sessionId) {
+        this.sessionId = sessionId;
+    }
+
+    public SessionType getSessionType() {
+        return sessionType;
+    }
+
+    public void setSessionType(SessionType sessionType) {
+        this.sessionType = sessionType;
+    }
+
+    public String getFirstTableTitle() {
+        return firstTableTitle;
+    }
+
+    public void setFirstTableTitle(String firstTableTitle) {
+        this.firstTableTitle = firstTableTitle;
+    }
+
+    public Long getFirstTableRecordQuantity() {
+        return firstTableRecordQuantity;
+    }
+
+    public void setFirstTableRecordQuantity(Long firstTableRecordQuantity) {
+        this.firstTableRecordQuantity = firstTableRecordQuantity;
+    }
+
+    public String getSecondTableRecordTitle() {
+        return secondTableRecordTitle;
+    }
+
+    public void setSecondTableRecordTitle(String secondTableRecordTitle) {
+        this.secondTableRecordTitle = secondTableRecordTitle;
+    }
+
+    public Long getSecondTableRecordQuantity() {
+        return secondTableRecordQuantity;
+    }
+
+    public void setSecondTableRecordQuantity(Long secondTableRecordQuantity) {
+        this.secondTableRecordQuantity = secondTableRecordQuantity;
+    }
+
+    public Long getQuantityDifference() {
+        return quantityDifference;
+    }
+
+    public void setQuantityDifference(Long quantityDifference) {
+        this.quantityDifference = quantityDifference;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        ResultContainer that = (ResultContainer) o;
+        return Objects.equals(sessionId, that.sessionId) && sessionType == that.sessionType && Objects.equals(firstTableTitle, that.firstTableTitle) && Objects.equals(firstTableRecordQuantity, that.firstTableRecordQuantity) && Objects.equals(secondTableRecordTitle, that.secondTableRecordTitle) && Objects.equals(secondTableRecordQuantity, that.secondTableRecordQuantity) && Objects.equals(quantityDifference, that.quantityDifference);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(sessionId, sessionType, firstTableTitle, firstTableRecordQuantity, secondTableRecordTitle, secondTableRecordQuantity, quantityDifference);
+    }
+
+    @Override
+    public String toString() {
+        return "ResultContainer{" +
+            "sessionId=" + sessionId +
+            ", sessionType=" + sessionType +
+            ", firstTableTitle='" + firstTableTitle + '\'' +
+            ", firstTableRecordQuantity=" + firstTableRecordQuantity +
+            ", secondTableRecordTitle='" + secondTableRecordTitle + '\'' +
+            ", secondTableRecordQuantity=" + secondTableRecordQuantity +
+            ", quantityDifference=" + quantityDifference +
+            '}';
+    }
 }

clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/repository/ControlRepository.java

@@ -25,43 +25,61 @@ public class ControlRepository {
     @Transactional
     public ResultContainer getFinlCountDiff() {
         String sqlQuery = """
-            WITH s_trades_count AS (SELECT COUNT(*) FROM s_trades),
+            WITH s_trades_count AS (SELECT COUNT(*) FROM s_trades WHERE CAST(created_at AS DATE) = CAST(NOW() AS DATE)),
-                 execution_deposit_count AS (SELECT COUNT(*) FROM execution_deposit)
+                 execution_deposit_count AS (SELECT COUNT(*) FROM execution_deposit WHERE CAST(created_at AS DATE) = CAST(NOW() AS DATE))
             SELECT (SELECT * FROM s_trades_count)                                AS first_table,
                    (SELECT * FROM execution_deposit_count)                       AS second_table,
                    (SELECT s_trades_count.count - execution_deposit_count.count) AS diff
             FROM s_trades_count,
                  execution_deposit_count;
             """;
-        return jdbcTemplate.queryForObject(sqlQuery, resultContainerMapper);
+        ResultContainer resultContainer = jdbcTemplate.queryForObject(sqlQuery, resultContainerMapper);
+        if (resultContainer != null) {
+            resultContainer.setFirstTableTitle("s_trades");
+            resultContainer.setSecondTableRecordTitle("execution_deposit");
+        }
+
+        return resultContainer;
     }
 
     @Transactional
     public ResultContainer getIpoTrdCountDiff() {
         String sqlQuery = """
-            WITH s_trades_fond_count AS (SELECT COUNT(*) FROM s_trades WHERE section = 'FOND'),
+            WITH s_trades_fond_count AS (SELECT COUNT(*) FROM s_trades WHERE section = 'FOND' AND CAST(created_at AS DATE) = CAST(NOW() AS DATE)),
-                 execution_fond_count AS (SELECT COUNT(*) FROM execution_fond)
+                 execution_fond_count AS (SELECT COUNT(*) FROM execution_fond WHERE CAST(created_at AS DATE) = CAST(NOW() AS DATE))
             SELECT (SELECT * FROM s_trades_fond_count)                             AS first_table,
                    (SELECT * FROM execution_fond_count)                            AS second_table,
                    (SELECT s_trades_fond_count.count - execution_fond_count.count) AS diff
             FROM s_trades_fond_count,
                  execution_fond_count;
             """;
-        return jdbcTemplate.queryForObject(sqlQuery, resultContainerMapper);
+        ResultContainer resultContainer = jdbcTemplate.queryForObject(sqlQuery, resultContainerMapper);
+        if (resultContainer != null) {
+            resultContainer.setFirstTableTitle("s_trades");
+            resultContainer.setSecondTableRecordTitle("execution_fond");
+        }
+
+        return resultContainer;
     }
 
     @Transactional
     public ResultContainer getCurrInitCountDiff() {
         String sqlQuery = """
-            WITH s_trades_curr_count AS (SELECT COUNT(*) FROM s_trades WHERE section = 'CURR'),
+            WITH s_trades_curr_count AS (SELECT COUNT(*) FROM s_trades WHERE section = 'CURR' AND CAST(created_at AS DATE) = CAST(NOW() AS DATE)),
-                 execution_currency_count AS (SELECT COUNT(*) FROM execution_currency)
+                 execution_currency_count AS (SELECT COUNT(*) FROM execution_currency WHERE CAST(created_at AS DATE) = CAST(NOW() AS DATE))
             SELECT (SELECT * FROM s_trades_curr_count)                                 AS first_table,
                    (SELECT * FROM execution_currency_count)                            AS second_table,
                    (SELECT s_trades_curr_count.count - execution_currency_count.count) AS diff
             FROM s_trades_curr_count,
                  execution_currency_count;
             """;
-        return jdbcTemplate.queryForObject(sqlQuery, resultContainerMapper);
+        ResultContainer resultContainer = jdbcTemplate.queryForObject(sqlQuery, resultContainerMapper);
+        if (resultContainer != null) {
+            resultContainer.setFirstTableTitle("s_trades");
+            resultContainer.setSecondTableRecordTitle("execution_currency");
+        }
+
+        return resultContainer;
     }
 
     @Transactional
@@ -72,10 +90,10 @@ public class ControlRepository {
     @Transactional
     public ResultContainer getFinlInternalCountDiff(Long sessionId) {
         String sqlQuery = """
-            WITH payment_instruction_ids AS (SELECT id FROM payment_instruction WHERE session_id = :sessionId),
+            WITH payment_instruction_ids AS (SELECT id FROM payment_instruction WHERE session_id = :sessionId AND CAST(created_at AS DATE) = CAST(NOW() AS DATE)),
                  s_df03_ids AS (SELECT s_df03.id
                                 FROM payment_instruction_ids
-                                         LEFT JOIN s_df03 ON payment_instruction_ids.id = s_df03.payment_instruction_id),
+                                         LEFT JOIN s_df03 ON payment_instruction_ids.id = s_df03.payment_instruction_id AND CAST(generation_time AS DATE) = CAST(NOW() AS DATE)),
                  payment_instruction_count AS (SELECT COUNT(*) FROM payment_instruction_ids),
                  s_df03_count AS (SELECT COUNT(*) FROM s_df03_ids)
             SELECT (SELECT payment_instruction_count.count FROM payment_instruction_count) AS first_table,
@@ -84,8 +102,14 @@ public class ControlRepository {
             FROM payment_instruction_count,
                  s_df03_count;
             """;
-        return namedParameterJdbcTemplate.queryForObject(sqlQuery,
+        ResultContainer resultContainer = namedParameterJdbcTemplate.queryForObject(sqlQuery,
             new MapSqlParameterSource("sessionId", sessionId),
             resultContainerMapper);
+        if (resultContainer != null) {
+            resultContainer.setFirstTableTitle("payment_instruction");
+            resultContainer.setSecondTableRecordTitle("s_df03");
+        }
+
+        return resultContainer;
     }
 }

clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/service/CommandService.java

@@ -0,0 +1,32 @@
+package ru.spcex.clearing.control.service;
+
+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.cud.clearing.ControlServiceRequest;
+import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
+import ru.spcex.platform.enumeration.Task;
+
+@Service
+public class CommandService extends QueueConsumer implements InitializingBean {
+    private final ControlService controlService;
+
+    public CommandService(Consumer<String, Object> kafkaQueue, ControlService controlService) {
+        super(kafkaQueue);
+        this.controlService = controlService;
+    }
+
+    @Override
+    public void afterPropertiesSet() throws Exception {
+        callback(ControlServiceRequest.class)
+            .setConsumer(request -> controlService.generalControl(request.getRequestPayload().getSessionType()))
+            .forDestination(Task.generalControl_GLCL.topic(), callbacks::put);
+        callback(ControlServiceRequest.class)
+            .setConsumer(request -> controlService.internalControl(request.getRequestPayload().getSessionId()))
+            .forDestination(Task.internalControl_ILCL.topic(), callbacks::put);
+        callback(ControlServiceRequest.class)
+            .setConsumer(request -> controlService.internalControlFinl(request.getRequestPayload().getSessionId()))
+            .forDestination(Task.internalFinlControl_IFCL.topic(), callbacks::put);
+        init();
+    }
+}

clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/service/ControlService.java

@@ -1,19 +1,27 @@
 package ru.spcex.clearing.control.service;
 
+import com.opencsv.bean.StatefulBeanToCsv;
+import com.opencsv.bean.StatefulBeanToCsvBuilder;
+import com.opencsv.exceptions.CsvDataTypeMismatchException;
+import com.opencsv.exceptions.CsvRequiredFieldEmptyException;
 import java.io.FileWriter;
 import java.io.IOException;
-import java.io.PrintWriter;
+import java.io.Writer;
 import java.nio.file.Path;
 import java.time.Instant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 import ru.spcex.clearing.control.config.settings.ControlServiceSettings;
 import ru.spcex.clearing.control.model.ResultContainer;
 import ru.spcex.clearing.control.repository.ControlRepository;
+import ru.spcex.clearing.control.util.CustomMappingStrategy;
 import ru.spcex.platform.enumeration.SessionType;
-import ru.spcex.platform.utils.collection.Pair;
 
 @Service
 public class ControlService {
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
     private final ControlRepository controlRepository;
     private final ControlServiceSettings settings;
     private final Path resultDirectory;
@@ -25,73 +33,58 @@ public class ControlService {
         resultDirectory = Path.of(settings.getResultDirectory());
     }
 
-    public String generalControl(SessionType sessionType) {
+    public void generalControl(SessionType sessionType) {
-        Pair<String, String> tables;
+        log.info("Starting general control with session type {}", sessionType);
-        ResultContainer resultContainer;
 
-        if (sessionType.equals(SessionType.FINL)) {
+        ResultContainer resultContainer = switch (sessionType) {
-            tables = new Pair<>("sTrade", "executionDeposit");
+            case FINL -> controlRepository.getFinlCountDiff();
-            resultContainer = controlRepository.getFinlCountDiff();
+            case IPOB, IPOT, IPO0, TRDT -> controlRepository.getIpoTrdCountDiff();
-        } else if (sessionType.equals(SessionType.IPOB) || sessionType.equals(SessionType.IPOT) || sessionType.equals(SessionType.IPO0) || sessionType.equals(SessionType.TRDT)) {
+            case CURR -> controlRepository.getCurrInitCountDiff();
-            tables = new Pair<>("sTradeFond", "executionFond");
+            default -> {
-            resultContainer = controlRepository.getIpoTrdCountDiff();
+                log.error("Unknown session type {}", sessionType);
-        } else if (sessionType.equals(SessionType.CURR)) {
+                throw new IllegalStateException("Unexpected value: " + sessionType);
-            tables = new Pair<>("sTradeCurrency", "executionCurrency");
+            }
-            resultContainer = controlRepository.getCurrInitCountDiff();
+        };
-        } else {
+
-            throw new IllegalStateException("Unexpected value: " + sessionType);
+        if (resultContainer.getQuantityDifference() != 0) {
+            resultContainer.setSessionType(sessionType);
+            writeToCsvFile(resultContainer);
         }
-
-        if (resultContainer.difference() != 0) {
-            writeToCsvFile(sessionType, resultContainer, tables);
-        }
-
-        return resultContainer.toString();
     }
 
-    public String internalControl(Long sessionId) {
+    public void internalControl(Long sessionId) {
+        log.info("Starting internal control with session id {}", sessionId);
+
         ResultContainer resultContainer = controlRepository.getInternalCountDiff(sessionId);
 
-        if (resultContainer.difference() != 0) {
+        if (resultContainer.getQuantityDifference() != 0) {
-            writeToCsvFileInternal(sessionId, resultContainer);
+            resultContainer.setSessionId(sessionId);
+            writeToCsvFile(resultContainer);
         }
-
-        return resultContainer.toString();
     }
 
-    public String internalControlFinl(Long sessionId) {
+    public void internalControlFinl(Long sessionId) {
+        log.info("Starting internal control FINL with session id {}", sessionId);
+
         ResultContainer resultContainer = controlRepository.getFinlInternalCountDiff(sessionId);
 
-        if (resultContainer.difference() != 0) {
+        if (resultContainer.getQuantityDifference() != 0) {
-            writeToCsvFileInternal(sessionId, resultContainer);
+            resultContainer.setSessionId(sessionId);
-        }
+            writeToCsvFile(resultContainer);
-
-        return resultContainer.toString();
-    }
-
-    private void writeToCsvFile(SessionType sessionType, ResultContainer resultContainer, Pair<String, String> tables) {
-        try (PrintWriter printWriter = new PrintWriter(new FileWriter(resultDirectory.resolve("result-" + sessionType.name() + "-" + Instant.now().getEpochSecond() + ".csv").toFile()))) {
-            printWriter.printf("session_type,%s,%s,diff%n", tables.getFirst(), tables.getSecond());
-            printWriter.printf("%s,%s,%s,%s%n",
-                sessionType.name(),
-                resultContainer.firstTable(),
-                resultContainer.secondTable(),
-                resultContainer.difference());
-        } catch (IOException e) {
-            throw new RuntimeException(e);
         }
     }
 
-    private void writeToCsvFileInternal(Long sessionId, ResultContainer resultContainer) {
+    private void writeToCsvFile(ResultContainer resultContainer) {
-        try (PrintWriter printWriter = new PrintWriter(new FileWriter(resultDirectory.resolve("result-internal-" + sessionId + "-" + Instant.now().getEpochSecond() + ".csv").toFile()))) {
+        try (Writer writer = new FileWriter(resultDirectory.resolve("result-" + Instant.now().getEpochSecond() + ".csv").toFile())) {
-            printWriter.println("session_id,paymentInstruction,s_df03,diff");
+            CustomMappingStrategy<ResultContainer> mappingStrategy = new CustomMappingStrategy<>();
-            printWriter.printf("%s,%s,%s,%s%n",
+            mappingStrategy.setType(ResultContainer.class);
-                sessionId,
+
-                resultContainer.firstTable(),
+            StatefulBeanToCsv<ResultContainer> sbc = new StatefulBeanToCsvBuilder<ResultContainer>(writer)
-                resultContainer.secondTable(),
+                .withMappingStrategy(mappingStrategy)
-                resultContainer.difference());
+                .build();
-        } catch (IOException e) {
+            sbc.write(resultContainer);
-            throw new RuntimeException(e);
+        } catch (CsvRequiredFieldEmptyException | CsvDataTypeMismatchException | IOException e) {
+            log.error(e.getMessage(), e);
         }
     }
 }

clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/util/CustomMappingStrategy.java

@@ -0,0 +1,52 @@
+package ru.spcex.clearing.control.util;
+
+import com.opencsv.bean.AbstractCsvConverter;
+import com.opencsv.bean.BeanField;
+import com.opencsv.bean.ColumnPositionMappingStrategy;
+import com.opencsv.bean.ConverterNumber;
+import com.opencsv.bean.CsvBindByName;
+import com.opencsv.bean.CsvBindByPosition;
+import com.opencsv.bean.CsvConverter;
+import com.opencsv.bean.CsvNumber;
+import com.opencsv.exceptions.CsvBadConverterException;
+import com.opencsv.exceptions.CsvChainedException;
+import com.opencsv.exceptions.CsvFieldAssignmentException;
+import com.opencsv.exceptions.CsvRequiredFieldEmptyException;
+import java.lang.reflect.Field;
+import java.math.BigDecimal;
+import java.util.Locale;
+
+public class CustomMappingStrategy<T> extends ColumnPositionMappingStrategy<T> {
+    public String[] generateHeader(T bean) throws CsvRequiredFieldEmptyException {
+        super.generateHeader(bean);
+        String[] headers = new String[this.getFieldMap().values().size()];
+        for (BeanField<T, Integer> field : this.getFieldMap().values()) {
+            CsvBindByPosition positionAnnotation = field.getField().getAnnotation(CsvBindByPosition.class);
+            CsvBindByName nameAnnotation = field.getField().getAnnotation(CsvBindByName.class);
+            headers[positionAnnotation.position()] = nameAnnotation.column();
+        }
+        return headers;
+    }
+
+    @Override
+    public String[] transmuteBean(T bean) throws CsvFieldAssignmentException, CsvChainedException {
+        return super.transmuteBean(bean);
+    }
+
+    @Override
+    protected CsvConverter determineConverter(Field field, Class<?> elementType, String locale, String writeLocale, Class<? extends AbstractCsvConverter> customConverter) throws CsvBadConverterException {
+        CsvConverter csvConverter = super.determineConverter(field, elementType, locale, writeLocale, customConverter);
+        if (csvConverter instanceof ConverterNumber && field.getType() == BigDecimal.class) {
+            CsvNumber csvNumberAnnotation = field.getAnnotation(CsvNumber.class);
+            String formatValue = csvNumberAnnotation.value();
+            String writeFormat = csvNumberAnnotation.writeFormat();
+            csvConverter = new ConverterNumber(BigDecimal.class,
+                Locale.getDefault().toString(),
+                Locale.getDefault().toString(),
+                Locale.getDefault(),
+                formatValue,
+                writeFormat);
+        }
+        return csvConverter;
+    }
+}

clearing-parent/control-service/src/main/resources/application.properties

@@ -1,7 +1,27 @@
+# Result directory
+control-service.result-directory=C:\\Users\\inikolaev\\Desktop\\result
+
 # Database settings
 control-service.database.username=clearing
 control-service.database.password=Aa111111
 control-service.database.url=jdbc:postgresql://localhost:5433/clearing
 
-# Result directory
+# Hazelcast settings
-control-service.result-directory=C:\\Users\\ivan\\Desktop\\result
+control-service.hazelcast.cluster-members=10.200.200.181:5701
+control-service.hazelcast.login=dev
+control-service.hazelcast.password=dev-pass
+
+# Kafka consumer settings
+control-service.kafka-consumer.bootstrap-servers=localhost:19092
+control-service.kafka-consumer.group-id=dev-group-balance-service
+control-service.kafka-consumer.enable-auto-commit=false
+control-service.kafka-consumer.session-timeout-ms=30000
+control-service.kafka-consumer.auto-offset-reset=latest
+
+# Kafka producer settings
+control-service.kafka-producer.bootstrap-servers=localhost:19092
+control-service.kafka-producer.acks=all
+control-service.kafka-producer.retries=0
+control-service.kafka-producer.batch-size=16384
+control-service.kafka-producer.linger-ms=1
+control-service.kafka-producer.buffer-memory=33554432

platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/Task.java

@@ -59,6 +59,9 @@ public enum Task implements IEnumKey {
     finishBadSessions("CCLR"),//Завершение неудачных клиринговых сессий
     sendLim_LIMC("LIMC"), // Выгрузка в Торговую систему остатков по валюте (отправка lim)"
     makeFiles_MTCR("MTCR"), // Формирование файлов с МТКР
+    generalControl_GLCL("GLCL"),
+    internalControl_ILCL("ILCL"),
+    internalFinlControl_IFCL("IFCL"),
     ;
 
     private final String key;

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/clearing/ControlServiceRequest.java

@@ -0,0 +1,27 @@
+package ru.spcex.clearing.platform.messaging.domain.cud.clearing;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import ru.spcex.platform.enumeration.SessionType;
+
+public class ControlServiceRequest {
+    @JsonProperty
+    private Long sessionId;
+    @JsonProperty
+    private SessionType sessionType;
+
+    public Long getSessionId() {
+        return sessionId;
+    }
+
+    public void setSessionId(Long sessionId) {
+        this.sessionId = sessionId;
+    }
+
+    public SessionType getSessionType() {
+        return sessionType;
+    }
+
+    public void setSessionType(SessionType sessionType) {
+        this.sessionType = sessionType;
+    }
+}