diff --git a/clearing-parent/backend-api/pom.xml b/clearing-parent/backend-api/pom.xml index a50260658..558019a31 100644 --- a/clearing-parent/backend-api/pom.xml +++ b/clearing-parent/backend-api/pom.xml @@ -38,8 +38,8 @@ platform-imdg-api-hazelcast-impl - org.apache.kafka - kafka-clients + ru.spcex.platform + platform-messaging diff --git a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/config/KafkaConfig.java b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/config/KafkaConfig.java index ee55cf976..3febf0247 100644 --- a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/config/KafkaConfig.java +++ b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/config/KafkaConfig.java @@ -1,14 +1,11 @@ package ru.spcex.clearing.backendapi.config; -import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import ru.spcex.clearing.backendapi.config.element.BackendApiSettings; -import ru.spcex.clearing.backendapi.config.element.KafkaSettings; - -import java.util.Properties; +import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory; @Configuration public class KafkaConfig { @@ -16,29 +13,7 @@ public class KafkaConfig { @Autowired @Bean public Producer createProducer(BackendApiSettings settings) { - KafkaSettings kafkaSettings = settings.getKafka(); - Properties kafkaProps = new Properties(); - - //Assign localhost id - kafkaProps.put("bootstrap.servers", kafkaSettings.getBootstrapServers()); - - //Set acknowledgements for producer requests. - kafkaProps.put("acks", kafkaSettings.getAcks()); - - //If the request fails, the producer can automatically retry, - kafkaProps.put("retries", kafkaSettings.getRetries()); - - //Specify buffer size in config - kafkaProps.put("batch.size", kafkaSettings.getBatchSize()); - - //Reduce the no of requests less than 0 - kafkaProps.put("linger.ms", kafkaSettings.getLingerMs()); - //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"); - - return new KafkaProducer<>(kafkaProps); + return KafkaProducerFactory.producer(settings.getKafka()); } } diff --git a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/config/element/BackendApiSettings.java b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/config/element/BackendApiSettings.java index 4f521ef5f..f4180e7c2 100644 --- a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/config/element/BackendApiSettings.java +++ b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/config/element/BackendApiSettings.java @@ -3,6 +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.platform.imdg.iml.hazelcast.config.HazelcastClientParams; @Component diff --git a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/cud/CudController.java b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/cud/CudController.java new file mode 100644 index 000000000..7d7891d1c --- /dev/null +++ b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/cud/CudController.java @@ -0,0 +1,86 @@ +package ru.spcex.clearing.backendapi.controller.cud; + +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.kafka.clients.producer.Producer; +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.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; + +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; + +@Controller +@RequestMapping("/cud") +public class CudController { + private final Producer kafka; + private final ObjectMapper json; + private final CudMetaService meta; + + @Autowired + public CudController(Producer kafka, CudMetaService meta) { + this.kafka = kafka; + this.meta = meta; + this.json = new ObjectMapper(); + } + + @RequestMapping(value = "/{destination}", method = RequestMethod.POST) + public BasicSpcexResponse add(@PathVariable("destination") String destination, + @RequestBody String body) throws JsonProcessingException, ExecutionException, InterruptedException { + Class> actionClazz = meta.byDestination(destination); + if (actionClazz == null) { + throw new UnsupportedOperationException("unsupported destination " + destination); + } + IAction iAction = json.readValue(body, actionClazz); + + Future send = kafka.send(new ProducerRecord<>(destination, body)); + RecordMetadata kafkaMetaData = send.get(); + MetaDataResponse responseToClient = new MetaDataResponse(); + responseToClient.setOffset(kafkaMetaData.offset()); + responseToClient.setPartition(kafkaMetaData.partition()); + responseToClient.setTopic(kafkaMetaData.topic()); + return responseToClient; + } + + private static class MetaDataResponse extends BasicSpcexResponse { + @JsonProperty + private Long offset; + @JsonProperty + private Integer partition; + @JsonProperty + private String topic; + + public Long getOffset() { + return offset; + } + + public void setOffset(Long offset) { + this.offset = offset; + } + + public Integer getPartition() { + return partition; + } + + public void setPartition(Integer partition) { + this.partition = partition; + } + + public String getTopic() { + return topic; + } + + public void setTopic(String topic) { + this.topic = topic; + } + } +} diff --git a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/request/cud/MoneyMarketCreate.java b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/request/cud/MoneyMarketCreate.java new file mode 100644 index 000000000..ac90c6039 --- /dev/null +++ b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/request/cud/MoneyMarketCreate.java @@ -0,0 +1,96 @@ +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 java.time.Instant; + +public class MoneyMarketCreate implements IAction { + @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 Object toRequest() { + return new Object(); + } + + 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; + } +} diff --git a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/domain/actions/IAction.java b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/domain/actions/IAction.java new file mode 100644 index 000000000..53710222b --- /dev/null +++ b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/domain/actions/IAction.java @@ -0,0 +1,5 @@ +package ru.spcex.clearing.backendapi.domain.actions; + +public interface IAction { + public T toRequest(); +} diff --git a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/meta/CudMetaService.java b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/meta/CudMetaService.java new file mode 100644 index 000000000..3de34ba57 --- /dev/null +++ b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/meta/CudMetaService.java @@ -0,0 +1,22 @@ +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.domain.actions.IAction; + +import java.util.HashMap; +import java.util.Map; + +@Service +public class CudMetaService { + private final Map>> mapping; + + public CudMetaService() { + this.mapping = new HashMap<>(); + this.mapping.put("money-market-security-new", MoneyMarketCreate.class); + } + + public > Class byDestination(String destination) { + return (Class) mapping.get(destination); + } +} diff --git a/platform-parent/platform-messaging/pom.xml b/platform-parent/platform-messaging/pom.xml new file mode 100644 index 000000000..9266c472f --- /dev/null +++ b/platform-parent/platform-messaging/pom.xml @@ -0,0 +1,29 @@ + + + 4.0.0 + platform-messaging + Platform messaging + jar + 1.0.0 + + + platform-parent + ru.spcex.platform + 1.0.0 + + + + 17 + 17 + + + + + org.apache.kafka + kafka-clients + + + + \ No newline at end of file diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/config/KafkaProducerFactory.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/config/KafkaProducerFactory.java new file mode 100644 index 000000000..dd533d501 --- /dev/null +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/config/KafkaProducerFactory.java @@ -0,0 +1,35 @@ +package ru.spcex.clearing.platform.messaging.config; + +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.Producer; +import ru.spcex.clearing.platform.messaging.config.element.KafkaSettings; + +import java.util.Properties; + +public class KafkaProducerFactory { + public static Producer producer(KafkaSettings kafkaSettings) { + Properties kafkaProps = new Properties(); + + //Assign localhost id + kafkaProps.put("bootstrap.servers", kafkaSettings.getBootstrapServers()); + + //Set acknowledgements for producer requests. + kafkaProps.put("acks", kafkaSettings.getAcks()); + + //If the request fails, the producer can automatically retry, + kafkaProps.put("retries", kafkaSettings.getRetries()); + + //Specify buffer size in config + kafkaProps.put("batch.size", kafkaSettings.getBatchSize()); + + //Reduce the no of requests less than 0 + kafkaProps.put("linger.ms", kafkaSettings.getLingerMs()); + //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"); + + return new KafkaProducer<>(kafkaProps); + + } +} diff --git a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/config/element/KafkaSettings.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/config/element/KafkaSettings.java similarity index 95% rename from clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/config/element/KafkaSettings.java rename to platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/config/element/KafkaSettings.java index dbfacdb87..27442b8d5 100644 --- a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/config/element/KafkaSettings.java +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/config/element/KafkaSettings.java @@ -1,4 +1,4 @@ -package ru.spcex.clearing.backendapi.config.element; +package ru.spcex.clearing.platform.messaging.config.element; public class KafkaSettings { private String bootstrapServers; diff --git a/platform-parent/pom.xml b/platform-parent/pom.xml index a9dae8e33..18e173035 100644 --- a/platform-parent/pom.xml +++ b/platform-parent/pom.xml @@ -22,5 +22,6 @@ platform-classes-base platform-imdg-api-hazelcast-impl platform-utils + platform-messaging \ No newline at end of file diff --git a/pom.xml b/pom.xml index 2dcaca762..bef2f8875 100644 --- a/pom.xml +++ b/pom.xml @@ -101,6 +101,12 @@ 1.0.0 + + ru.spcex.platform + platform-messaging + 1.0.0 + + com.github.albfernandez javadbf