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

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

Browse source
This commit is contained in:
ialbert 2022-12-07 17:37:07 +03:00
parent a5a3136ed3
commit 2de7c9876e
11 changed files with 121 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
clearing-parent/backend-api/src/main/java/ru/spcex/clearing/backendapi/controller/queue/scheduler/LauncherController.java
View file

@ -13,6 +13,7 @@ import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.*;
import ru.clearing.classes.statics.data.scheduler.Launcher;
import ru.clearing.classes.statics.data.user.User;
import ru.clearing.platform.dictionary.AbstractDictionary;
import ru.spcex.clearing.backendapi.controller.queue.AbstractQueueController;
import ru.spcex.clearing.backendapi.controller.request.cud.schedule.LauncherNew;
import ru.spcex.clearing.backendapi.controller.response.BasicSpcexResponse;
@ -23,7 +24,6 @@ import ru.spcex.clearing.backendapi.security.KeycloakUtils;
import ru.spcex.clearing.backendapi.service.IOperator;
import ru.spcex.clearing.backendapi.service.IStateLoader;
import ru.spcex.clearing.imdg.IMDGDistributedNames;
import ru.spcex.clearing.platform.messaging.domain.Consts;
import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.ImdgProvider;
import ru.spcex.platform.utils.enumeration.EnumMessage;
@ -38,12 +38,14 @@ import java.util.concurrent.ExecutionException;
public class LauncherController extends AbstractQueueController {
private final IStateLoader stateLoader;
private final Imdg<User> userImdg;
private final Imdg<AbstractDictionary> taskDictionary;
@Autowired
public LauncherController(IStateLoader stateLoader, IOperator operator, ImdgProvider imdgProvider) {
super(operator);
this.stateLoader = stateLoader;
this.userImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_User, User.class);
this.taskDictionary = imdgProvider.getImdg(IMDGDistributedNames.Map_TaskDictionary, AbstractDictionary.class);
}
@ApiOperation(value = "get all Launchers.")
@ -63,6 +65,10 @@ public class LauncherController extends AbstractQueueController {
@ResponseBody
public CudResponse add(@ApiParam(value = "Код задания из taskDictionary", required = true, example = "ABLK")
@PathVariable("task-code") String dictionaryName) throws ExecutionException, InterruptedException {
AbstractDictionary taskEnum = taskDictionary.getSingleObjectByFieldValues(Map.of("code", dictionaryName));
if (taskEnum == null) {
throw new NotFound404Exception("task dictionary element with code '" + dictionaryName + "'");
}
LauncherNew launcherCommand = new LauncherNew();
Authentication authentication = SecurityContextHolder.getContext().getAuthentication();
String username = KeycloakUtils.getUserNameFromAuthentication(authentication);
@ -72,7 +78,9 @@ public class LauncherController extends AbstractQueueController {
}
launcherCommand.setTask(dictionaryName);
launcherCommand.setUserId(user.getId());
return processRequest(Consts.LAUNCHER_NEW, launcherCommand);
//топики ограничиваются наличием в taskDictionary
//подписываются на разные топики в разных модулях, см. ru.spcex.platform.enumeration.Task#topic
return processRequest("launcher-" + dictionaryName, launcherCommand);
}
@Autowired

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

@ -32,7 +32,7 @@ public class ClearingService {
}
public void paymentUpdateBySdf04(Long sdf04GroupId) {
log.info("adding sdf03/11 from STLD payments task to queue");
log.info("updating payment.transactionStatus by sdf04 task added to queue");
executor.execute(() -> paymentUpdater.updatePayments(sdf04GroupId));
}
}

9
clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/Sdf04Receiver.java → clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/service/EventsReceiver.java
View file

@ -5,12 +5,14 @@ import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;
import ru.spcex.clearing.platform.messaging.domain.Consts;
import ru.spcex.clearing.platform.messaging.domain.cud.clearing.Sdf04Request;
import ru.spcex.clearing.platform.messaging.domain.cud.schedule.LauncherCommandRequest;
import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
import ru.spcex.platform.enumeration.Task;
@Service
public class Sdf04Receiver extends QueueConsumer implements InitializingBean {
public class EventsReceiver extends QueueConsumer implements InitializingBean {
private final ClearingService clearingService;
public Sdf04Receiver(Consumer<String, Object> kafkaQueue, ClearingService clearingService) {
public EventsReceiver(Consumer<String, Object> kafkaQueue, ClearingService clearingService) {
super(kafkaQueue);
this.clearingService = clearingService;
}
@ -23,6 +25,9 @@ public class Sdf04Receiver extends QueueConsumer implements InitializingBean {
clearingService.paymentUpdateBySdf04(requestPayload.getGroupId());
})
.forDestination(Consts.SDF04_PROCESS, callbacks::put);
callback(LauncherCommandRequest.class)
.setConsumer(event -> clearingService.sdfCreate())
.forDestination(Task.createOrder.topic(), callbacks::put);
init();
}
}

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

@ -66,6 +66,10 @@
<groupId>ru.spcex.platform</groupId>
<artifactId>platform-messaging</artifactId>
</dependency>
<dependency>
<groupId>ru.spcex.platform</groupId>
<artifactId>platform-enum</artifactId>
</dependency>
</dependencies>
<build>

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

@ -1,11 +1,15 @@
package ru.spcex.clearing.dbf.importer.config;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.producer.Producer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;
import ru.spcex.clearing.dbf.importer.config.settings.ImportDBFServiceSettings;
import ru.spcex.clearing.imdg.IMDGDistributedNames;
import ru.spcex.clearing.platform.messaging.config.KafkaConsumerFactory;
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;
@ -35,4 +39,11 @@ public class KafkaConfig {
.build();
};
}
@Autowired
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
@Bean
public Consumer<String, Object> createConsumer(ImportDBFServiceSettings settings) {
return KafkaConsumerFactory.consumer(settings.getKafkaConsumer());
}
}

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.KafkaConsumerSettings;
import ru.spcex.clearing.platform.messaging.config.element.KafkaProducerSettings;
import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams;
@ -12,6 +13,7 @@ import ru.spcex.platform.imdg.iml.hazelcast.config.HazelcastClientParams;
public class ImportDBFServiceSettings {
private HazelcastClientParams hazelcast;
private KafkaProducerSettings kafkaProducer;
private KafkaConsumerSettings kafkaConsumer;
private Common common;
private Store store;
private Cron cron;
@ -32,6 +34,14 @@ public class ImportDBFServiceSettings {
this.kafkaProducer = kafkaProducer;
}
public KafkaConsumerSettings getKafkaConsumer() {
return kafkaConsumer;
}
public void setKafkaConsumer(KafkaConsumerSettings kafkaConsumer) {
this.kafkaConsumer = kafkaConsumer;
}
public Common getCommon() {
return common;
}

10
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/DBFImporterService.java
View file

@ -1,5 +1,7 @@
package ru.spcex.clearing.dbf.importer.services;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
@ -16,6 +18,7 @@ import java.util.Map;
@Service("dbfImporterService")
@EnableScheduling
public class DBFImporterService {
private final Logger log = LoggerFactory.getLogger(getClass());
private final FileChecker fileChecker;
private final ThreadPoolTaskExecutor executorService;
private final Processor processor;
@ -30,7 +33,12 @@ public class DBFImporterService {
@Scheduled(cron = "${import-dbf-service.scheduler.check-src-dir-cron}")
public void run() {
Map<ETable, List<File>> newFiles = fileChecker.checkNewFiles();
run(null);
}
public void run(ETable specificTable) {
log.info("performing import {}", specificTable != null ? specificTable.name() : "");
Map<ETable, List<File>> newFiles = fileChecker.checkNewFiles(specificTable);
for (Map.Entry<ETable, List<File>> newFilesEntry : newFiles.entrySet()) {
ETable currTable = newFilesEntry.getKey();
List<File> fileList = newFilesEntry.getValue();

7
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/FileChecker.java
View file

@ -16,6 +16,10 @@ public class FileChecker {
}
public Map<ETable, List<File>> checkNewFiles() {
return checkNewFiles(null);
}
public Map<ETable, List<File>> checkNewFiles(ETable specificTable) {
Map<ETable, List<File>> newFiles = new EnumMap<>(ETable.class);
String srcDir = settings.getStore().getSrcDir();
@ -27,11 +31,10 @@ public class FileChecker {
ETable currTable = ETable.getTableForFilename(dbfFile.getName());
if (currTable == null) continue;
if (specificTable != null && !specificTable.equals(currTable)) continue;
List<File> currList = newFiles.computeIfAbsent(currTable, list -> new LinkedList<>());
currList.add(dbfFile);
}
return newFiles;
}

33
clearing-parent/dbf-importer/src/main/java/ru/spcex/clearing/dbf/importer/services/LauncherCommandReceiver.java Normal file
View file

@ -0,0 +1,33 @@
package ru.spcex.clearing.dbf.importer.services;
import org.apache.kafka.clients.consumer.Consumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import ru.spcex.clearing.dbf.importer.logic.data.enums.ETable;
import ru.spcex.clearing.platform.messaging.domain.cud.schedule.LauncherCommandRequest;
import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
import ru.spcex.platform.enumeration.Task;
@Service
public class LauncherCommandReceiver extends QueueConsumer implements InitializingBean {
private final Logger log = LoggerFactory.getLogger(getClass());
private final DBFImporterService importer;
@Autowired
public LauncherCommandReceiver(Consumer<String, Object> kafkaQueue,
DBFImporterService importer) {
super(kafkaQueue);
this.importer = importer;
}
@Override
public void afterPropertiesSet() {
callback(LauncherCommandRequest.class)
.setConsumer(action -> importer.run(ETable.DF_04))
.forDestination(Task.createOrderConfirm.topic(), callbacks::put);
init();
}
}

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

@ -22,4 +22,12 @@ 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
import-dbf-service.kafka-producer.buffer-memory=33554432
import-dbf-service.kafka-consumer.bootstrap-servers=localhost:9092
import-dbf-service.kafka-consumer.group-id=dev-group-clearing-service
import-dbf-service.kafka-consumer.enable-auto-commit=false
import-dbf-service.kafka-consumer.session-timeout-ms=30000
import-dbf-service.kafka-consumer.auto-offset-reset=latest
import-dbf-service.kafka-consumer.linger-ms=1
import-dbf-service.kafka-consumer.buffer-memory=33554432

22
platform-parent/platform-enum/src/main/java/ru/spcex/platform/enumeration/Task.java Normal file
View file

@ -0,0 +1,22 @@
package ru.spcex.platform.enumeration;
import ru.spcex.platform.utils.enumeration.IEnumKey;
public enum Task implements IEnumKey {
createOrder("CORD"), createOrderConfirm("CORC");
private final String key;
Task(String key) {
this.key = key;
}
@Override
public String getKey() {
return key;
}
public String topic() {
return "launcher-" + getKey();
}
}