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

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

@@ -12,7 +12,7 @@ public class KafkaConfig {
 
     @Autowired
     @Bean
-    public Producer<String, String> createProducer(BackendApiSettings settings) {
+    public Producer<String, Object> createProducer(BackendApiSettings settings) {
         return KafkaProducerFactory.producer(settings.getKafka());
     }
 

clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/cud/CudController.java

@@ -8,10 +8,7 @@ import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Controller;
-import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.*;
-import org.springframework.web.bind.annotation.RequestBody;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RequestMethod;
 import ru.spcex.clearing.backendapi.controller.response.BasicSpcexResponse;
 import ru.spcex.clearing.backendapi.domain.actions.IAction;
 import ru.spcex.clearing.backendapi.meta.CudMetaService;
@@ -22,18 +19,19 @@ import java.util.concurrent.Future;
 @Controller
 @RequestMapping("/cud")
 public class CudController {
-    private final Producer<String, String> kafka;
+    private final Producer<String, Object> kafka;
     private final ObjectMapper json;
     private final CudMetaService meta;
 
     @Autowired
-    public CudController(Producer<String, String> kafka, CudMetaService meta) {
+    public CudController(Producer<String, Object> kafka, CudMetaService meta) {
         this.kafka = kafka;
         this.meta = meta;
         this.json = new ObjectMapper();
     }
 
     @RequestMapping(value = "/{destination}", method = RequestMethod.POST)
+    @ResponseBody
     public BasicSpcexResponse add(@PathVariable("destination") String destination,
                                   @RequestBody String body) throws JsonProcessingException, ExecutionException, InterruptedException {
         Class<IAction<?>> actionClazz = meta.byDestination(destination);
@@ -41,8 +39,7 @@ public class CudController {
             throw new UnsupportedOperationException("unsupported destination " + destination);
         }
         IAction<?> iAction = json.readValue(body, actionClazz);
-
+        Future<RecordMetadata> send = kafka.send(new ProducerRecord<>(destination, iAction.toRequest()));
-        Future<RecordMetadata> send = kafka.send(new ProducerRecord<>(destination, body));
         RecordMetadata kafkaMetaData = send.get();
         MetaDataResponse responseToClient = new MetaDataResponse();
         responseToClient.setOffset(kafkaMetaData.offset());

clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/request/cud/MoneyMarketCreateAction.java

@@ -0,0 +1,106 @@
+package ru.spcex.clearing.backendapi.controller.request.cud;
+
+import com.fasterxml.jackson.annotation.JsonFormat;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import ru.spcex.clearing.backendapi.domain.actions.IAction;
+import ru.spcex.clearing.platform.messaging.domain.cud.securitites.MoneyMarketCreateRequest;
+
+import java.time.Instant;
+
+public class MoneyMarketCreateAction implements IAction<MoneyMarketCreateRequest> {
+    @JsonFormat(pattern = "yyyy-MM-dd", timezone = "Europe/Moscow")
+    @JsonProperty
+    public Instant startDate;
+    @JsonProperty
+    public Instant endDate;
+    @JsonProperty
+    public Double nominalValue;
+    @JsonProperty
+    public Long nominalCurrencyId;
+    @JsonProperty
+    public String instrumentType;
+    @JsonProperty
+    public String fullName;
+    @JsonProperty
+    public String securitySymbol;
+    @JsonProperty
+    public Double lotSize;
+
+    @Override
+    public MoneyMarketCreateRequest toRequest() {
+        var req = new MoneyMarketCreateRequest();
+        req.setStartDate(this.getStartDate());
+        req.setEndDate(this.getEndDate());
+        req.setNominalValue(this.getNominalValue());
+        req.setNominalCurrencyId(this.getNominalCurrencyId());
+        req.setInstrumentType(this.getInstrumentType());
+        req.setFullName(this.getFullName());
+        req.setSecuritySymbol(this.getSecuritySymbol());
+        req.setLotSize(this.getLotSize());
+        return req;
+    }
+
+    public Instant getStartDate() {
+        return startDate;
+    }
+
+    public void setStartDate(Instant startDate) {
+        this.startDate = startDate;
+    }
+
+    public Instant getEndDate() {
+        return endDate;
+    }
+
+    public void setEndDate(Instant endDate) {
+        this.endDate = endDate;
+    }
+
+    public Double getNominalValue() {
+        return nominalValue;
+    }
+
+    public void setNominalValue(Double nominalValue) {
+        this.nominalValue = nominalValue;
+    }
+
+    public Long getNominalCurrencyId() {
+        return nominalCurrencyId;
+    }
+
+    public void setNominalCurrencyId(Long nominalCurrencyId) {
+        this.nominalCurrencyId = nominalCurrencyId;
+    }
+
+    public String getInstrumentType() {
+        return instrumentType;
+    }
+
+    public void setInstrumentType(String instrumentType) {
+        this.instrumentType = instrumentType;
+    }
+
+    public String getFullName() {
+        return fullName;
+    }
+
+    public void setFullName(String fullName) {
+        this.fullName = fullName;
+    }
+
+    public String getSecuritySymbol() {
+        return securitySymbol;
+    }
+
+    public void setSecuritySymbol(String securitySymbol) {
+        this.securitySymbol = securitySymbol;
+    }
+
+    public Double getLotSize() {
+        return lotSize;
+    }
+
+    public void setLotSize(Double lotSize) {
+        this.lotSize = lotSize;
+    }
+}

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

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

clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/security/WebSecurityConfig.java

@@ -39,6 +39,7 @@ public class WebSecurityConfig extends KeycloakWebSecurityConfigurerAdapter {
                 .antMatchers("/anonymous/**").permitAll()
                 .antMatchers("/sso/login").permitAll()
                 .antMatchers("/error").permitAll()
+                .antMatchers( "/cud/**").permitAll() //todo remove
                 .antMatchers("/backend-api-login/**").permitAll()
                 .anyRequest().hasAnyRole("admin", "default-roles-master");
     }

platform-parent/platform-messaging/pom.xml

@@ -24,6 +24,21 @@
             <groupId>org.apache.kafka</groupId>
             <artifactId>kafka-clients</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-core</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>ru.spcex.platform</groupId>
+            <artifactId>platform-utils</artifactId>
+            <scope>provided</scope>
+        </dependency>
 
     </dependencies>
 </project>

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

@@ -2,12 +2,14 @@ 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.serialization.JsonSerializer;
 
 import java.util.Properties;
 
 public class KafkaProducerFactory {
-    public static Producer<String, String> producer(KafkaSettings kafkaSettings) {
+    public static Producer<String, Object> producer(KafkaSettings kafkaSettings) {
         Properties kafkaProps = new Properties();
 
         //Assign localhost id
@@ -27,7 +29,10 @@ public class KafkaProducerFactory {
         //The buffer.memory controls the total amount of memory available to the producer for buffering.
         kafkaProps.put("buffer.memory", kafkaSettings.getBufferMemory());
         kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
-        kafkaProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+        kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class.getName());
+//        kafkaProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+                //ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+        //        prop.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, Employee.class.getName());
 
         return new KafkaProducer<>(kafkaProps);
 

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

@@ -1,12 +1,11 @@
-package ru.spcex.clearing.backendapi.controller.request.cud;
+package ru.spcex.clearing.platform.messaging.domain.cud.securitites;
 
 import com.fasterxml.jackson.annotation.JsonFormat;
 import com.fasterxml.jackson.annotation.JsonProperty;
-import ru.spcex.clearing.backendapi.domain.actions.IAction;
 
 import java.time.Instant;
 
-public class MoneyMarketCreate implements IAction<Object> {
+public class MoneyMarketCreateRequest {
     @JsonFormat(pattern = "yyyy-MM-dd", timezone = "Europe/Moscow")
     @JsonProperty
     public Instant startDate;
@@ -25,11 +24,6 @@ public class MoneyMarketCreate implements IAction<Object> {
     @JsonProperty
     public Double lotSize;
 
-    @Override
-    public Object toRequest() {
-        return new Object();
-    }
-
     public Instant getStartDate() {
         return startDate;
     }

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/serialization/JsonSerializer.java

@@ -0,0 +1,24 @@
+package ru.spcex.clearing.platform.messaging.serialization;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.kafka.common.serialization.Serializer;
+
+import java.nio.charset.StandardCharsets;
+
+public class JsonSerializer implements Serializer<Object> {
+    private final ObjectMapper json;
+
+    public JsonSerializer() {
+        this.json = new ObjectMapper();
+    }
+
+    @Override
+    public byte[] serialize(String topic, Object data) {
+        try {
+            String s = json.writeValueAsString(data);
+            return s.getBytes(StandardCharsets.UTF_8);
+        } catch (Throwable e) {
+            throw new RuntimeException("couldn't serialize kafka message ", e);
+        }
+    }
+}