diff --git a/clearing-parent/control-service/pom.xml b/clearing-parent/control-service/pom.xml index 3fed973f9..56887fec3 100644 --- a/clearing-parent/control-service/pom.xml +++ b/clearing-parent/control-service/pom.xml @@ -16,14 +16,15 @@ + 5.5.2 7.0.1 - + org.springframework.boot - spring-boot-starter-web + spring-boot-starter @@ -40,6 +41,23 @@ com.mchange c3p0 + + ru.spcex.platform + platform-imdg-api-hazelcast-impl + + + + + com.opencsv + opencsv + ${opencsv.version} + + + + + ru.spcex.platform + platform-messaging + @@ -47,21 +65,24 @@ logstash-logback-encoder ${logstash-logback-encoder.version} - - com.fasterxml.jackson.core - jackson-databind - + ru.spcex.platform platform-enum + org.springframework.boot spring-boot-starter-test test + + ru.spcex.clearing + test-clearing + test + diff --git a/clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/config/ImdgConfig.java b/clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/config/ImdgConfig.java new file mode 100644 index 000000000..c7129f970 --- /dev/null +++ b/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; + } + +} diff --git a/clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/config/KafkaConfig.java b/clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/config/KafkaConfig.java new file mode 100644 index 000000000..73f81de6c --- /dev/null +++ b/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 createConsumer(ControlServiceSettings settings) { + return KafkaConsumerFactory.consumer(settings.getKafkaConsumer()); + } + + @Autowired + @Bean + public Producer createProducer(ControlServiceSettings settings) { + return settings.getKafkaProducer() != null ? + KafkaProducerFactory.producer(settings.getKafkaProducer()) : + null; + } +} diff --git a/clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/config/KafkaSenderConfig.java b/clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/config/KafkaSenderConfig.java new file mode 100644 index 000000000..0855112c1 --- /dev/null +++ b/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 pf(ControlServiceSettings settings) { + if (settings.getKafkaProducer() == null) { + return null; + } + KafkaProducerSettings kafkaSettings = settings.getKafkaProducer(); + return KafkaProducerFactory.producerFactory(kafkaSettings); + } + + @Bean("kafkaTemplate") + public KafkaTemplate kafkaTemplate(ProducerFactory pf) { + if (pf == null) { + return null; + } + return new KafkaTemplate<>(pf); + } + + @Bean + public Supplier kafkaSenderSupplier(KafkaTemplate kafkaTemplate, + ImdgProvider imdgProvider) { + ImdgId imdgIdGenerator = imdgProvider.getImdgIdGenerator(); + return () -> KafkaSender + .setup() + .setKafkaTemplate(kafkaTemplate) + .idGenerator(imdgIdGenerator::nextId) + .saveRequestInfo(false) + .build(); + } +} diff --git a/clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/config/settings/ControlServiceSettings.java b/clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/config/settings/ControlServiceSettings.java index 6a3e73ceb..5116d326d 100644 --- a/clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/config/settings/ControlServiceSettings.java +++ b/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; } diff --git a/clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/controller/ControlController.java b/clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/controller/ControlController.java deleted file mode 100644 index 72ee5e8aa..000000000 --- a/clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/controller/ControlController.java +++ /dev/null @@ -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); - } -} diff --git a/clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/model/ResultContainer.java b/clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/model/ResultContainer.java index 63894e134..152373461 100644 --- a/clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/model/ResultContainer.java +++ b/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 + + '}'; + } } diff --git a/clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/repository/ControlRepository.java b/clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/repository/ControlRepository.java index 990be1513..707f537ec 100644 --- a/clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/repository/ControlRepository.java +++ b/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), - execution_deposit_count AS (SELECT COUNT(*) FROM execution_deposit) + 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 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'), - execution_fond_count AS (SELECT COUNT(*) FROM execution_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 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'), - execution_currency_count AS (SELECT COUNT(*) FROM execution_currency) + 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 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; } } diff --git a/clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/service/CommandService.java b/clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/service/CommandService.java new file mode 100644 index 000000000..3d80698b2 --- /dev/null +++ b/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 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(); + } +} diff --git a/clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/service/ControlService.java b/clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/service/ControlService.java index a16223bcb..783a4a908 100644 --- a/clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/service/ControlService.java +++ b/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) { - Pair tables; - ResultContainer resultContainer; + public void generalControl(SessionType sessionType) { + log.info("Starting general control with session type {}", sessionType); - if (sessionType.equals(SessionType.FINL)) { - tables = new Pair<>("sTrade", "executionDeposit"); - resultContainer = controlRepository.getFinlCountDiff(); - } else if (sessionType.equals(SessionType.IPOB) || sessionType.equals(SessionType.IPOT) || sessionType.equals(SessionType.IPO0) || sessionType.equals(SessionType.TRDT)) { - tables = new Pair<>("sTradeFond", "executionFond"); - resultContainer = controlRepository.getIpoTrdCountDiff(); - } else if (sessionType.equals(SessionType.CURR)) { - tables = new Pair<>("sTradeCurrency", "executionCurrency"); - resultContainer = controlRepository.getCurrInitCountDiff(); - } else { - throw new IllegalStateException("Unexpected value: " + sessionType); + ResultContainer resultContainer = switch (sessionType) { + case FINL -> controlRepository.getFinlCountDiff(); + case IPOB, IPOT, IPO0, TRDT -> controlRepository.getIpoTrdCountDiff(); + case CURR -> controlRepository.getCurrInitCountDiff(); + default -> { + log.error("Unknown session type {}", sessionType); + 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) { - writeToCsvFileInternal(sessionId, resultContainer); + if (resultContainer.getQuantityDifference() != 0) { + 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) { - writeToCsvFileInternal(sessionId, resultContainer); - } - - return resultContainer.toString(); - } - - private void writeToCsvFile(SessionType sessionType, ResultContainer resultContainer, Pair 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); + if (resultContainer.getQuantityDifference() != 0) { + resultContainer.setSessionId(sessionId); + writeToCsvFile(resultContainer); } } - private void writeToCsvFileInternal(Long sessionId, ResultContainer resultContainer) { - try (PrintWriter printWriter = new PrintWriter(new FileWriter(resultDirectory.resolve("result-internal-" + sessionId + "-" + Instant.now().getEpochSecond() + ".csv").toFile()))) { - printWriter.println("session_id,paymentInstruction,s_df03,diff"); - printWriter.printf("%s,%s,%s,%s%n", - sessionId, - resultContainer.firstTable(), - resultContainer.secondTable(), - resultContainer.difference()); - } catch (IOException e) { - throw new RuntimeException(e); + private void writeToCsvFile(ResultContainer resultContainer) { + try (Writer writer = new FileWriter(resultDirectory.resolve("result-" + Instant.now().getEpochSecond() + ".csv").toFile())) { + CustomMappingStrategy mappingStrategy = new CustomMappingStrategy<>(); + mappingStrategy.setType(ResultContainer.class); + + StatefulBeanToCsv sbc = new StatefulBeanToCsvBuilder(writer) + .withMappingStrategy(mappingStrategy) + .build(); + sbc.write(resultContainer); + } catch (CsvRequiredFieldEmptyException | CsvDataTypeMismatchException | IOException e) { + log.error(e.getMessage(), e); } } } diff --git a/clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/util/CustomMappingStrategy.java b/clearing-parent/control-service/src/main/java/ru/spcex/clearing/control/util/CustomMappingStrategy.java new file mode 100644 index 000000000..5d796c896 --- /dev/null +++ b/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 extends ColumnPositionMappingStrategy { + public String[] generateHeader(T bean) throws CsvRequiredFieldEmptyException { + super.generateHeader(bean); + String[] headers = new String[this.getFieldMap().values().size()]; + for (BeanField 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 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; + } +} diff --git a/clearing-parent/control-service/src/main/resources/application.properties b/clearing-parent/control-service/src/main/resources/application.properties index 7e62ee8b0..50e6b5083 100644 --- a/clearing-parent/control-service/src/main/resources/application.properties +++ b/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 -control-service.result-directory=C:\\Users\\ivan\\Desktop\\result +# Hazelcast settings +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 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 50988fa24..954b168c2 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 @@ -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; diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/clearing/ControlServiceRequest.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/clearing/ControlServiceRequest.java new file mode 100644 index 000000000..8a78f463d --- /dev/null +++ b/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; + } +}