diff --git a/clearing-parent/scheduler-service/src/main/java/ru/spcex/clearing/scheduler/service/TaskManager.java b/clearing-parent/scheduler-service/src/main/java/ru/spcex/clearing/scheduler/service/TaskManager.java index bcc7fbc07..9b332b54e 100644 --- a/clearing-parent/scheduler-service/src/main/java/ru/spcex/clearing/scheduler/service/TaskManager.java +++ b/clearing-parent/scheduler-service/src/main/java/ru/spcex/clearing/scheduler/service/TaskManager.java @@ -1,36 +1,43 @@ package ru.spcex.clearing.scheduler.service; -import com.hazelcast.core.EntryEvent; -import com.hazelcast.map.listener.EntryAddedListener; -import com.hazelcast.map.listener.EntryRemovedListener; -import com.hazelcast.map.listener.EntryUpdatedListener; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.context.annotation.Lazy; -import org.springframework.lang.NonNull; import org.springframework.scheduling.TaskScheduler; import org.springframework.stereotype.Service; -import ru.clearing.classes.statics.data.scheduler.Launcher; import ru.clearing.classes.statics.data.scheduler.PlannerAllToday; +import ru.spcex.clearing.imdg.IMDGDistributedNames; +import ru.spcex.clearing.platform.messaging.domain.BaseRequest; +import ru.spcex.clearing.platform.messaging.domain.Consts; +import ru.spcex.clearing.platform.messaging.domain.cud.common.CommonDeleteRequest; +import ru.spcex.clearing.platform.messaging.domain.cud.schedule.*; +import ru.spcex.clearing.platform.messaging.service.QueueConsumer; +import ru.spcex.clearing.platform.messaging.service.RequestInfoUpdate; import ru.spcex.platform.enumeration.Status; import ru.spcex.platform.enumeration.Task; import ru.spcex.platform.imdg.api.Imdg; import ru.spcex.platform.imdg.api.ImdgProvider; -import ru.spcex.platform.imdg.iml.hazelcast.adapter.ImdgHazelcast; -import java.time.*; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.LocalTime; +import java.time.ZoneId; import java.util.ArrayList; import java.util.Collection; -import java.util.Date; import java.util.Objects; +import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; import java.util.concurrent.ScheduledFuture; import java.util.stream.Collectors; -import static ru.spcex.clearing.imdg.IMDGDistributedNames.Map_Launcher; -import static ru.spcex.clearing.imdg.IMDGDistributedNames.Map_PlannerAllToday; +import static ru.spcex.clearing.platform.messaging.domain.Consts.LAUNCHER_NEW; import static ru.spcex.platform.enumeration.Status.*; import static ru.spcex.platform.utils.enumeration.IEnumKey.getEnumByKey; @@ -38,69 +45,187 @@ import static ru.spcex.platform.utils.enumeration.IEnumKey.getEnumByKey; * Планировщик задач, расписание берёт из Hazelcast map. *

*/ -@Service("userReportTask") -@Lazy -public class TaskManager implements EntryAddedListener, - EntryUpdatedListener, EntryRemovedListener, - InitializingBean { - private static final Logger log = LoggerFactory.getLogger(TaskManager.class); +@Service +public class TaskManager extends QueueConsumer implements InitializingBean { + private final Logger log = LoggerFactory.getLogger(getClass()); + private final TaskScheduler taskScheduler; private final ImdgProvider imdgProvider; - private Imdg launcherMap; private Imdg plannerAllTodayMap; private ConcurrentHashMap scheduledJobs; - private LauncherSender launcherSender; + private Producer kafkaProducer; @Autowired - TaskManager(TaskScheduler taskScheduler, - ImdgProvider imdgProvider, - LauncherSender launcherSender) { - this.taskScheduler = taskScheduler; + public TaskManager(TaskScheduler taskScheduler, + Consumer kafkaQueue, + Producer kafkaProducer, + ImdgProvider imdgProvider) { + super(kafkaQueue, kafkaProducer); this.imdgProvider = imdgProvider; - this.launcherSender = launcherSender; + this.taskScheduler = taskScheduler; + this.kafkaProducer = kafkaProducer; } - private static LocalDateTime dateOldTypeConvert(@NonNull Date oldDate) { - return LocalDateTime.ofInstant(oldDate.toInstant(), ZoneId.systemDefault()); - } - - private static LocalDate dateTypeConvert(@NonNull Date oldDate) { - return dateOldTypeConvert(oldDate).toLocalDate(); - } - - private static LocalTime timeTypeConvert(@NonNull Date oldDate) { - return dateOldTypeConvert(oldDate).toLocalTime(); - } @Override - public void afterPropertiesSet() { - this.launcherMap = imdgProvider.getImdg(Map_Launcher, Launcher.class); - this.plannerAllTodayMap = imdgProvider.getImdg(Map_PlannerAllToday, PlannerAllToday.class); - if (plannerAllTodayMap instanceof ImdgHazelcast plannerAllTodayImdgHazelcast) { - plannerAllTodayImdgHazelcast.getMap().addEntryListener(this, true); - } + public void afterPropertiesSet() throws Exception { + plannerAllTodayMap = imdgProvider.getImdg(IMDGDistributedNames.Map_PlannerAllToday, PlannerAllToday.class); scheduledJobs = new ConcurrentHashMap<>(); updateScheduler(); + callback(PlannerNewRequest.class) + .setFunction(this::newPlanner) + .forDestination(Consts.DESTINATION_PLANNER_NEW, callbacks::put);//fixme + callback(PlannerUpdateRequest.class) + .setFunction(this::updatePlanner) + .forDestination(Consts.DESTINATION_PLANNER_UPDATE, callbacks::put);//fixme + callback(CommonDeleteRequest.class) + .setFunction(this::deletePlanner) + .forDestination(Consts.DESTINATION_PLANNER_DELETE, callbacks::put);//fixme + callback(PlannerTemplateNewRequest.class) + .setFunction(this::newPlannerTemplate) + .forDestination(Consts.DESTINATION_PLANNER_TEMPLATE_NEW, callbacks::put); + callback(PlannerTemplateUpdateRequest.class) + .setFunction(this::updatePlannerTemplate) + .forDestination(Consts.DESTINATION_PLANNER_TEMPLATE_UPDATE, callbacks::put); + callback(CommonDeleteRequest.class) + .setFunction(this::deletePlannerTemplate) + .forDestination(Consts.DESTINATION_PLANNER_TEMPLATE_DELETE, callbacks::put); + callback(ClearingCalendarNewRequest.class) + .setFunction(this::newClearingCalendar) + .forDestination(Consts.DESTINATION_CLEARING_CALENDAR_NEW, callbacks::put); + callback(ClearingCalendarUpdateRequest.class) + .setFunction(this::updateClearingCalendar) + .forDestination(Consts.DESTINATION_CLEARING_CALENDAR_UPDATE, callbacks::put); + callback(CommonDeleteRequest.class) + .setFunction(this::deleteClearingCalendar) + .forDestination(Consts.DESTINATION_CLEARING_CALENDAR_DELETE, callbacks::put); + init(); } - // --- Слушатели Hazelcast Map --- - @Override - public void entryAdded(EntryEvent event) { - PlannerAllToday task = event.getValue(); + // --- Планирование задач --- + protected void updateScheduler() { + Collection schedulerAllTodays = plannerAllTodayMap.getAllValues(); + Collection sortedSchedulers = + schedulerAllTodays.stream().sorted((o1, o2) -> (Active.equalsByKey(o1.getTaskStatus()) && Cancel.equalsByKey(o2.getTaskStatus())) ? -1 : 0) + .collect(Collectors.toCollection(ArrayList::new)); + for (PlannerAllToday schedulerAllToday : sortedSchedulers) { + processTask(schedulerAllToday); + } + } + + public RequestInfoUpdate newPlanner(BaseRequest userRequest) { + RequestInfoUpdate requestInfoUpdate = new RequestInfoUpdate(); + PlannerNewRequest request = userRequest.getRequestPayload(); + PlannerAllToday plannerAllToday = new PlannerAllToday(); + plannerAllToday.setTask(request.getTask()); + plannerAllToday.setTaskTime(request.getTaskTime()); + plannerAllToday.setClearingDate(request.getClearingDate()); + plannerAllToday.setMarket(request.getMarket()); + plannerAllToday.setTaskStatus(request.getTaskStatus()); + plannerAllToday.setCompanyId(request.getCompanyId()); + plannerAllToday.setSecurityId(request.getSecurityId()); + plannerAllToday.setParent("PLNR"); + plannerAllToday.setParentId(request.getParentId()); + + processTask(plannerAllToday); + + return requestInfoUpdate; + } + + protected RequestInfoUpdate updatePlanner(BaseRequest userRequest) { + RequestInfoUpdate requestInfoUpdate = new RequestInfoUpdate(); + PlannerUpdateRequest request = userRequest.getRequestPayload(); + PlannerAllToday plannerAllToday = new PlannerAllToday(); + plannerAllToday.setTask(request.getTask()); + plannerAllToday.setTaskTime(request.getTaskTime()); + plannerAllToday.setClearingDate(request.getClearingDate()); + plannerAllToday.setMarket(request.getMarket()); + plannerAllToday.setTaskStatus(request.getTaskStatus()); + plannerAllToday.setCompanyId(request.getCompanyId()); + plannerAllToday.setSecurityId(request.getSecurityId()); + plannerAllToday.setParent("PLNR"); + plannerAllToday.setParentId(request.getParentId()); + return requestInfoUpdate; + + } + + protected RequestInfoUpdate deletePlanner(BaseRequest userRequest) { + RequestInfoUpdate requestInfoUpdate = new RequestInfoUpdate(); + return requestInfoUpdate; + + } + + protected RequestInfoUpdate newPlannerTemplate(BaseRequest userRequest) { + RequestInfoUpdate requestInfoUpdate = new RequestInfoUpdate(); + return requestInfoUpdate; + + } + + protected RequestInfoUpdate updatePlannerTemplate(BaseRequest userRequest) { + RequestInfoUpdate requestInfoUpdate = new RequestInfoUpdate(); + return requestInfoUpdate; + + } + + protected RequestInfoUpdate deletePlannerTemplate(BaseRequest userRequest) { + RequestInfoUpdate requestInfoUpdate = new RequestInfoUpdate(); + return requestInfoUpdate; + + } + + protected RequestInfoUpdate newClearingCalendar(BaseRequest userRequest) { + RequestInfoUpdate requestInfoUpdate = new RequestInfoUpdate(); + return requestInfoUpdate; + + } + + protected RequestInfoUpdate updateClearingCalendar(BaseRequest userRequest) { + RequestInfoUpdate requestInfoUpdate = new RequestInfoUpdate(); + return requestInfoUpdate; + + } + + protected RequestInfoUpdate deleteClearingCalendar(BaseRequest userRequest) { + RequestInfoUpdate requestInfoUpdate = new RequestInfoUpdate(); + + return requestInfoUpdate; + + } + + private void deleteEntry(PlannerAllToday plannerAllToday) { + removeTask(plannerAllToday.getTask(), plannerAllToday.getTaskTime()); + } + + private void newEntry(PlannerAllToday task) { Task taskType = getEnumByKey(Task.class, task.getTask()); if (taskType == null) { log.warn("Task skipped, {} task type not recognized", task.getTask()); return; } + plannerAllTodayMap.insert(task); processTask(task); } - @Override - public void entryUpdated(EntryEvent event) { - PlannerAllToday task = event.getValue(); - PlannerAllToday oldTask = event.getOldValue(); + private void updateEntry(PlannerAllToday plannerAllToday) { + PlannerAllToday task = plannerAllToday; + PlannerAllToday oldTask; + Optional optionalOldTask = plannerAllTodayMap.getAllValues().stream().filter((x) -> { + if (!x.getTaskTime().equals(task.getTaskTime())) { + return false; + } + if (!x.getTask().equalsIgnoreCase(task.getTask())) { + return false; + } + return true; + }).findFirst(); + + if (!optionalOldTask.isPresent()) { + throw new IllegalStateException("Could not find plannerAllToday"); + } + oldTask = optionalOldTask.get(); + task.setId(oldTask.getId()); LocalTime oldTime = oldTask.getTaskTime(); //если таск относится к другому обработчику, пропускаем @@ -114,32 +239,17 @@ public class TaskManager implements EntryAddedListener, } else { log.debug("task with type {}, time {} execution cancelled, adding altered task...", task.getTask(), oldTime.toString()); } + plannerAllTodayMap.update(task); processTask(task); } else if (Cancel.equalsByKey(oldTask.getTaskStatus())) { //&& TaskStatuses.CANCEL.equalsById(task.getTaskStatusId()) restorePreviouslyRemovedTask(oldTime, oldTask); + plannerAllTodayMap.update(task); processTask(task); } else if (Blocked.equalsByKey(oldTask.getTaskStatus())) { + plannerAllTodayMap.update(task); processTask(task); } - } - @Override - public void entryRemoved(EntryEvent event) { - PlannerAllToday taskToRemove = event.getOldValue(); - if (taskToRemove == null) { - log.debug("no task in removed event"); - return; - } - LocalTime removedTaskTime = taskToRemove.getTaskTime(); - - if (Active.equalsByKey(taskToRemove.getTaskStatus())) { - if (removeTask(taskToRemove.getTask(), removedTaskTime)) - log.debug("task successfully canceled"); - } else if (Cancel.equalsByKey(taskToRemove.getTaskStatus())) { - restorePreviouslyRemovedTask(removedTaskTime, taskToRemove); - } else if (Blocked.equalsByKey(taskToRemove.getTaskStatus())) { - log.debug("BLOCKED task removed; do nothing"); - } } private void restorePreviouslyRemovedTask(LocalTime oldTime, PlannerAllToday oldTask) { @@ -154,23 +264,54 @@ public class TaskManager implements EntryAddedListener, } } - // --- Планирование задач --- - protected void updateScheduler() { - Collection schedulerAllTodays = plannerAllTodayMap.getAllValues(); - Collection sortedSchedulers = - schedulerAllTodays.stream().sorted((o1, o2) -> (Active.equalsByKey(o1.getTaskStatus()) && Cancel.equalsByKey(o2.getTaskStatus())) ? -1 : 0) - .collect(Collectors.toCollection(ArrayList::new)); - for (PlannerAllToday schedulerAllToday : sortedSchedulers) { - processTask(schedulerAllToday); + private void removePlannerAllToday(PlannerAllToday plannerAllToday) { + removePlannerAllToday(plannerAllToday.getTask(), plannerAllToday.getTaskTime()); + } + + private void removePlannerAllToday(String task, LocalTime taskTime) { + for (PlannerAllToday x : plannerAllTodayMap.getAllValues()) { + if (x.getTaskTime().equals(taskTime) && x.getTask().equalsIgnoreCase(task)) { + plannerAllTodayMap.delete(x); + break; + } } } - // --- Работа с задачами --- + private boolean removeTask(String task, LocalTime taskTime) { + removePlannerAllToday(task, taskTime); + ScheduledFuture future = scheduledJobs.remove(taskTime); + if (future == null) { + log.debug("can't cancel task type {}, time {}, not found", task, taskTime); + return false; + } + return future.cancel(false); + } + + protected void doJob(PlannerAllToday task) { + Task taskE = getEnumByKey(Task.class, task.getTask()); + if (taskE != null) { + LauncherCommandRequest request = new LauncherCommandRequest(); + request.setTaskName(taskE.name()); + request.setUserId(task.getId()); + log.debug("Send command {} to {}", request, LAUNCHER_NEW); + Future send = kafkaProducer.send(new ProducerRecord<>(LAUNCHER_NEW, request)); + try { + send.get(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException("Command " + request + " not send", e); + } catch (ExecutionException e) { + throw new RuntimeException("Command " + request + " not send", e); + } + } + } + + // --- Работа с тасками --- private void processTask(PlannerAllToday task) { LocalTime taskTime = task.getTaskTime(); Task taskType = getEnumByKey(Task.class, task.getTask()); Status taskStatus = getEnumByKey(Status.class, task.getTaskStatus()); - if (taskStatus == null) throw new IllegalStateException("task status from core can't be null"); + if (taskStatus == null) throw new IllegalStateException("task status can't be null"); if (taskTime.isBefore(LocalTime.now())) { log.debug("Task skipped - taskTime {} is before now, taskType {} ok", taskTime, task.getTask()); return; @@ -210,33 +351,4 @@ public class TaskManager implements EntryAddedListener, log.warn("cancelling task " + (success ? "success" : "fail")); } } - - private boolean removeTask(String task, LocalTime taskTime) { - ScheduledFuture future = scheduledJobs.remove(taskTime); - if (future == null) { - log.debug("can't cancel task type {}, time {}, not found", task, taskTime); - return false; - } - return future.cancel(false); - } - - // --- Реализация выполнения задач --- - - protected void doJob(PlannerAllToday task) { - Task taskE = getEnumByKey(Task.class, task.getTask()); - if (taskE != null) { - log.debug("LauncherCommandRequest received"); - // 1. cохранить команду - Instant created = Instant.now(); - Launcher launcher = new Launcher(); - launcher.setTask(task.getTask()); - launcher.setSenderId(task.getParentId()); - launcher.setCreated(created); - launcher.setUpdated(created); - launcherMap.insert(launcher); - // 2. отправить сообщение - launcherSender.sendCommandToQueue(taskE, task.getParentId()); - log.debug("successfully processed, new id {}", launcher.getId()); - } - } } diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/schedule/PlannerNewRequest.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/schedule/PlannerNewRequest.java index 104469fbc..15ec7096d 100644 --- a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/schedule/PlannerNewRequest.java +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/schedule/PlannerNewRequest.java @@ -38,6 +38,17 @@ public class PlannerNewRequest { @JsonProperty public Long securityId; + @JsonProperty(required = false) + public Long parentId; + + public Long getParentId() { + return parentId; + } + + public void setParentId(Long parentId) { + this.parentId = parentId; + } + public String getTask() { return task; } diff --git a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/schedule/PlannerUpdateRequest.java b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/schedule/PlannerUpdateRequest.java index 3c186a7a5..a52365efa 100644 --- a/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/schedule/PlannerUpdateRequest.java +++ b/platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/cud/schedule/PlannerUpdateRequest.java @@ -34,6 +34,18 @@ public class PlannerUpdateRequest { @JsonProperty public Long securityId; + @JsonProperty(required = false) + public Long parentId; + + public Long getParentId() { + return parentId; + } + + public void setParentId(Long parentId) { + this.parentId = parentId; + } + + public Long getId() { return id; }