http://jira.mfd.msk:8088/browse/CLS-16

clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/config/MessagesConfig.java

@@ -0,0 +1,21 @@
+package ru.spcex.clearing.backendapi.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.support.ResourceBundleMessageSource;
+
+import java.util.Locale;
+
+@Configuration
+public class MessagesConfig {
+    @Bean
+    public ResourceBundleMessageSource messages() {
+        ResourceBundleMessageSource source = new ResourceBundleMessageSource();
+        source.setBasenames("messages/response");
+        source.setUseCodeAsDefaultMessage(true);
+        source.setDefaultEncoding("utf8");
+        source.setDefaultLocale(Locale.ROOT);
+
+        return source;
+    }
+}

clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/config/element/BackendApiSettings.java

@@ -3,7 +3,7 @@ package ru.spcex.clearing.backendapi.config.element;
 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.KafkaSettings;
+import ru.spcex.clearing.platform.messaging.config.element.KafkaProducerSettings;
 import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams;
 
 @Component
@@ -11,7 +11,7 @@ import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams;
 @ConfigurationProperties("backend-api")
 public class BackendApiSettings {
     private HazelcastClientParams hazelcast;
-    private KafkaSettings kafka;
+    private KafkaProducerSettings kafka;
     private String exampleSetting;
 
     public HazelcastClientParams getHazelcast() {
@@ -22,11 +22,11 @@ public class BackendApiSettings {
         this.hazelcast = hazelcast;
     }
 
-    public KafkaSettings getKafka() {
+    public KafkaProducerSettings getKafka() {
         return kafka;
     }
 
-    public void setKafka(KafkaSettings kafka) {
+    public void setKafka(KafkaProducerSettings kafka) {
         this.kafka = kafka;
     }
 

clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/meta/CudMetaService.java

@@ -3,6 +3,7 @@ package ru.spcex.clearing.backendapi.meta;
 import org.springframework.stereotype.Service;
 import ru.spcex.clearing.backendapi.controller.request.cud.MoneyMarketCreateAction;
 import ru.spcex.clearing.backendapi.domain.actions.IAction;
+import ru.spcex.clearing.platform.messaging.domain.Consts;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -13,7 +14,7 @@ public class CudMetaService {
 
     public CudMetaService() {
         this.mapping = new HashMap<>();
-        this.mapping.put("money-market-security-new", MoneyMarketCreateAction.class);
+        this.mapping.put(Consts.DESTINATION_MONEY_MARKET_SECURITY_NEW, MoneyMarketCreateAction.class);
     }
 
     public <T extends IAction<?>> Class<T> byDestination(String destination) {

clearing-parent/pom.xml

@@ -23,6 +23,7 @@
         <module>db-scripts</module>
         <module>dbf-importer</module>
         <module>dbf-exporter</module>
+        <module>securities-service</module>
     </modules>
 
     <properties>

clearing-parent/securities-service/pom.xml

@@ -0,0 +1,62 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>clearing-parent</artifactId>
+        <groupId>ru.spcex.clearing</groupId>
+        <version>SPCEX-1.0.0.0</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>securities-service</artifactId>
+
+    <properties>
+        <maven.compiler.source>17</maven.compiler.source>
+        <maven.compiler.target>17</maven.compiler.target>
+    </properties>
+    <dependencies>
+        <dependency>
+            <groupId>ru.spcex.platform</groupId>
+            <artifactId>platform-messaging</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>ru.spcex.platform</groupId>
+            <artifactId>platform-imdg-api-hazelcast-impl</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter</artifactId>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <resources>
+            <resource>
+                <directory>src/main/resources</directory>
+                <excludes>
+                    <exclude>application.properties</exclude>
+                </excludes>
+                <filtering>false</filtering>
+            </resource>
+        </resources>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>repackage</goal>
+                        </goals>
+                    </execution>
+                </executions>
+                <configuration>
+                    <finalName>${project.artifactId}</finalName>
+                </configuration>
+            </plugin>
+        </plugins>
+
+    </build>
+
+</project>

clearing-parent/securities-service/src/main/java/ru/spcex/clearing/securities/SecuritiesServiceApplication.java

@@ -0,0 +1,12 @@
+package ru.spcex.clearing.securities;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class SecuritiesServiceApplication {
+    public static void main(String[] args) {
+        SpringApplication app = new SpringApplication(SecuritiesServiceApplication.class);
+        app.run(args);
+    }
+}

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

@@ -0,0 +1,19 @@
+package ru.spcex.clearing.securities.config;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import ru.spcex.clearing.platform.messaging.config.KafkaConsumerFactory;
+import ru.spcex.clearing.securities.config.element.SecuritiesServiceSettings;
+
+@Configuration
+public class KafkaConfig {
+
+    @Autowired
+    @Bean
+    public Consumer<String, Object> createProducer(SecuritiesServiceSettings settings) {
+        return KafkaConsumerFactory.consumer(settings.getKafka());
+    }
+
+}

clearing-parent/securities-service/src/main/java/ru/spcex/clearing/securities/config/SecuritiesServiceImdgConfig.java

@@ -0,0 +1,48 @@
+package ru.spcex.clearing.securities.config;
+
+import org.springframework.beans.factory.annotation.Autowired;
+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.securities.config.element.SecuritiesServiceSettings;
+import ru.spcex.platform.imdg.api.ImdgProvider;
+import ru.spcex.platform.imdg.iml.hazelcast.service.HazelcastService;
+
+@Configuration
+public class SecuritiesServiceImdgConfig {
+    @Bean(name = "taskExecutorHazelcastClientInitializer")
+    public ThreadPoolTaskExecutor taskExecutorHazelcastClientInitializer() {
+        return createThreadPoolTaskExecutor(1, true);
+    }
+
+    @Bean(name = "taskExecutorIdGeneratorAwaiter")
+    public ThreadPoolTaskExecutor taskExecutorIdGeneratorAwaiter() {
+        return createThreadPoolTaskExecutor(1, false);
+    }
+
+    @Autowired
+    @Bean
+    public ImdgProvider imdgProvider(
+            @Qualifier("taskExecutorHazelcastClientInitializer") ThreadPoolTaskExecutor taskExecutorHazelcastClientInitializer,
+            @Qualifier("taskExecutorIdGeneratorAwaiter") ThreadPoolTaskExecutor taskExecutorIdGeneratorAwaiter,
+            SecuritiesServiceSettings settings
+    ) {
+        return new HazelcastService(taskExecutorHazelcastClientInitializer,
+                taskExecutorIdGeneratorAwaiter,
+                settings.getHazelcast());
+    }
+
+
+    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/securities-service/src/main/java/ru/spcex/clearing/securities/config/element/SecuritiesServiceSettings.java

@@ -0,0 +1,31 @@
+package ru.spcex.clearing.securities.config.element;
+
+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("securities-service")
+public class SecuritiesServiceSettings {
+    private HazelcastClientParams hazelcast;
+    private KafkaConsumerSettings kafka;
+
+    public HazelcastClientParams getHazelcast() {
+        return hazelcast;
+    }
+
+    public void setHazelcast(HazelcastClientParams hazelcast) {
+        this.hazelcast = hazelcast;
+    }
+
+    public KafkaConsumerSettings getKafka() {
+        return kafka;
+    }
+
+    public void setKafka(KafkaConsumerSettings kafka) {
+        this.kafka = kafka;
+    }
+}

clearing-parent/securities-service/src/main/java/ru/spcex/clearing/securities/service/cud/MoneyMarketSecurityService.java

@@ -0,0 +1,27 @@
+package ru.spcex.clearing.securities.service.cud;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import ru.spcex.clearing.platform.messaging.domain.Consts;
+import ru.spcex.clearing.platform.messaging.domain.cud.securitites.MoneyMarketCreateRequest;
+import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
+
+@Service
+public class MoneyMarketSecurityService extends QueueConsumer implements InitializingBean {
+    @Autowired
+    public MoneyMarketSecurityService(Consumer<String, Object> kafkaQueue) {
+        super(kafkaQueue);
+    }
+
+    @Override
+    public void afterPropertiesSet() {
+        callback(Consts.DESTINATION_MONEY_MARKET_SECURITY_NEW, MoneyMarketCreateRequest.class)
+                .define(this::newMoneyMarket);
+    }
+
+    private void newMoneyMarket(MoneyMarketCreateRequest req) {
+
+    }
+}

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

@@ -0,0 +1,12 @@
+spring.main.web-application-type=none
+
+securities-service.hazelcast.cluster-members=127.0.0.1
+securities-service.hazelcast.login=dev
+securities-service.hazelcast.password=dev-pass
+
+securities-service.kafka.bootstrap-servers=localhost:9092
+securities-service.kafka.acks=all
+securities-service.kafka.retries=0
+securities-service.kafka.batch-size=16384
+securities-service.kafka.linger-ms=1
+securities-service.kafka.buffer-memory=33554432

clearing-parent/securities-service/src/main/resources/logback.xml

@@ -0,0 +1,38 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+	<appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
+		<encoder>
+<!--            |%X{ru.nbch.scoring.web.logging.mdc_key}-->
+			<Pattern>%date{HH:mm:ss.SSS} [%thread] %-5level %class{0}:%line - %message%n</Pattern>
+			<charset>utf-8</charset>
+		</encoder>
+	</appender>
+	<appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
+        <file>./logs/securities-service.log</file>
+        <encoder>
+<!--            |%X{ru.nbch.scoring.web.logging.mdc_key}-->
+            <Pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %class{0}:%msg%n</Pattern>
+            <charset>utf8</charset>
+        </encoder>
+        <rollingPolicy class="ch.qos.logback.core.rolling.FixedWindowRollingPolicy">
+            <fileNamePattern>
+                ./logs/securities-service.%i.log
+            </fileNamePattern>
+            <minIndex>1</minIndex>
+            <maxIndex>10</maxIndex>
+        </rollingPolicy>
+        <triggeringPolicy class="ch.qos.logback.core.rolling.SizeBasedTriggeringPolicy">
+            <maxFileSize>500MB</maxFileSize>
+        </triggeringPolicy>
+    </appender>
+
+	<root level="warn">
+		<appender-ref ref="CONSOLE"/>
+		<appender-ref ref="FILE"/>
+	</root>
+
+    <logger name="ru.spcex" level="debug" additivity="false">
+        <appender-ref ref="FILE"/>
+        <appender-ref ref="CONSOLE"/>
+    </logger>
+</configuration>

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/config/KafkaConsumerFactory.java

@@ -0,0 +1,23 @@
+package ru.spcex.clearing.platform.messaging.config;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import ru.spcex.clearing.platform.messaging.config.element.KafkaConsumerSettings;
+
+import java.util.Properties;
+
+public class KafkaConsumerFactory {
+    public static Consumer<String, Object> consumer(KafkaConsumerSettings kafkaSettings) {
+        Properties props = new Properties();
+        props.put("bootstrap.servers", kafkaSettings.getBootstrapServers());
+        if (kafkaSettings.getGroupId() != null && kafkaSettings.getGroupId().length() > 0) {
+            props.put("group.id", kafkaSettings.getGroupId());
+        }
+        props.put("enable.auto.commit", kafkaSettings.getEnableAutoCommit().toString());
+        props.put("session.timeout.ms", kafkaSettings.getSessionTimeoutMs().toString());
+        props.put("auto.offset.reset", kafkaSettings.getAutoOffsetReset());
+        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+        return new KafkaConsumer<>(props);
+    }
+}

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/config/KafkaProducerFactory.java

@@ -3,13 +3,13 @@ package ru.spcex.clearing.platform.messaging.config;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerConfig;
-import ru.spcex.clearing.platform.messaging.config.element.KafkaSettings;
+import ru.spcex.clearing.platform.messaging.config.element.KafkaProducerSettings;
 import ru.spcex.clearing.platform.messaging.serialization.JsonSerializer;
 
 import java.util.Properties;
 
 public class KafkaProducerFactory {
-    public static Producer<String, Object> producer(KafkaSettings kafkaSettings) {
+    public static Producer<String, Object> producer(KafkaProducerSettings kafkaSettings) {
         Properties kafkaProps = new Properties();
 
         //Assign localhost id

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/config/element/KafkaConsumerSettings.java

@@ -0,0 +1,50 @@
+package ru.spcex.clearing.platform.messaging.config.element;
+
+public class KafkaConsumerSettings {
+    private String bootstrapServers;
+    private String groupId;
+    private Boolean enableAutoCommit;
+    private Integer sessionTimeoutMs;
+    private String autoOffsetReset;
+
+
+    public String getBootstrapServers() {
+        return bootstrapServers;
+    }
+
+    public void setBootstrapServers(String bootstrapServers) {
+        this.bootstrapServers = bootstrapServers;
+    }
+
+    public String getGroupId() {
+        return groupId;
+    }
+
+    public void setGroupId(String groupId) {
+        this.groupId = groupId;
+    }
+
+    public Boolean getEnableAutoCommit() {
+        return enableAutoCommit;
+    }
+
+    public void setEnableAutoCommit(Boolean enableAutoCommit) {
+        this.enableAutoCommit = enableAutoCommit;
+    }
+
+    public Integer getSessionTimeoutMs() {
+        return sessionTimeoutMs;
+    }
+
+    public void setSessionTimeoutMs(Integer sessionTimeoutMs) {
+        this.sessionTimeoutMs = sessionTimeoutMs;
+    }
+
+    public String getAutoOffsetReset() {
+        return autoOffsetReset;
+    }
+
+    public void setAutoOffsetReset(String autoOffsetReset) {
+        this.autoOffsetReset = autoOffsetReset;
+    }
+}

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/config/element/KafkaProducerSettings.java

@@ -1,6 +1,6 @@
 package ru.spcex.clearing.platform.messaging.config.element;
 
-public class KafkaSettings {
+public class KafkaProducerSettings {
     private String bootstrapServers;
     private String acks;
     private Integer retries;

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

@@ -0,0 +1,6 @@
+package ru.spcex.clearing.platform.messaging.domain;
+
+public interface Consts {
+    String DESTINATION_MONEY_MARKET_SECURITY_NEW = "money-market-security-new";
+
+}

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/QueueConsumer.java

@@ -0,0 +1,96 @@
+package ru.spcex.clearing.platform.messaging.service;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.common.errors.WakeupException;
+
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class QueueConsumer implements AutoCloseable {
+    private final AtomicBoolean closed = new AtomicBoolean(false);
+    private final Consumer<String, Object> consumer;
+    private final ExecutorService executor;
+    private final Map<String, ConsumerWithClass<?>> callbacks;
+    private final ObjectMapper json;
+
+    public QueueConsumer(Consumer<String, Object> kafkaQueue) {
+        this.consumer = kafkaQueue;
+        this.callbacks = new HashMap<>();
+        this.executor = Executors.newSingleThreadExecutor();
+        this.json = new ObjectMapper();
+    }
+
+    protected <T> void addCallBack(String destination, ConsumerWithClass<T> callback) {
+        this.callbacks.put(destination, callback);
+    }
+
+    public void init() throws Exception {
+        executor.submit(() -> {
+            try {
+                consumer.subscribe(callbacks.keySet());
+                while (!closed.get()) {
+                    ConsumerRecords<String, Object> records = consumer.poll(Duration.of(10, ChronoUnit.SECONDS));
+                    for (ConsumerRecord<String, Object> next : records) {
+                        ConsumerWithClass<?> callback = callbacks.get(next.topic());
+                        Object o = json.readValue((String) next.value(), callback.getClazz());
+                        callback.acceptRaw(o);
+                    }
+                }
+            } catch (WakeupException e) {
+                if (!closed.get()) throw e;
+            } catch (JsonProcessingException e) {
+                throw new RuntimeException(e);
+            } finally {
+                consumer.close();
+            }
+        });
+    }
+
+    @Override
+    public void close() throws Exception {
+        closed.set(true);
+        consumer.wakeup();
+    }
+
+    protected <T> ConsumerWithClass<T> callback(String destination, Class<T> clazz) {
+        ConsumerWithClass<T> callback = new ConsumerWithClass<>(clazz);
+        this.callbacks.put(destination, callback);
+        return callback;
+    }
+
+    protected static class ConsumerWithClass<T> {
+        private final Class<T> clazz;
+        private java.util.function.Consumer<T> consumer;
+        private ConsumerWithClass(Class<T> clazz) {
+            this.clazz = clazz;
+        }
+
+
+
+        public ConsumerWithClass<T> define(java.util.function.Consumer<T> consumer) {
+            this.consumer = consumer;
+            return this;
+        }
+
+        public void accept(T obj) {
+            this.consumer.accept(obj);
+        }
+
+        public void acceptRaw(Object obj) {
+            this.consumer.accept((T) obj);
+        }
+
+        public Class<T> getClazz() {
+            return clazz;
+        }
+    }
+}