добавил создание реквестов при запросе в backend-api

добавил транзакции для ImdgProvider

clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/service/impl/OperatorImpl.java

@@ -9,6 +9,8 @@ import ru.spcex.clearing.backendapi.domain.actions.IAction;
 import ru.spcex.clearing.backendapi.errors.ActionValidationException;
 import ru.spcex.clearing.backendapi.service.IOperator;
 import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
+import ru.spcex.clearing.platform.messaging.service.RequestInfo;
+import ru.spcex.platform.imdg.api.Imdg;
 import ru.spcex.platform.imdg.api.ImdgId;
 import ru.spcex.platform.imdg.api.ImdgProvider;
 import ru.spcex.platform.utils.enumeration.EnumMessage;
@@ -21,10 +23,12 @@ import java.util.concurrent.Future;
 public class OperatorImpl implements IOperator {
     private final Producer<String, Object> kafka;
     private final ImdgId idGenerator;
+    private final ImdgProvider imdgProvider;
 
     public OperatorImpl(Producer<String, Object> kafka, ImdgProvider imdgProvider) {
         this.kafka = kafka;
         this.idGenerator = imdgProvider.getImdgIdGenerator();
+        this.imdgProvider = imdgProvider;
     }
 
     @Override
@@ -34,11 +38,19 @@ public class OperatorImpl implements IOperator {
         request.setId(idGenerator.nextId());
         request.setActionType(iAction.getActionType());
         request.setRequestPayload(iAction.toRequest());
+        //сохраняет данные о запросе в хранилище
+        saveRequestToStorage(destination, request);
         Future<RecordMetadata> send = kafka.send(new ProducerRecord<>(destination, request));
         send.get();
         return new QueueSuccessResponse(request.getActionType(), request.getId());
     }
 
+    private void saveRequestToStorage(String destination, BaseRequest<Object> request) {
+        Imdg<RequestInfo> requestStorage = imdgProvider.getImdg(destination, RequestInfo.class);
+        RequestInfo requestInfo = RequestInfo.create(request.getId());
+        requestStorage.insert(requestInfo);
+    }
+
     private void throwValidate(IAction<?> iAction) {
         Collection<EnumMessage> validationErrors = iAction.validate();
         if (validationErrors.size() > 0) {

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

@@ -14,8 +14,12 @@ import ru.spcex.clearing.platform.messaging.domain.cud.common.CommonDeleteReques
 import ru.spcex.clearing.platform.messaging.domain.cud.securitites.MoneyMarketSecurityNewRequest;
 import ru.spcex.clearing.platform.messaging.domain.cud.securitites.MoneyMarketSecurityUpdateRequest;
 import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
+import ru.spcex.clearing.platform.messaging.service.RequestInfo;
+import ru.spcex.clearing.platform.messaging.service.Status;
 import ru.spcex.platform.imdg.api.Imdg;
 import ru.spcex.platform.imdg.api.ImdgProvider;
+import ru.spcex.platform.imdg.api.ImdgTransaction;
+import ru.spcex.platform.utils.log.ExceptionUtils;
 
 import java.math.BigDecimal;
 
@@ -23,10 +27,27 @@ import java.math.BigDecimal;
 public class MoneyMarketSecurityService extends QueueConsumer implements InitializingBean {
     private final Logger log = LoggerFactory.getLogger(getClass());
     private final Imdg<MoneyMarketSecurity> moneyMarketSecurityMap;
+    private final ImdgProvider imdgProvider;
     @Autowired
     public MoneyMarketSecurityService(Consumer<String, Object> kafkaQueue, ImdgProvider imdgProvider) {
         super(kafkaQueue);
         this.moneyMarketSecurityMap = imdgProvider.getImdg(IMDGDistributedNames.Map_MoneyMarketSecurity, MoneyMarketSecurity.class);
+        this.imdgProvider = imdgProvider;
+    }
+
+    @Override
+    protected boolean needsProcessing(String destination, BaseRequest<?> request) {
+        Imdg<RequestInfo> requestSpecificImdg = imdgProvider.getImdg(destination, RequestInfo.class);
+        RequestInfo requestFound = requestSpecificImdg.getSingleObjectByID(request.getId());
+        if (requestFound == null) {
+            log.error("couldn't find request info by id={}, action type={}", request.getId(), request.getActionType());
+            return false;
+        }
+        if (requestFound.getStatus().equals(Status.Error) || requestFound.getStatus().equals(Status.Success)) {
+            log.debug("request id = {} already processed, status {}", request.getId(), requestFound.getStatus());
+            return false;
+        }
+        return true;
     }
 
     @Override
@@ -40,10 +61,14 @@ public class MoneyMarketSecurityService extends QueueConsumer implements Initial
         callback(CommonDeleteRequest.class)
                 .setConsumer(this::deleteMoneyMarket)
                 .forDestination(Consts.DESTINATION_MONEY_MARKET_SECURITY_DELETE, callbacks::put);
+        imdgProvider.waitAvailable();
         init();
     }
 
     private void newMoneyMarket(BaseRequest<MoneyMarketSecurityNewRequest> userRequest) {
+        Imdg<RequestInfo> requestInfoImdg = imdgProvider.getImdg(Consts.DESTINATION_MONEY_MARKET_SECURITY_NEW, RequestInfo.class);
+        RequestInfo reqInfo = requestInfoImdg.getSingleObjectByID(userRequest.getId());
+        ImdgTransaction transaction = imdgProvider.newTransaction();
         MoneyMarketSecurityNewRequest req = userRequest.getRequestPayload();
         log.debug("MoneyMarketSecurityNewRequest received");
         MoneyMarketSecurity mms = new MoneyMarketSecurity();
@@ -54,8 +79,22 @@ public class MoneyMarketSecurityService extends QueueConsumer implements Initial
         mms.setInstrumentType(req.getInstrumentType());
         mms.setFullName(req.getFullName());
         mms.setSecuritySymbol(req.getSecuritySymbol());
-        moneyMarketSecurityMap.insert(mms);
-        log.debug("successfully processed, new id {}", mms.getId());
+        try {
+            transaction.beginTransaction();
+            Imdg<MoneyMarketSecurity> moneyMarketSecurityMap = transaction.getImdg(IMDGDistributedNames.Map_MoneyMarketSecurity, MoneyMarketSecurity.class);
+            Imdg<RequestInfo> reqInfoMap = transaction.getImdg(Consts.DESTINATION_MONEY_MARKET_SECURITY_NEW, RequestInfo.class);
+            //одним скопом выполняем реквест
+            moneyMarketSecurityMap.insert(mms);
+            RequestInfo.update(reqInfo, Status.Success, "success");
+            //и сохраняем обновленный
+            reqInfoMap.insert(reqInfo);
+            transaction.commitTransaction();
+        } catch (Throwable e) {
+            transaction.rollbackTransaction();
+            RequestInfo.update(reqInfo, Status.Error, ExceptionUtils.getStackTrace(e));
+            requestInfoImdg.insert(reqInfo);
+        }
+        log.debug("successfully processed, request id {}, new object id {}", userRequest.getId(), mms.getId());
     }
 
     private void updateMoneyMarket(BaseRequest<MoneyMarketSecurityUpdateRequest> userRequest) {

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

@@ -25,7 +25,9 @@ public class ImdgHazelcast<T extends SpcexObjectBase> implements Imdg<T> {
 
     @Override
     public void insert(T paramT) {
-        paramT.setId(idGenerator.newId());
+        if (paramT.getId() == null) {
+            paramT.setId(idGenerator.newId());
+        }
         map.put(paramT.getId(), paramT);
     }
 

platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/adapter/ImdgTransactionProviderHazelcast.java

@@ -0,0 +1,65 @@
+package ru.spcex.platform.imdg.iml.hazelcast.adapter;
+
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.transaction.TransactionContext;
+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.ImdgId;
+import ru.spcex.platform.imdg.api.ImdgTransaction;
+
+public class ImdgTransactionProviderHazelcast implements ImdgTransaction {
+    private TransactionContext ctx;
+    private HazelcastInstance hz;
+
+    @Override
+    public <T extends SpcexObjectBase> Imdg<T> getImdg(String key, Class<T> clazz) {
+        ImdgTransactionalHazelcast<T> imdg = new ImdgTransactionalHazelcast<>();
+        imdg.setMap(ctx.getMap(key));
+        imdg.setHzInstance(hz);
+        imdg.setMapName(key);
+        imdg.setIdGenerator(hz.getIdGenerator(IMDGDistributedNames.MAP_SEQUENCE_NAME));
+        return imdg;
+    }
+
+    @Override
+    public ImdgId getImdgIdGenerator() {
+        throw new UnsupportedOperationException("no need to create id generator inside a transaction");
+    }
+
+    @Override
+    public ImdgTransaction newTransaction() {
+        throw new UnsupportedOperationException("cannot generate second transactional context");
+    }
+
+    @Override
+    public void commitTransaction() {
+        this.ctx.commitTransaction();
+    }
+
+    @Override
+    public void beginTransaction() {
+        this.ctx.beginTransaction();
+    }
+
+    @Override
+    public void rollbackTransaction() {
+        this.ctx.rollbackTransaction();
+    }
+
+    public TransactionContext getCtx() {
+        return ctx;
+    }
+
+    public void setCtx(TransactionContext ctx) {
+        this.ctx = ctx;
+    }
+
+    public HazelcastInstance getHz() {
+        return hz;
+    }
+
+    public void setHz(HazelcastInstance hz) {
+        this.hz = hz;
+    }
+}

platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/adapter/ImdgTransactionalHazelcast.java

@@ -0,0 +1,148 @@
+package ru.spcex.platform.imdg.iml.hazelcast.adapter;
+
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.core.ILock;
+import com.hazelcast.core.IdGenerator;
+import com.hazelcast.core.TransactionalMap;
+import com.hazelcast.query.Predicate;
+import com.hazelcast.query.Predicates;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import ru.spcex.platform.classes.base.SpcexObjectBase;
+import ru.spcex.platform.imdg.api.Imdg;
+import ru.spcex.platform.utils.log.ExceptionUtils;
+
+import java.util.*;
+
+public class ImdgTransactionalHazelcast<T extends SpcexObjectBase> implements Imdg<T> {
+    private final Logger log = LoggerFactory.getLogger(getClass());
+    private IdGenerator idGenerator;
+
+    private TransactionalMap<Long, T> map;
+
+    private String mapName;
+
+    private HazelcastInstance hzInstance;
+
+    @Override
+    public void insert(T paramT) {
+        if (paramT.getId() == null) {
+            paramT.setId(idGenerator.newId());
+        }
+        map.put(paramT.getId(), paramT);
+    }
+
+    @Override
+    public void lockAndPerform(Runnable operation) {
+        ILock hzLock = null;
+        try {
+            //в качестве альтернативы getCPSubsystem, поддерживается только при 3+ нодах хазелкаста
+            hzLock = hzInstance.getLock(mapName);
+            hzLock.lock();
+            operation.run();
+        } finally {
+            if (hzLock != null) {
+                try {
+                    hzLock.unlock();
+                } catch (Throwable e) {
+                    log.error("{} {}", mapName, ExceptionUtils.getStackTrace(e));
+                }
+            }
+        }
+    }
+
+    @Override
+    public void update(T paramT) {
+        map.put(paramT.getId(), paramT);
+    }
+
+    @Override
+    public void delete(T paramT) {
+        map.remove(paramT.getId());
+    }
+
+    @Override
+    public T getSingleObjectByID(Long paramLong) {
+        return map.get(paramLong);
+    }
+
+    /**
+     * возвращает объект удовлетворяющий условиям
+     * не проверяет, если удовлетворяющих условиям > 1
+     */
+    @SuppressWarnings("unchecked")
+    @Override
+    public T getSingleObjectByFieldValues(Map<String, ? extends Comparable<?>> conditions) {
+        Predicate<Long, ?>[] predicates = new Predicate[conditions.size()];
+        final int[] i = {0};
+        conditions.forEach((key, value) -> {
+            predicates[i[0]] = Predicates.equal(key, value);
+            i[0]++;
+        });
+        Predicate<Long, T> or = Predicates.or(predicates);
+        Set<Long> ids = map.keySet(or);
+        Iterator<Long> idsIterator = ids.iterator();
+        if (idsIterator.hasNext()) {
+            return map.get(idsIterator.next());
+        } else {
+            return null;
+        }
+    }
+
+    /**
+     * возвращает коллекцию объектов, удовлетворяющий условиям на поля
+     */
+    @SuppressWarnings("unchecked")
+    @Override
+    public Collection<T> getCollectionObjectsByFieldValues(Map<String, ? extends Comparable<?>> conditions) {
+        Predicate<Long, ?>[] predicates = new Predicate[conditions.size()];
+        final int[] i = {0};
+        conditions.forEach((key, value) -> {
+            predicates[i[0]] = Predicates.equal(key, value);
+            i[0]++;
+        });
+        Predicate<Long, T> or = Predicates.or(predicates);
+        Set<Long> ids = map.keySet(or);
+        Iterator<Long> idIterator = ids.iterator();
+        Collection<T> searchResult = new ArrayList<>();
+        while(idIterator.hasNext()) {
+            T element = map.get(idIterator.next());
+            if (element != null) {
+                searchResult.add(element);
+            }
+        }
+        return searchResult;
+    }
+
+    public TransactionalMap<Long, T> getMap() {
+        return map;
+    }
+
+    public void setMap(TransactionalMap<Long, T> map) {
+        this.map = map;
+    }
+
+    public IdGenerator getIdGenerator() {
+        return idGenerator;
+    }
+
+    public void setIdGenerator(IdGenerator idGenerator) {
+        this.idGenerator = idGenerator;
+    }
+
+    public HazelcastInstance getHzInstance() {
+        return hzInstance;
+    }
+
+    public void setHzInstance(HazelcastInstance hzInstance) {
+        this.hzInstance = hzInstance;
+    }
+
+    public String getMapName() {
+        return mapName;
+    }
+
+    public void setMapName(String mapName) {
+        this.mapName = mapName;
+    }
+}

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

@@ -5,6 +5,7 @@ import com.hazelcast.client.config.ClientConfig;
 import com.hazelcast.core.HazelcastInstance;
 import com.hazelcast.core.IdGenerator;
 import com.hazelcast.core.LifecycleEvent;
+import com.hazelcast.transaction.TransactionOptions;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.core.env.Environment;
@@ -14,13 +15,16 @@ import ru.spcex.platform.classes.base.SpcexObjectBase;
 import ru.spcex.platform.imdg.api.Imdg;
 import ru.spcex.platform.imdg.api.ImdgId;
 import ru.spcex.platform.imdg.api.ImdgProvider;
+import ru.spcex.platform.imdg.api.ImdgTransaction;
 import ru.spcex.platform.imdg.iml.hazelcast.adapter.ImdgHazelcast;
 import ru.spcex.platform.imdg.iml.hazelcast.adapter.ImdgIdGeneratorHazelcast;
+import ru.spcex.platform.imdg.iml.hazelcast.adapter.ImdgTransactionProviderHazelcast;
 import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams;
 import ru.spcex.platform.imdg.iml.hazelcast.util.HazelcastHelper;
 import ru.spcex.platform.utils.log.ExceptionUtils;
 
 import java.util.*;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 public abstract class HazelcastServiceBase
 //        implements IHazelcastService
@@ -235,6 +239,52 @@ public abstract class HazelcastServiceBase
         return imdg;
     }
 
+
+    @Override
+    public void waitAvailable() {
+        Object lock = new Object();
+        AtomicBoolean hzAvailable = new AtomicBoolean();
+        statusSubscribe(new IHazelcastClusterStatus() {
+            @Override
+            public void getAvailable(HazelcastInstance hazelcastNotInited) {
+                synchronized (lock) {
+                    hzAvailable.set(true);
+                    lock.notifyAll();
+                }
+            }
+            @Override
+            public void getUnavailable(HazelcastInstance hazelcastNotInited) {
+            }
+        });
+        synchronized (lock) {
+            try {
+                if (!hzAvailable.get()) {
+                    lock.wait();
+                }
+            } catch (InterruptedException e) {
+                log.error(ExceptionUtils.getStackTrace(e));
+            }
+        }
+    }
+
+    @Override
+    public ImdgTransaction newTransaction() {
+        ImdgTransactionProviderHazelcast transactionalContext = new ImdgTransactionProviderHazelcast();
+        statusSubscribe(new IHazelcastClusterStatus() {
+            @Override
+            public void getAvailable(HazelcastInstance hazelcastNotInited) {
+                transactionalContext.setHz(hazelcastInstance);
+                transactionalContext.setCtx(hazelcastNotInited.newTransactionContext(
+                        new TransactionOptions().setTransactionType(TransactionOptions.TransactionType.TWO_PHASE)
+                ));
+            }
+            @Override
+            public void getUnavailable(HazelcastInstance hazelcastNotInited) {
+            }
+        });
+        return transactionalContext;
+    }
+
     @Override
     public ImdgId getImdgIdGenerator() {
         ImdgIdGeneratorHazelcast imdgId = new ImdgIdGeneratorHazelcast();

platform-parent/platform-imdg-api/src/main/java/ru/spcex/platform/imdg/api/ImdgProvider.java

@@ -6,8 +6,13 @@ public interface ImdgProvider {
     /**
      * пока простой интерфейс для получения доступа к мапам
      */
-    public <T extends SpcexObjectBase> Imdg<T> getImdg(String key, Class<T> clazz);
+    <T extends SpcexObjectBase> Imdg<T> getImdg(String key, Class<T> clazz);
 
-    public ImdgId getImdgIdGenerator();
+    ImdgId getImdgIdGenerator();
 
+    ImdgTransaction newTransaction();
+
+    default void waitAvailable() {
+        throw new IllegalStateException("waitHzAvailable not implemented");
+    }
 }

platform-parent/platform-imdg-api/src/main/java/ru/spcex/platform/imdg/api/ImdgTransaction.java

@@ -0,0 +1,13 @@
+package ru.spcex.platform.imdg.api;
+
+/**
+ * из TransactionalContext тоже можно доставать мапы
+ * поэтому объект транзацкии тоже является ImdgProvider
+ */
+public interface ImdgTransaction extends ImdgProvider {
+    void commitTransaction();
+
+    void beginTransaction();
+
+    void rollbackTransaction();
+}

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/OffsetChanger.java

@@ -0,0 +1,35 @@
+package ru.spcex.clearing.platform.messaging.service;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+import org.apache.kafka.common.TopicPartition;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class OffsetChanger implements ConsumerRebalanceListener {
+    private final Consumer<String, Object> consumer;
+    private final Collection<String> topics;
+
+    public OffsetChanger(Consumer<String, Object> consumer, Set<String> topics) {
+        this.consumer = consumer;
+        this.topics = topics;
+    }
+
+    @Override
+    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {}
+
+    @Override
+    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
+        Map<TopicPartition, Long> offsets = partitions
+                .stream()
+                .collect(Collectors.toMap(topicPartition -> topicPartition, topicPartition -> System.currentTimeMillis() - 300 * 1000));
+        consumer.offsetsForTimes(offsets)
+                .entrySet()
+                .stream()
+                .filter(entr -> entr.getValue() != null)
+                .forEach(entry -> consumer.seek(entry.getKey(), entry.getValue().offset()));
+    }
+}

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

@@ -39,10 +39,22 @@ public class QueueConsumer implements AutoCloseable {
         this.json = new ObjectMapper();
     }
 
+    protected boolean needsProcessing(String topicName, BaseRequest<?> request) {
+        return true;
+    }
+
+    protected boolean supportStartOffsetTimeWindow() {
+        return false;
+    }
+
     public void init() {
         executor.submit(() -> {
             try {
-                consumer.subscribe(callbacks.keySet());
+                if (supportStartOffsetTimeWindow()) {
+                    consumer.subscribe(callbacks.keySet(), new OffsetChanger(consumer, callbacks.keySet()));
+                } else {
+                    consumer.subscribe(callbacks.keySet());
+                }
                 while (!closed.get()) {
                     try {
                         ConsumerRecords<String, Object> records = consumer.poll(Duration.of(10, ChronoUnit.SECONDS));
@@ -51,7 +63,9 @@ public class QueueConsumer implements AutoCloseable {
                             Class<?> clazz = callback.getClazz();
                             JavaType payloadType = json.getTypeFactory().constructParametricType(BaseRequest.class, clazz);
                             Object o = json.readValue((String) next.value(), payloadType);
-                            callback.acceptRaw(o);
+                            if (needsProcessing(next.topic(), (BaseRequest<?>) o)) {
+                                callback.acceptRaw(o);
+                            }
                         }
                     } catch (Throwable e) {
                         log.error(ExceptionUtils.getStackTrace(e));

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/RequestInfo.java

@@ -0,0 +1,50 @@
+package ru.spcex.clearing.platform.messaging.service;
+
+import ru.spcex.platform.classes.base.SpcexObjectBase;
+
+import java.io.Serializable;
+import java.time.Instant;
+
+public class RequestInfo extends SpcexObjectBase implements Serializable {
+    public static RequestInfo create(Long id) {
+        RequestInfo newReq = new RequestInfo();
+        newReq.created = Instant.now();
+        newReq.status = Status.Processing;
+        newReq.setId(id);
+        return newReq;
+    }
+
+    public static void update(RequestInfo req, Status status, String message) {
+        req.status = status;
+        req.message = message;
+    }
+
+
+    private Status status;
+    private Instant created;
+    private String message;
+
+    public Status getStatus() {
+        return status;
+    }
+
+    public void setStatus(Status status) {
+        this.status = status;
+    }
+
+    public Instant getCreated() {
+        return created;
+    }
+
+    public void setCreated(Instant created) {
+        this.created = created;
+    }
+
+    public String getMessage() {
+        return message;
+    }
+
+    public void setMessage(String message) {
+        this.message = message;
+    }
+}

platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/Status.java

@@ -0,0 +1,5 @@
+package ru.spcex.clearing.platform.messaging.service;
+
+public enum Status {
+    Processing, Success, Error
+}