reports-service добавление kafka, начало

clearing-parent/reports-service/pom.xml

@@ -62,6 +62,10 @@
             <groupId>ru.spcex.platform</groupId>
             <artifactId>platform-imdg-api-hazelcast-impl</artifactId>
         </dependency>
+        <dependency>
+            <groupId>ru.spcex.platform</groupId>
+            <artifactId>platform-messaging</artifactId>
+        </dependency>
 
         <dependency>
             <groupId>org.junit.jupiter</groupId>

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

@@ -0,0 +1,28 @@
+package ru.spcex.clearing.reports.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.platform.messaging.config.KafkaConsumerFactory;
+import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory;
+import ru.spcex.clearing.reports.config.element.ReportsServiceSettings;
+
+@Configuration
+public class KafkaConfig {
+    @Autowired
+    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
+    @Bean
+    public Consumer<String, Object> createConsumer(ReportsServiceSettings settings) {
+        return KafkaConsumerFactory.consumer(settings.getKafkaConsumer());
+    }
+
+//    @Autowired
+//    @Bean
+//    public Producer<String, Object> createProducer(ReportsServiceSettings settings) {
+//        return KafkaProducerFactory.producer(settings.getKafkaProducer());
+//    }
+}

clearing-parent/reports-service/src/main/java/ru/spcex/clearing/reports/config/element/ReportsServiceSettings.java

@@ -3,6 +3,8 @@ package ru.spcex.clearing.reports.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.clearing.platform.messaging.config.element.KafkaProducerSettings;
 import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams;
 
 /**
@@ -15,7 +17,8 @@ import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams;
 @ConfigurationProperties("reports-service")
 public class ReportsServiceSettings {
     private HazelcastClientParams hazelcast;
-    private String reportModuleOut;
+    private KafkaConsumerSettings kafkaConsumer;
+    private KafkaProducerSettings kafkaProducer;
 
     public HazelcastClientParams getHazelcast() {
         return hazelcast;
@@ -25,12 +28,19 @@ public class ReportsServiceSettings {
         this.hazelcast = hazelcast;
     }
 
-
-    public String getReportModuleOut() {
-        return reportModuleOut;
+    public KafkaConsumerSettings getKafkaConsumer() {
+        return kafkaConsumer;
     }
 
-    public void setReportModuleOut(String reportModuleOut) {
-        this.reportModuleOut = reportModuleOut;
+    public void setKafkaConsumer(KafkaConsumerSettings kafkaConsumer) {
+        this.kafkaConsumer = kafkaConsumer;
+    }
+
+    public KafkaProducerSettings getKafkaProducer() {
+        return kafkaProducer;
+    }
+
+    public void setKafkaProducer(KafkaProducerSettings kafkaProducer) {
+        this.kafkaProducer = kafkaProducer;
     }
 }

clearing-parent/reports-service/src/main/java/ru/spcex/clearing/reports/services/QCommandExecutor.java

@@ -0,0 +1,49 @@
+package ru.spcex.clearing.reports.services;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.InitializingBean;
+import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
+import ru.spcex.clearing.platform.messaging.domain.Consts;
+//import ru.spcex.clearing.platform.messaging.domain.cud.balance.ExportToFileRequest; todo use class?
+import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
+import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
+import ru.spcex.platform.enumeration.Task;
+import ru.spcex.platform.imdg.api.Imdg;
+import ru.spcex.platform.imdg.api.ImdgId;
+import ru.spcex.platform.imdg.api.ImdgProvider;
+
+import java.util.List;
+
+public class QCommandExecutor extends QueueConsumer implements InitializingBean {
+    private final Logger log = LoggerFactory.getLogger(getClass());
+//    private final ImdgId idGenerator;
+//    private final KafkaSender kafkaReqProducer;
+
+
+    public QCommandExecutor(Consumer<String, Object> kafkaQueue, ImdgProvider imdgProvider,
+                            List<ReportDataCollector> listOfCollectors) {
+        super(kafkaQueue);
+//     todo constructor
+    }
+
+    @Override
+    public void afterPropertiesSet() {
+        log.debug("Init queue listener {}", getClass().getSimpleName());
+//        callback(Object.class)
+//                .setConsumer(this::newReportDo)
+//                .forDestination(Task.getVerification!!!.topic(), callbacks::put); // todo здесь надо указать команду из ТЗ или согласовать с фронтэндом
+//        init();
+    }
+
+    private void newReportDo(BaseRequest<Object> userRequest) {
+        log.debug("newReportDo request received");
+        // todo logic
+        log.debug("successfully processed");
+    }
+}

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

@@ -2,7 +2,24 @@ server.port=8080
 server.servlet.context-path=/reports_service
 spring.main.web-application-type=servlet
 
+
+reports-service.kafka-consumer.bootstrap-servers=localhost:9092
+reports-service.kafka-consumer.group-id=dev-group-balance-service
+reports-service.kafka-consumer.enable-auto-commit=false
+reports-service.kafka-consumer.session-timeout-ms=30000
+reports-service.kafka-consumer.auto-offset-reset=latest
+reports-service.kafka-consumer.linger-ms=1
+reports-service.kafka-consumer.buffer-memory=33554432
+
+#reports-service.kafka-producer.bootstrap-servers=localhost:9092
+#reports-service.kafka-producer.acks=all
+#reports-service.kafka-producer.retries=0
+#reports-service.kafka-producer.batch-size=16384
+#reports-service.kafka-producer.linger-ms=1
+#reports-service.kafka-producer.buffer-memory=33554432
+
 reports-service.hazelcast.cluster-members=127.0.0.1:5701,10.200.200.181:5701
 reports-service.hazelcast.login=dev
 reports-service.hazelcast.password=dev-pass
+
 reports-service.ReportModuleOut=./report_out