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-12 13:17:55 +03:00
parent 007072afa6
commit a8c6ca0945
13 changed files with 122 additions and 71 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
clearing-parent/imdg-util/pom.xml
View file

@ -1,20 +0,0 @@
<?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>imdg-util</artifactId>
<description>Клиентские утилиты для подключения к IMDG хранилищу</description>
<properties>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
</properties>
</project>

7
clearing-parent/imdg/pom.xml
View file

@ -42,13 +42,6 @@
<scope>compile</scope>
</dependency>
<dependency>
<groupId>ru.spcex.clearing</groupId>
<artifactId>imdg-util</artifactId>
<version>SPCEX-1.0.0.0</version>
<scope>compile</scope>
</dependency>
<!-- JDBC -->
<dependency>

1
clearing-parent/pom.xml
View file

@ -20,7 +20,6 @@
<module>classes</module>
<module>backend-api</module>
<module>imdg</module>
<module>imdg-util</module>
<module>db-scripts</module>
<module>dbf-importer</module>
<module>dbf-exporter</module>

4
clearing-parent/securities-service/pom.xml
View file

@ -24,6 +24,10 @@
<groupId>ru.spcex.platform</groupId>
<artifactId>platform-imdg-api-hazelcast-impl</artifactId>
</dependency>
<dependency>
<groupId>ru.spcex.clearing</groupId>
<artifactId>classes</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>

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

@ -4,24 +4,44 @@ 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.clearing.classes.StaticData.Misc.MoneyMarketSecurity;
import ru.spcex.clearing.imdg.IMDGDistributedNames;
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;
import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.ImdgProvider;
import java.math.BigDecimal;
@Service
public class MoneyMarketSecurityService extends QueueConsumer implements InitializingBean {
private final Imdg<MoneyMarketSecurity> moneyMarketSecurityMap;
@Autowired
public MoneyMarketSecurityService(Consumer<String, Object> kafkaQueue) {
public MoneyMarketSecurityService(Consumer<String, Object> kafkaQueue, ImdgProvider imdgProvider) {
super(kafkaQueue);
this.moneyMarketSecurityMap = imdgProvider.getImdg(IMDGDistributedNames.Map_MoneyMarketSecurity, MoneyMarketSecurity.class);
}
@Override
public void afterPropertiesSet() {
callback(Consts.DESTINATION_MONEY_MARKET_SECURITY_NEW, MoneyMarketCreateRequest.class)
.define(this::newMoneyMarket);
callback(MoneyMarketCreateRequest.class)
.setConsumer(this::newMoneyMarket)
.setDestination(Consts.DESTINATION_MONEY_MARKET_SECURITY_NEW, callbacks::put)
.build();
init();
}
private void newMoneyMarket(MoneyMarketCreateRequest req) {
MoneyMarketSecurity mms = new MoneyMarketSecurity();
mms.setStartDate(req.getStartDate());
mms.setEndDate(req.getEndDate());
mms.setNominalValue(BigDecimal.valueOf(req.getNominalValue()));
mms.setNominalCurrency(req.getNominalCurrencyId());
mms.setInstrumentType(req.getInstrumentType());
mms.setFullName(req.getFullName());
mms.setSecuritySymbol(req.getSecuritySymbol());
moneyMarketSecurityMap.insert(mms);
// (req.getLotSize());
}
}

16
platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/adapter/ImdgHazelcast.java
View file

@ -1,10 +1,12 @@
package ru.spcex.platform.imdg.iml.hazelcast.adapter;
import com.hazelcast.core.IMap;
import com.hazelcast.core.IdGenerator;
import ru.spcex.platform.classes.base.SpcexObjectBase;
import ru.spcex.platform.imdg.api.Imdg;
public class ImdgHazelcast<T extends SpcexObjectBase> implements Imdg<T> {
private IdGenerator idGenerator;
private IMap<Long, T> map;
@ -15,5 +17,17 @@ public class ImdgHazelcast<T extends SpcexObjectBase> implements Imdg<T> {
public void setMap(IMap<Long, T> map) {
this.map = map;
}
//todo implement methods
public IdGenerator getIdGenerator() {
return idGenerator;
}
public void setIdGenerator(IdGenerator idGenerator) {
this.idGenerator = idGenerator;
}
@Override
public void insert(T paramT) {
map.put(idGenerator.newId(), paramT);
}
}

4
platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/service/HazelcastServiceBase.java
View file

@ -3,11 +3,13 @@ package ru.spcex.platform.imdg.iml.hazelcast.service;
import com.hazelcast.client.HazelcastClient;
import com.hazelcast.client.config.ClientConfig;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.IdGenerator;
import com.hazelcast.core.LifecycleEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.env.Environment;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import ru.spcex.clearing.imdg.IMDGDistributedNames;
import ru.spcex.platform.classes.base.SpcexObjectBase;
import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.ImdgProvider;
@ -214,7 +216,9 @@ public abstract class HazelcastServiceBase
@Override
public <T extends SpcexObjectBase> Imdg<T> getImdg(String key, Class<T> clazz) {
ImdgHazelcast<T> imdg = new ImdgHazelcast<>();
IdGenerator generator = hazelcastInstance.getIdGenerator(IMDGDistributedNames.MAP_SEQUENCE_NAME);
imdg.setMap(getHazelcast().getMap(key));
imdg.setIdGenerator(generator);
return imdg;
}

0
clearing-parent/imdg-util/src/main/java/ru/spcex/clearing/imdg/IMDGDistributedNames.java → platform-parent/platform-imdg-api/src/main/java/ru/spcex/clearing/imdg/IMDGDistributedNames.java
View file

5
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/logic/functional/Builder.java Normal file
View file

@ -0,0 +1,5 @@
package ru.spcex.clearing.platform.messaging.logic.functional;
public interface Builder<T1> {
ConsumerSpecificClass<T1> build();
}

7
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/logic/functional/BuilderConsumerStep.java Normal file
View file

@ -0,0 +1,7 @@
package ru.spcex.clearing.platform.messaging.logic.functional;
import java.util.function.Consumer;
public interface BuilderConsumerStep<T1> {
BuilderDestinationStep<T1> setConsumer(Consumer<T1> consumer);
}

7
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/logic/functional/BuilderDestinationStep.java Normal file
View file

@ -0,0 +1,7 @@
package ru.spcex.clearing.platform.messaging.logic.functional;
import java.util.function.BiConsumer;
public interface BuilderDestinationStep<T1> {
Builder<T1> setDestination(String destination, BiConsumer<String, ConsumerSpecificClass<?>> mappingFun);
}

47
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/logic/functional/ConsumerSpecificClass.java Normal file
View file

@ -0,0 +1,47 @@
package ru.spcex.clearing.platform.messaging.logic.functional;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
public class ConsumerSpecificClass<T> implements BuilderConsumerStep<T>, BuilderDestinationStep<T>, Builder<T> {
private Class<T> clazz;
private java.util.function.Consumer<T> consumer;
public ConsumerSpecificClass(Class<T> clazz) {
this.clazz = clazz;
}
public Builder<T> setDestination(String destination, BiConsumer<String, ConsumerSpecificClass<?>> fun) {
fun.accept(destination, this);
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;
}
public static <T1> BuilderConsumerStep<T1> build(Class<T1> clazz) {
return new ConsumerSpecificClass<>(clazz);
}
@Override
public BuilderDestinationStep<T> setConsumer(Consumer<T> consumer) {
this.consumer = consumer;
return this;
}
@Override
public ConsumerSpecificClass<T> build() {
return this;
}
}

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

@ -6,6 +6,8 @@ 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 ru.spcex.clearing.platform.messaging.logic.functional.BuilderConsumerStep;
import ru.spcex.clearing.platform.messaging.logic.functional.ConsumerSpecificClass;
import java.time.Duration;
import java.time.temporal.ChronoUnit;
@ -19,7 +21,7 @@ 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;
protected final Map<String, ConsumerSpecificClass<?>> callbacks;
private final ObjectMapper json;
public QueueConsumer(Consumer<String, Object> kafkaQueue) {
@ -29,18 +31,14 @@ public class QueueConsumer implements AutoCloseable {
this.json = new ObjectMapper();
}
protected <T> void addCallBack(String destination, ConsumerWithClass<T> callback) {
this.callbacks.put(destination, callback);
}
public void init() throws Exception {
public void init() {
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());
ConsumerSpecificClass<?> callback = callbacks.get(next.topic());
Object o = json.readValue((String) next.value(), callback.getClazz());
callback.acceptRaw(o);
}
@ -55,42 +53,15 @@ public class QueueConsumer implements AutoCloseable {
});
}
protected <T> BuilderConsumerStep<T> callback(Class<T> clazz) {
return ConsumerSpecificClass.build(clazz);
}
@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;
}
}
}