diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/ClearingService.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/ClearingService.java index 8706f32d5..9de8bc295 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/ClearingService.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/ClearingService.java @@ -8,6 +8,7 @@ import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.stereotype.Service; import ru.spcex.clearing.platform.messaging.domain.BaseRequest; import ru.spcex.clearing.platform.messaging.domain.cud.common.CommonIdRequest; +import ru.spcex.clearing.platform.messaging.domain.cud.utilities.STradesImportedRequest; import ru.spcex.clearing.service.execution.ExecutionCurrencyComponent; import ru.spcex.clearing.service.execution.ExecutionDepositComponent; import ru.spcex.clearing.service.execution.ExecutionFondComponent; @@ -105,12 +106,14 @@ public class ClearingService implements DisposableBean { } } - public void executeSTrade() { + public void executeSTrade(STradesImportedRequest eventBody) { executor.execute(() -> { try { - executionDepositComponent.processNewTS(); - executionFondComponent.processNewTS(); - executionCurrencyComponent.processNewTS(); + Long fromId = eventBody == null ? null : eventBody.getMinId(); + log.trace("executeSTrade(fromId={})", fromId); + executionDepositComponent.processNewTS(fromId); + executionFondComponent.processNewTS(fromId); + executionCurrencyComponent.processNewTS(fromId); } catch (Throwable e) { log.error("{}", ExceptionUtils.getStackTrace(e)); } diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/EventsReceiver.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/EventsReceiver.java index e5001fcba..048997fa2 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/EventsReceiver.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/EventsReceiver.java @@ -173,7 +173,7 @@ public class EventsReceiver extends QueueConsumerV2 implements InitializingBean .forDestination(Consts.ASSET_OPERATION_APPROVAL, callbacks::put); callback(STradesImportedRequest.class) - .setConsumer(event -> clearingService.executeSTrade()) + .setConsumer(event -> clearingService.executeSTrade(event.getRequestPayload())) .forDestination(S_TRADES_IMPORTED, callbacks::put); callback(CreateRegistryRequest.class) .setConsumer(event -> registryService.updateRegistryIfNeeded(event.getRequestPayload())) diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionCurrencyComponent.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionCurrencyComponent.java index f0200fa4a..53380373a 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionCurrencyComponent.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionCurrencyComponent.java @@ -33,6 +33,7 @@ import ru.spcex.platform.enumeration.Section; import ru.spcex.platform.enumeration.Side; import ru.spcex.platform.imdg.api.Imdg; import ru.spcex.platform.imdg.api.ImdgProvider; +import ru.spcex.platform.imdg.api.predicate.ImdgPredicate; import ru.spcex.platform.imdg.api.predicate.ImdgPredicateBuilder; import ru.spcex.platform.utils.enumeration.EnumMessage; import ru.spcex.platform.utils.enumeration.IEnumId; @@ -92,15 +93,26 @@ public class ExecutionCurrencyComponent { // log.info("Reset trading day for search STrade: tradeNum={}, tradeDat={}", tradeNum, tradingDay); // } - public void processNewTS() { + public void processNewTS(Long fromId) { LocalDate today = LocalDate.now(); - log.debug("Start check new S_TRADE at {}", today); + log.debug("Start check new S_TRADE at {}, fromId={}", today, fromId); //выбираем STrades на сегодня с правильным section - Collection sTrades = sTradeImdg.getCollectionObjectsByFieldValues(Map.of( - "tradeDate", today, - "section", Section.CURR.getKey() - )); - log.info("Found {} s_trade for today", sTrades.size()); + Collection sTrades; + if (fromId == null) { + sTrades = sTradeImdg.getCollectionObjectsByFieldValues(Map.of( + "tradeDate", LocalDate.now(), + "section", Section.CURR.getKey() + )); + } else { + ImdgPredicateBuilder pb = sTradeImdg.predicateBuilder(); + ImdgPredicate filter = pb.and( + pb.equals("tradeDate", LocalDate.now()), + pb.equals("section", Section.CURR.getKey()), + pb.greatEqual("id", fromId) + ); + sTrades = sTradeImdg.getCollectionObjectsByPredicate(filter); + } + log.info("Found {} s_trade for today (fromId={})", sTrades.size(), fromId); if (sTrades.isEmpty()) { logError(ClearingError.NewDealsNotFound); diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionDepositComponent.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionDepositComponent.java index 0d1fa7746..337b1db81 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionDepositComponent.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionDepositComponent.java @@ -35,6 +35,7 @@ import ru.spcex.platform.enumeration.Section; import ru.spcex.platform.enumeration.Side; import ru.spcex.platform.imdg.api.Imdg; import ru.spcex.platform.imdg.api.ImdgProvider; +import ru.spcex.platform.imdg.api.predicate.ImdgPredicate; import ru.spcex.platform.imdg.api.predicate.ImdgPredicateBuilder; import ru.spcex.platform.utils.enumeration.EnumMessage; import ru.spcex.platform.utils.enumeration.IEnumId; @@ -96,14 +97,25 @@ public class ExecutionDepositComponent { log.info("Reset trading day for search STrade: tradeNum={}, tradeDat={}", tradeNum, tradingDay); } - public void processNewTS() { - log.debug("Start check new S_TRADE after {}", tradingDay); + public void processNewTS(Long fromId) { + log.debug("Start check new S_TRADE after {}, fromId={}", tradingDay, fromId); //выбираем STrades на сегодня с правильным section - Collection sTrades = sTradeImdg.getCollectionObjectsByFieldValues(Map.of( - "tradeDate", LocalDate.now(), - "section", Section.MKR.getKey() - )); - log.info("Found {} s_trade for today", sTrades.size()); + Collection sTrades; + if (fromId == null) { + sTrades = sTradeImdg.getCollectionObjectsByFieldValues(Map.of( + "tradeDate", LocalDate.now(), + "section", Section.MKR.getKey() + )); + } else { + ImdgPredicateBuilder pb = sTradeImdg.predicateBuilder(); + ImdgPredicate filter = pb.and( + pb.equals("tradeDate", LocalDate.now()), + pb.equals("section", Section.MKR.getKey()), + pb.greatEqual("id", fromId) + ); + sTrades = sTradeImdg.getCollectionObjectsByPredicate(filter); + } + log.info("Found {} s_trade for today (fromId={})", sTrades.size(), fromId); if (sTrades.isEmpty()) { logError(ClearingError.NewDealsNotFound); diff --git a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionFondComponent.java b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionFondComponent.java index 70fb5a45b..471daad2b 100644 --- a/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionFondComponent.java +++ b/clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/execution/ExecutionFondComponent.java @@ -116,14 +116,25 @@ public class ExecutionFondComponent { log.info("Reset trading day for search STrade: tradeNum={}, tradeDat={}", tradeNum, tradingDay); } - public void processNewTS() { - log.debug("Start check new S_TRADE after {}", tradingDay); + public void processNewTS(Long fromId) { + log.debug("Start check new S_TRADE after {}, fromId={}", tradingDay, fromId); //выбираем STrades на сегодня с правильным section - Collection sTrades = sTradeImdg.getCollectionObjectsByFieldValues(Map.of( - "tradeDate", LocalDate.now(), - "section", Section.FOND.getKey() - )); - log.info("Found {} s_trade for today", sTrades.size()); + Collection sTrades; + if (fromId == null) { + sTrades = sTradeImdg.getCollectionObjectsByFieldValues(Map.of( + "tradeDate", LocalDate.now(), + "section", Section.FOND.getKey() + )); + } else { + ImdgPredicateBuilder pb = sTradeImdg.predicateBuilder(); + ImdgPredicate filter = pb.and( + pb.equals("tradeDate", LocalDate.now()), + pb.equals("section", Section.FOND.getKey()), + pb.greatEqual("id", fromId) + ); + sTrades = sTradeImdg.getCollectionObjectsByPredicate(filter); + } + log.info("Found {} s_trade for today (fromId={})", sTrades.size(), fromId); if (sTrades.isEmpty()) { logError(ClearingError.NewDealsNotFound); diff --git a/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/services/TradeImporterService.java b/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/services/TradeImporterService.java index 83182926c..cd8eaa510 100644 --- a/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/services/TradeImporterService.java +++ b/clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/services/TradeImporterService.java @@ -77,6 +77,7 @@ public class TradeImporterService { AtomicLong rows = new AtomicLong(); AtomicLong created = new AtomicLong(); AtomicLong updated = new AtomicLong(); + AtomicLong minCreatedId = new AtomicLong(Long.MAX_VALUE); final long logProgressTime = 60_000L; // интервал вывода в лог каждую минуту AtomicLong logTime = new AtomicLong(System.currentTimeMillis() + logProgressTime); String query = String.format("SELECT * FROM %s.Trades", schema); @@ -104,6 +105,9 @@ public class TradeImporterService { tradesDb.setUpdated(currentInstant); sTradesImdg.insert(tradesDb); created.incrementAndGet(); + if (minCreatedId.get() > tradesDb.getId()) { + minCreatedId.set(tradesDb.getId()); + } } } else { log.warn(messageResolver.resolve(new EnumMessage(sTradesNotValid, tradesDb.getTradeNum()))); @@ -119,6 +123,9 @@ public class TradeImporterService { log.debug("Successfully import STrades from DB by scheduled. Send to kafka command, topic={}", S_TRADES_IMPORTED); } STradesImportedRequest sTradesImportedRequest = new STradesImportedRequest(); + if (!byCommand) { + sTradesImportedRequest.setMinId(minCreatedId.get()); + } kafka.get().sendRequestToQueue(S_TRADES_IMPORTED, sTradesImportedRequest); } else { log.debug("No STrades were created or updated from DB. Kafka command will not be send"); diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/utilities/STradesImportedRequest.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/utilities/STradesImportedRequest.java index 30383c910..96b2d936f 100644 --- a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/utilities/STradesImportedRequest.java +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/utilities/STradesImportedRequest.java @@ -5,6 +5,8 @@ import com.fasterxml.jackson.annotation.JsonProperty; public class STradesImportedRequest { @JsonProperty private Long tradeNum; + @JsonProperty + private Long minId; public Long getTradeNum() { return tradeNum; @@ -13,4 +15,12 @@ public class STradesImportedRequest { public void setTradeNum(Long tradeNum) { this.tradeNum = tradeNum; } + + public Long getMinId() { + return minId; + } + + public void setMinId(Long minId) { + this.minId = minId; + } }