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

scheduler-service LauncherSender поправил передачу дополнительных параметров CLS-293

Browse source
This commit is contained in:
AKurakin 2023-09-12 11:20:06 +03:00
parent b2d0d0c742
commit a3ac454cb3
3 changed files with 17 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
clearing-parent/scheduler-service/src/main/java/ru/spcex/clearing/scheduler/service/LauncherSender.java
View file

@ -8,6 +8,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import ru.clearing.classes.statics.data.scheduler.PlannerAllToday;
import ru.spcex.clearing.platform.messaging.domain.ActionType;
import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
import ru.spcex.clearing.platform.messaging.domain.cud.schedule.LauncherCommandRequest;
@ -30,10 +31,16 @@ public class LauncherSender {
this.idGenerator = imdgProvider.getImdgIdGenerator();
}
protected BaseRequest<Object> makeCmdRequest(Task byTask, Long userId) {
protected BaseRequest<Object> makeCmdRequest(PlannerAllToday byTask, Long userId) {
Object toRequest;
LauncherCommandRequest taskRunnerCommandRequest = new LauncherCommandRequest();
taskRunnerCommandRequest.setTaskName(byTask.getKey());
taskRunnerCommandRequest.setTaskName(byTask.getTask());
// taskRunnerCommandRequest.setMarket(byTask.getMarket());
taskRunnerCommandRequest.setSection(byTask.getSection());
taskRunnerCommandRequest.setSessionType(byTask.getSessionType());
taskRunnerCommandRequest.setCompanyId(byTask.getCompanyId());
taskRunnerCommandRequest.setSecurityId(byTask.getSecurityId());
taskRunnerCommandRequest.setUserId(userId);
toRequest = taskRunnerCommandRequest;
BaseRequest<Object> request = new BaseRequest<>();
@ -44,8 +51,8 @@ public class LauncherSender {
return request;
}
public void sendCommandToQueue(Task toTaskQueue, Long userId) {
BaseRequest<Object> request = makeCmdRequest(toTaskQueue, userId);
public void sendCommandToQueue(Task toTaskQueue, PlannerAllToday task, Long userId) {
BaseRequest<Object> request = makeCmdRequest(task, userId);
String destination = toTaskQueue.topic(); // "launcher-" + getKey()
log.debug("Send command {} to {}", toTaskQueue, destination);
Future<RecordMetadata> send = kafka.send(new ProducerRecord<>(destination, request));

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

@ -228,7 +228,7 @@ public class TaskManager implements InitializingBean, AutoCloseable {
launcher.setUpdated(created);
launcherMap.insert(launcher);
// 2. отправить сообщение
launcherSender.sendCommandToQueue(taskE, systemId);
launcherSender.sendCommandToQueue(taskE, task, systemId);
log.debug("Successfully processed, new id {}", launcher.getId());
}
}

10
clearing-parent/scheduler-service/src/test/java/ru/spcex/clearing/scheduler/service/TaskManagerTest.java
View file

@ -48,7 +48,7 @@ class TaskManagerTest extends AbstractServiceTest {
//ACT
addToPlannerQueue(TaskManager.Process.add, plannerAllToday);
//ASSERT
waitingWhenAddedLauncherCommandRequestAndCheckIt(getEnumByKey(Task.class, plannerAllToday.getTask()), systemId);
waitingWhenAddedLauncherCommandRequestAndCheckIt(getEnumByKey(Task.class, plannerAllToday.getTask()), plannerAllToday, systemId);
}
/**
@ -67,7 +67,7 @@ class TaskManagerTest extends AbstractServiceTest {
//ACT
addToPlannerQueue(TaskManager.Process.update, oldPlanner);
//ASSERT
waitingWhenAddedLauncherCommandRequestAndCheckIt(getEnumByKey(Task.class, plannerAllToday.getTask()), systemId);
waitingWhenAddedLauncherCommandRequestAndCheckIt(getEnumByKey(Task.class, plannerAllToday.getTask()), plannerAllToday, systemId);
}
@ -86,7 +86,7 @@ class TaskManagerTest extends AbstractServiceTest {
//ACT
addToPlannerQueue(TaskManager.Process.add, plannerAllToday);
//ASSERT
waitingWhenAddedLauncherCommandRequestAndCheckIt(getEnumByKey(Task.class, plannerAllToday.getTask()), systemId);
waitingWhenAddedLauncherCommandRequestAndCheckIt(getEnumByKey(Task.class, plannerAllToday.getTask()), plannerAllToday, systemId);
}
protected PlannerAllToday getPlannerAllToday(String task) {
@ -98,8 +98,8 @@ class TaskManagerTest extends AbstractServiceTest {
return plannerAllToday;
}
public void waitingWhenAddedLauncherCommandRequestAndCheckIt(Task toTaskQueue, Long userId) {
BaseRequest<Object> predictableBaseRequest = launcherSender.makeCmdRequest(toTaskQueue, userId);
public void waitingWhenAddedLauncherCommandRequestAndCheckIt(Task toTaskQueue, PlannerAllToday plannerAllToday, Long userId) {
BaseRequest<Object> predictableBaseRequest = launcherSender.makeCmdRequest(plannerAllToday, userId);
ArgumentCaptor<ProducerRecord> producerRecord = getCaptor(mockProducer);
//waiting for kafka producer send message (finale event)