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

http://jira.mfd.msk:8088/browse/CLS-439 обновлять STrades только по команде (не по шкедулеру).

Browse source 
команда на клиринг сервис отправляется только в случае если было обновление или создание strades
This commit is contained in:
ialbert 2023-07-26 12:22:45 +03:00
parent b36ea99b28
commit 87c114370c
1 changed files with 12 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
clearing-parent/trade-importer/src/main/java/ru/spcex/clearing/trade/importer/services/TradeImporterService.java
View file

@ -59,11 +59,11 @@ public class TradeImporterService {
} }
@Scheduled(cron = "${trade-importer.cron.load-from-db-cron}") @Scheduled(cron = "${trade-importer.cron.load-from-db-cron}")
public void run() { public synchronized void run() {
process(false); process(false);
} }
public void process(boolean byCommand) { public synchronized void process(boolean byCommand) {
log.debug("Start process import STrades from DB"); log.debug("Start process import STrades from DB");
Collection<STrades> tradesFromDB = jdbcTemplate.query(String.format("SELECT * FROM %s.Trades", schema), Collection<STrades> tradesFromDB = jdbcTemplate.query(String.format("SELECT * FROM %s.Trades", schema),
(resultSet, i) -> readSTrades(resultSet)); (resultSet, i) -> readSTrades(resultSet));
@ -75,7 +75,7 @@ public class TradeImporterService {
for (STrades tradesDb : tradesFromDB) { for (STrades tradesDb : tradesFromDB) {
if (isValidTrades(tradesDb) && fillNessessaryFields(tradesDb)) { if (isValidTrades(tradesDb) && fillNessessaryFields(tradesDb)) {
STrades sTrades = getSTradesFromImdg(tradesDb, sTradesImdg); STrades sTrades = getSTradesFromImdg(tradesDb, sTradesImdg);
if (sTrades != null) { if (sTrades != null && byCommand) {
// Обновление всех полей объекта из БД // Обновление всех полей объекта из БД
tradesDb.setId(sTrades.getId()); tradesDb.setId(sTrades.getId());
tradesDb.setCreated(sTrades.getCreated()); tradesDb.setCreated(sTrades.getCreated());
@ -94,14 +94,17 @@ public class TradeImporterService {
} }
} }
log.debug("Created {} new STrades, {} updated.", created, updated); log.debug("Created {} new STrades, {} updated.", created, updated);
if (created > 0 || updated > 0) {
if (byCommand) { if (byCommand) {
log.debug("Successfully import STrades from DB by command. Send to kafka command, topic={}", S_TRADES_IMPORTED); log.debug("Successfully import STrades from DB by command. Send to kafka command, topic={}", S_TRADES_IMPORTED);
} else {
log.debug("Successfully import STrades from DB by scheduled. Send to kafka command, topic={}", S_TRADES_IMPORTED);
}
STradesImportedRequest sTradesImportedRequest = new STradesImportedRequest();
kafka.get().sendRequestToQueue(S_TRADES_IMPORTED, sTradesImportedRequest);
} else { } else {
log.debug("Successfully import STrades from DB by scheduled. Send to kafka command, topic={}", S_TRADES_IMPORTED); log.debug("No STrades were created or updated from DB. Kafka command will not be send");
} }
STradesImportedRequest sTradesImportedRequest = new STradesImportedRequest();
kafka.get().sendRequestToQueue(S_TRADES_IMPORTED, sTradesImportedRequest);
} }
/** /**