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

xml-importer saveRequestInfo disabled on kafka

Browse source
This commit is contained in:
Ivan Nikolaev-Axenov 2024-09-12 13:52:35 +03:00
parent 4966cd552e
commit d404d58d6c
3 changed files with 15 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
clearing-parent/xml-importer/src/main/java/ru/spcex/clearing/xml/importer/config/KafkaConfig.java
View file

@ -11,14 +11,11 @@ import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope; import org.springframework.context.annotation.Scope;
import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.core.ProducerFactory;
import ru.spcex.clearing.imdg.IMDGDistributedNames;
import ru.spcex.clearing.platform.messaging.config.KafkaConsumerFactory; import ru.spcex.clearing.platform.messaging.config.KafkaConsumerFactory;
import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory; import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory;
import ru.spcex.clearing.platform.messaging.config.element.KafkaProducerSettings; import ru.spcex.clearing.platform.messaging.config.element.KafkaProducerSettings;
import ru.spcex.clearing.platform.messaging.service.RequestInfo;
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender; import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
import ru.spcex.clearing.xml.importer.config.settings.ImportXMLServiceSettings; import ru.spcex.clearing.xml.importer.config.settings.ImportXMLServiceSettings;
import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.ImdgId; import ru.spcex.platform.imdg.api.ImdgId;
import ru.spcex.platform.imdg.api.ImdgProvider; import ru.spcex.platform.imdg.api.ImdgProvider;
@ -60,10 +57,7 @@ public class KafkaConfig {
.setup() .setup()
.setKafkaTemplate(kafkaTemplate) .setKafkaTemplate(kafkaTemplate)
.idGenerator(imdgIdGenerator::nextId) .idGenerator(imdgIdGenerator::nextId)
.imdgProvider(s -> { .saveRequestInfo(false)
Imdg<RequestInfo> imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class);
return imdg::insert;
})
.build(); .build();
} }
@ -77,10 +71,7 @@ public class KafkaConfig {
.setup() .setup()
.setKafkaTemplate(kafkaTemplate) .setKafkaTemplate(kafkaTemplate)
.idGenerator(imdgIdGenerator::nextId) .idGenerator(imdgIdGenerator::nextId)
.imdgProvider(s -> { .saveRequestInfo(false)
Imdg<RequestInfo> imdg = imdgProvider.getImdg(IMDGDistributedNames.Map_RequestInfo, RequestInfo.class);
return imdg::insert;
})
.build(); .build();
} }

8
clearing-parent/xml-importer/src/main/java/ru/spcex/clearing/xml/importer/logic/steps/ImportToDB.java
View file

@ -99,9 +99,9 @@ public class ImportToDB {
bankAccountNewRequest.setOktmo(accountTagRub.getOktmo()); bankAccountNewRequest.setOktmo(accountTagRub.getOktmo());
// No DISCONTINUATION_DATE in BankAccountNewAction // No DISCONTINUATION_DATE in BankAccountNewAction
kafkaMessenger.sendToBankAccount(bankAccountNewRequest); boolean sendResult = kafkaMessenger.sendToBankAccount(bankAccountNewRequest);
return StageResult.OK; return sendResult ? StageResult.OK : StageResult.ERROR;
} else if (resultContainer.getXmlTable().equals(ETable.ACCOUNT_LIST_CUR)) { } else if (resultContainer.getXmlTable().equals(ETable.ACCOUNT_LIST_CUR)) {
AccountTagCur accountTagCur = ((AccountListCur) resultContainer.getXmlFile()) AccountTagCur accountTagCur = ((AccountListCur) resultContainer.getXmlFile())
.getRegistrator() .getRegistrator()
@ -128,9 +128,9 @@ public class ImportToDB {
bankAccountNewRequest.setIntermediarySwiftCode2(accountTagCur.getIntermediarySwiftCode2()); bankAccountNewRequest.setIntermediarySwiftCode2(accountTagCur.getIntermediarySwiftCode2());
bankAccountNewRequest.setBankAccount2(accountTagCur.getBankAccount2()); bankAccountNewRequest.setBankAccount2(accountTagCur.getBankAccount2());
kafkaMessenger.sendToBankAccount(bankAccountNewRequest); boolean sendResult = kafkaMessenger.sendToBankAccount(bankAccountNewRequest);
return StageResult.OK; return sendResult ? StageResult.OK : StageResult.ERROR;
} }
log.error("Table type {} is not supported", resultContainer.getXmlTable()); log.error("Table type {} is not supported", resultContainer.getXmlTable());

11
clearing-parent/xml-importer/src/main/java/ru/spcex/clearing/xml/importer/logic/steps/XmlImportKafkaMessenger.java
View file

@ -56,9 +56,16 @@ public class XmlImportKafkaMessenger implements InitializingBean {
messengers.put(ETable.DF_57, groupId -> messageStatement(groupId, SdfTable.SDF_57, FileType.SDF)); messengers.put(ETable.DF_57, groupId -> messageStatement(groupId, SdfTable.SDF_57, FileType.SDF));
} }
public void sendToBankAccount(BankAccountNewRequest bankAccountNewRequest) { public boolean sendToBankAccount(BankAccountNewRequest bankAccountNewRequest) {
kafkaSdf.get().sendRequestToQueue(Consts.DESTINATION_BANK_ACCOUNT_NEW, bankAccountNewRequest); try {
kafkaSdf.get().sendRequestToQueue(Consts.DESTINATION_BANK_ACCOUNT_NEW, bankAccountNewRequest);
} catch (Exception e) {
log.error("Error occurred on sending message to kafka {}", e.getMessage(), e);
return false;
}
log.info("Message to {} was sent {}", Consts.DESTINATION_BANK_ACCOUNT_NEW, bankAccountNewRequest); log.info("Message to {} was sent {}", Consts.DESTINATION_BANK_ACCOUNT_NEW, bankAccountNewRequest);
return true;
} }
/** /**