From 1a005ae58404f8dd2cb993c92989244f7511426e Mon Sep 17 00:00:00 2001 From: ialbert Date: Tue, 13 Sep 2022 20:36:43 +0300 Subject: [PATCH] =?UTF-8?q?=D0=B4=D0=BE=D0=B1=D0=B0=D0=B2=D0=B8=D0=BB=20?= =?UTF-8?q?=D1=81=D0=BE=D0=B7=D0=B4=D0=B0=D0=BD=D0=B8=D0=B5=20=D1=80=D0=B5?= =?UTF-8?q?=D0=BA=D0=B2=D0=B5=D1=81=D1=82=D0=BE=D0=B2=20=D0=BF=D1=80=D0=B8?= =?UTF-8?q?=20=D0=B7=D0=B0=D0=BF=D1=80=D0=BE=D1=81=D0=B5=20=D0=B2=20backen?= =?UTF-8?q?d-api=20=D0=B4=D0=BE=D0=B1=D0=B0=D0=B2=D0=B8=D0=BB=20=D1=82?= =?UTF-8?q?=D1=80=D0=B0=D0=BD=D0=B7=D0=B0=D0=BA=D1=86=D0=B8=D0=B8=20=D0=B4?= =?UTF-8?q?=D0=BB=D1=8F=20ImdgProvider?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../backendapi/service/impl/OperatorImpl.java | 12 ++ .../cud/MoneyMarketSecurityService.java | 43 ++++- .../iml/hazelcast/adapter/ImdgHazelcast.java | 4 +- .../ImdgTransactionProviderHazelcast.java | 65 ++++++++ .../adapter/ImdgTransactionalHazelcast.java | 148 ++++++++++++++++++ .../service/HazelcastServiceBase.java | 50 ++++++ .../spcex/platform/imdg/api/ImdgProvider.java | 9 +- .../platform/imdg/api/ImdgTransaction.java | 13 ++ .../messaging/service/OffsetChanger.java | 35 +++++ .../messaging/service/QueueConsumer.java | 18 ++- .../messaging/service/RequestInfo.java | 50 ++++++ .../platform/messaging/service/Status.java | 5 + 12 files changed, 445 insertions(+), 7 deletions(-) create mode 100644 platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/adapter/ImdgTransactionProviderHazelcast.java create mode 100644 platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/adapter/ImdgTransactionalHazelcast.java create mode 100644 platform-parent/platform-imdg-api/src/main/java/ru/spcex/platform/imdg/api/ImdgTransaction.java create mode 100644 platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/OffsetChanger.java create mode 100644 platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/RequestInfo.java create mode 100644 platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/Status.java diff --git a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/service/impl/OperatorImpl.java b/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/service/impl/OperatorImpl.java index 6617e0319..dabfee7e6 100644 --- a/clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/service/impl/OperatorImpl.java +++ b/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 kafka; private final ImdgId idGenerator; + private final ImdgProvider imdgProvider; public OperatorImpl(Producer 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 send = kafka.send(new ProducerRecord<>(destination, request)); send.get(); return new QueueSuccessResponse(request.getActionType(), request.getId()); } + private void saveRequestToStorage(String destination, BaseRequest request) { + Imdg requestStorage = imdgProvider.getImdg(destination, RequestInfo.class); + RequestInfo requestInfo = RequestInfo.create(request.getId()); + requestStorage.insert(requestInfo); + } + private void throwValidate(IAction iAction) { Collection validationErrors = iAction.validate(); if (validationErrors.size() > 0) { diff --git a/clearing-parent/securities-service/src/main/java/ru/spcex/clearing/securities/service/cud/MoneyMarketSecurityService.java b/clearing-parent/securities-service/src/main/java/ru/spcex/clearing/securities/service/cud/MoneyMarketSecurityService.java index 33f330bc6..279c09bd5 100644 --- a/clearing-parent/securities-service/src/main/java/ru/spcex/clearing/securities/service/cud/MoneyMarketSecurityService.java +++ b/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 moneyMarketSecurityMap; + private final ImdgProvider imdgProvider; @Autowired public MoneyMarketSecurityService(Consumer 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 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 userRequest) { + Imdg 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 moneyMarketSecurityMap = transaction.getImdg(IMDGDistributedNames.Map_MoneyMarketSecurity, MoneyMarketSecurity.class); + Imdg 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 userRequest) { diff --git a/platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/adapter/ImdgHazelcast.java b/platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/adapter/ImdgHazelcast.java index c1e19049e..45ca264ae 100644 --- a/platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/adapter/ImdgHazelcast.java +++ b/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 implements Imdg { @Override public void insert(T paramT) { - paramT.setId(idGenerator.newId()); + if (paramT.getId() == null) { + paramT.setId(idGenerator.newId()); + } map.put(paramT.getId(), paramT); } diff --git a/platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/adapter/ImdgTransactionProviderHazelcast.java b/platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/adapter/ImdgTransactionProviderHazelcast.java new file mode 100644 index 000000000..f36ccea77 --- /dev/null +++ b/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 Imdg getImdg(String key, Class clazz) { + ImdgTransactionalHazelcast 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; + } +} diff --git a/platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/adapter/ImdgTransactionalHazelcast.java b/platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/adapter/ImdgTransactionalHazelcast.java new file mode 100644 index 000000000..0aca655c3 --- /dev/null +++ b/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 implements Imdg { + private final Logger log = LoggerFactory.getLogger(getClass()); + private IdGenerator idGenerator; + + private TransactionalMap 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> conditions) { + Predicate[] predicates = new Predicate[conditions.size()]; + final int[] i = {0}; + conditions.forEach((key, value) -> { + predicates[i[0]] = Predicates.equal(key, value); + i[0]++; + }); + Predicate or = Predicates.or(predicates); + Set ids = map.keySet(or); + Iterator idsIterator = ids.iterator(); + if (idsIterator.hasNext()) { + return map.get(idsIterator.next()); + } else { + return null; + } + } + + /** + * возвращает коллекцию объектов, удовлетворяющий условиям на поля + */ + @SuppressWarnings("unchecked") + @Override + public Collection getCollectionObjectsByFieldValues(Map> conditions) { + Predicate[] predicates = new Predicate[conditions.size()]; + final int[] i = {0}; + conditions.forEach((key, value) -> { + predicates[i[0]] = Predicates.equal(key, value); + i[0]++; + }); + Predicate or = Predicates.or(predicates); + Set ids = map.keySet(or); + Iterator idIterator = ids.iterator(); + Collection searchResult = new ArrayList<>(); + while(idIterator.hasNext()) { + T element = map.get(idIterator.next()); + if (element != null) { + searchResult.add(element); + } + } + return searchResult; + } + + public TransactionalMap getMap() { + return map; + } + + public void setMap(TransactionalMap 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; + } +} diff --git a/platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/service/HazelcastServiceBase.java b/platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/service/HazelcastServiceBase.java index dcf70320b..e68defd6d 100644 --- a/platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/service/HazelcastServiceBase.java +++ b/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(); diff --git a/platform-parent/platform-imdg-api/src/main/java/ru/spcex/platform/imdg/api/ImdgProvider.java b/platform-parent/platform-imdg-api/src/main/java/ru/spcex/platform/imdg/api/ImdgProvider.java index 3ccd0405e..4ff60bf67 100644 --- a/platform-parent/platform-imdg-api/src/main/java/ru/spcex/platform/imdg/api/ImdgProvider.java +++ b/platform-parent/platform-imdg-api/src/main/java/ru/spcex/platform/imdg/api/ImdgProvider.java @@ -6,8 +6,13 @@ public interface ImdgProvider { /** * пока простой интерфейс для получения доступа к мапам */ - public Imdg getImdg(String key, Class clazz); + Imdg getImdg(String key, Class clazz); - public ImdgId getImdgIdGenerator(); + ImdgId getImdgIdGenerator(); + ImdgTransaction newTransaction(); + + default void waitAvailable() { + throw new IllegalStateException("waitHzAvailable not implemented"); + } } diff --git a/platform-parent/platform-imdg-api/src/main/java/ru/spcex/platform/imdg/api/ImdgTransaction.java b/platform-parent/platform-imdg-api/src/main/java/ru/spcex/platform/imdg/api/ImdgTransaction.java new file mode 100644 index 000000000..bd1f2eb32 --- /dev/null +++ b/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(); +} diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/OffsetChanger.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/OffsetChanger.java new file mode 100644 index 000000000..9995a51d3 --- /dev/null +++ b/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 consumer; + private final Collection topics; + + public OffsetChanger(Consumer consumer, Set topics) { + this.consumer = consumer; + this.topics = topics; + } + + @Override + public void onPartitionsRevoked(Collection partitions) {} + + @Override + public void onPartitionsAssigned(Collection partitions) { + Map 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())); + } +} diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/QueueConsumer.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/QueueConsumer.java index 2bcc35950..7a1f96915 100644 --- a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/QueueConsumer.java +++ b/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 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)); diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/RequestInfo.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/RequestInfo.java new file mode 100644 index 000000000..eb93f8a2e --- /dev/null +++ b/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; + } +} \ No newline at end of file diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/Status.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/Status.java new file mode 100644 index 000000000..f012a03d1 --- /dev/null +++ b/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 +}