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

http://jira.mfd.msk:8088/browse/CLS-36

Browse source
This commit is contained in:
ialbert 2022-09-28 18:39:02 +03:00
parent db38a4230a
commit 2e46a484ff
7 changed files with 81 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
clearing-parent/balance-service/src/main/java/ru/spcex/clearing/balance/service/Sdf02Service.java
View file

@ -14,6 +14,8 @@ import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.ImdgProvider;
//fixme remove
@Deprecated
@Service
public class Sdf02Service extends QueueConsumer implements InitializingBean {
private final Logger log = LoggerFactory.getLogger(getClass());

4
clearing-parent/dbf-importer/pom.xml
View file

@ -62,6 +62,10 @@
<groupId>ru.spcex.clearing</groupId>
<artifactId>classes</artifactId>
</dependency>
<dependency>
<groupId>ru.spcex.platform</groupId>
<artifactId>platform-messaging</artifactId>
</dependency>
</dependencies>
<build>

38
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/config/KafkaConfig.java Normal file
View file

@ -0,0 +1,38 @@
package ru.spcex.clearing.dbf.importer.config;
import org.apache.kafka.clients.producer.Producer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import ru.spcex.clearing.dbf.importer.config.settings.ImportDBFServiceSettings;
import ru.spcex.clearing.platform.messaging.config.KafkaProducerFactory;
import ru.spcex.clearing.platform.messaging.service.RequestInfo;
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.ImdgId;
import ru.spcex.platform.imdg.api.ImdgProvider;
@Configuration
public class KafkaConfig {
@Autowired
@Bean
public Producer<String, Object> createProducer(ImportDBFServiceSettings settings) {
return KafkaProducerFactory.producer(settings.getKafkaProducer());
}
@Autowired
@Bean
public KafkaSender kafkaSender(Producer<String, Object> kafkaProducer, ImdgProvider imdgProvider) {
ImdgId imdgIdGenerator = imdgProvider.getImdgIdGenerator();
return KafkaSender
.setup()
.producer(kafkaProducer)
.idGenerator(imdgIdGenerator::nextId)
.imdgProvider(s -> {
Imdg<RequestInfo> imdg = imdgProvider.getImdg(s, RequestInfo.class);
return imdg::insert;
})
.build();
}
}

10
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/config/settings/ImportDBFServiceSettings.java
View file

@ -3,6 +3,7 @@ package ru.spcex.clearing.dbf.importer.config.settings;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.PropertySource;
import org.springframework.stereotype.Component;
import ru.spcex.clearing.platform.messaging.config.element.KafkaProducerSettings;
import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams;
@Component
@ -10,6 +11,7 @@ import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams;
@ConfigurationProperties("import-dbf-service")
public class ImportDBFServiceSettings {
private HazelcastClientParams hazelcast;
private KafkaProducerSettings kafkaProducer;
public HazelcastClientParams getHazelcast() {
return hazelcast;
@ -18,4 +20,12 @@ public class ImportDBFServiceSettings {
public void setHazelcast(HazelcastClientParams hazelcast) {
this.hazelcast = hazelcast;
}
public KafkaProducerSettings getKafkaProducer() {
return kafkaProducer;
}
public void setKafkaProducer(KafkaProducerSettings kafkaProducer) {
this.kafkaProducer = kafkaProducer;
}
}

20
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/logic/stages/ImportToDB.java
View file

@ -8,6 +8,9 @@ import ru.spcex.clearing.dbf.importer.logic.data.enums.ETable;
import ru.spcex.clearing.dbf.importer.logic.data.enums.StageResult;
import ru.spcex.clearing.dbf.importer.logic.data.tables.AbstractTable;
import ru.spcex.clearing.dbf.importer.properties.AProperties;
import ru.spcex.clearing.platform.messaging.domain.Consts;
import ru.spcex.clearing.platform.messaging.domain.cud.balance.StatementRequest;
import ru.spcex.clearing.platform.messaging.service.sender.KafkaSender;
import ru.spcex.platform.imdg.iml.hazelcast.service.HazelcastService;
import java.io.ByteArrayInputStream;
@ -24,12 +27,15 @@ public class ImportToDB extends Stage {
private final AProperties properties;
private final HazelcastService hazelcastService;
private final Map<ETable, AbstractTable> mappingEnumTableObjectTable;
private final KafkaSender kafka;
public ImportToDB(@Qualifier("dbfImporterProperties") AProperties properties,
HazelcastService hazelcastService, @Qualifier("mapOfTable") Map<ETable, AbstractTable> mappingEnumTableObjectTable) {
HazelcastService hazelcastService, @Qualifier("mapOfTable") Map<ETable, AbstractTable> mappingEnumTableObjectTable,
KafkaSender kafka) {
this.properties = properties;
this.hazelcastService = hazelcastService;
this.mappingEnumTableObjectTable = mappingEnumTableObjectTable;
this.kafka = kafka;
}
@Override
@ -43,11 +49,15 @@ public class ImportToDB extends Stage {
AbstractTable table = mappingEnumTableObjectTable.get(currTable);
table.setHazelcastService(hazelcastService);
table.setFilename(resultContainer.getDbfFile().getName());
table.setFileId(hazelcastService.getImdgIdGenerator().nextId());
Long fileId = hazelcastService.getImdgIdGenerator().nextId();
table.setFileId(fileId);
for (int i = 0; i < dbfReader.getRecordCount(); i++) {
Object[] entity = dbfReader.nextRecord();
table.injectEntity(table.getEntity(entity));
}
if (ETable.DF_01.equals(currTable)) {
sendStatementRequest(fileId);
}
} catch (IOException exception) {
log.warn(exception.getMessage());
return StageResult.ERROR;
@ -56,4 +66,10 @@ public class ImportToDB extends Stage {
return StageResult.OK;
}
private void sendStatementRequest(Long fileId) {
StatementRequest statementRequest = new StatementRequest();
statementRequest.setSdf01GroupId(fileId);
kafka.sendRequestToQueue(Consts.STATEMENT_PROCESS, statementRequest);
}
}

10
clearing-parent/dbf-importer/src/main/resources/application.properties
View file

@ -10,4 +10,12 @@ dbf.out-dir=/opt/clearing/file/importer/loaded/
dbf.threads-count=10
import-dbf-service.hazelcast.cluster-members=127.0.0.1:5701
import-dbf-service.hazelcast.login=dev
import-dbf-service.hazelcast.password=dev-pass
import-dbf-service.hazelcast.password=dev-pass
import-dbf-service.kafka-producer.bootstrap-servers=localhost:9092
import-dbf-service.kafka-producer.acks=all
import-dbf-service.kafka-producer.retries=0
import-dbf-service.kafka-producer.batch-size=16384
import-dbf-service.kafka-producer.linger-ms=1
import-dbf-service.kafka-producer.buffer-memory=33554432

1
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/balance/StatementRequest.java
View file

@ -12,7 +12,6 @@ public class StatementRequest {
//from accountBalance creation
@JsonProperty
List<AccountSdf01ToStatementRequestPart> accountCreationResults = new ArrayList<>();
//from sDf02 creation
// public StatementRequestType getType() {
// if (errorCode != null || errorText != null || status != null) {