treshenkov/clearing
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

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

Browse source
This commit is contained in:
ialbert 2022-08-11 15:07:10 +03:00
parent 845d38167b
commit 5e56ac91d7
12 changed files with 286 additions and 30 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
clearing-parent/backend-api/pom.xml
View file

@ -38,8 +38,8 @@
<artifactId>platform-imdg-api-hazelcast-impl</artifactId>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<groupId>ru.spcex.platform</groupId>
<artifactId>platform-messaging</artifactId>
</dependency>
</dependencies>

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

@ -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<String, String> 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());
}
}

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

@ -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

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

@ -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<String, String> kafka;
private final ObjectMapper json;
private final CudMetaService meta;
@Autowired
public CudController(Producer<String, String> 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<IAction<?>> actionClazz = meta.byDestination(destination);
if (actionClazz == null) {
throw new UnsupportedOperationException("unsupported destination " + destination);
}
IAction<?> iAction = json.readValue(body, actionClazz);
Future<RecordMetadata> 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;
}
}
}

96
clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/request/cud/MoneyMarketCreate.java Normal file
View file

@ -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<Object> {
@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;
}
}

5
clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/domain/actions/IAction.java Normal file
View file

@ -0,0 +1,5 @@
package ru.spcex.clearing.backendapi.domain.actions;
public interface IAction<T> {
public T toRequest();
}

22
clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/meta/CudMetaService.java Normal file
View file

@ -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<String, Class<? extends IAction<?>>> mapping;
public CudMetaService() {
this.mapping = new HashMap<>();
this.mapping.put("money-market-security-new", MoneyMarketCreate.class);
}
public <T extends IAction<?>> Class<T> byDestination(String destination) {
return (Class<T>) mapping.get(destination);
}
}

29
platform-parent/platform-messaging/pom.xml Normal file
View file

@ -0,0 +1,29 @@
<?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">
<modelVersion>4.0.0</modelVersion>
<artifactId>platform-messaging</artifactId>
<name>Platform messaging</name>
<packaging>jar</packaging>
<version>1.0.0</version>
<parent>
<artifactId>platform-parent</artifactId>
<groupId>ru.spcex.platform</groupId>
<version>1.0.0</version>
</parent>
<properties>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
</dependency>
</dependencies>
</project>

35
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/config/KafkaProducerFactory.java Normal file
View file

@ -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<String, String> 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);
}
}

2
clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/config/element/KafkaSettings.java → platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/config/element/KafkaSettings.java
View file

@ -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;

1
platform-parent/pom.xml
View file

@ -22,5 +22,6 @@
<module>platform-classes-base</module>
<module>platform-imdg-api-hazelcast-impl</module>
<module>platform-utils</module>
<module>platform-messaging</module>
</modules>
</project>

6
pom.xml
View file

@ -101,6 +101,12 @@
<version>1.0.0</version>
</dependency>
<dependency>
<groupId>ru.spcex.platform</groupId>
<artifactId>platform-messaging</artifactId>
<version>1.0.0</version>
</dependency>
<dependency>
<groupId>com.github.albfernandez</groupId>
<artifactId>javadbf</artifactId>