Scheduling update

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 org.apache.kafka.clients.consumer.Consumer;
-import com.hazelcast.map.listener.EntryAddedListener;
+import org.apache.kafka.clients.producer.Producer;
-import com.hazelcast.map.listener.EntryRemovedListener;
+import org.apache.kafka.clients.producer.ProducerRecord;
-import com.hazelcast.map.listener.EntryUpdatedListener;
+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.platform.messaging.domain.Consts.LAUNCHER_NEW;
-import static ru.spcex.clearing.imdg.IMDGDistributedNames.Map_PlannerAllToday;
 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.
  * <p>
  */
-@Service("userReportTask")
+@Service
-@Lazy
+public class TaskManager extends QueueConsumer implements InitializingBean {
-public class TaskManager implements EntryAddedListener<Long, PlannerAllToday>,
+    private final Logger log = LoggerFactory.getLogger(getClass());
-        EntryUpdatedListener<Long, PlannerAllToday>, EntryRemovedListener<Long, PlannerAllToday>,
+
-        InitializingBean {
-    private static final Logger log = LoggerFactory.getLogger(TaskManager.class);
     private final TaskScheduler taskScheduler;
     private final ImdgProvider imdgProvider;
-    private Imdg<Launcher> launcherMap;
     private Imdg<PlannerAllToday> plannerAllTodayMap;
     private ConcurrentHashMap<LocalTime, ScheduledFuture> scheduledJobs;
-    private LauncherSender launcherSender;
+    private Producer<String, Object> kafkaProducer;
 
     @Autowired
-    TaskManager(TaskScheduler taskScheduler,
+    public TaskManager(TaskScheduler taskScheduler,
-                ImdgProvider imdgProvider,
+                       Consumer<String, Object> kafkaQueue,
-                LauncherSender launcherSender) {
+                       Producer<String, Object> kafkaProducer,
-        this.taskScheduler = taskScheduler;
+                       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() {
+    public void afterPropertiesSet() throws Exception {
-        this.launcherMap = imdgProvider.getImdg(Map_Launcher, Launcher.class);
+        plannerAllTodayMap = imdgProvider.getImdg(IMDGDistributedNames.Map_PlannerAllToday, PlannerAllToday.class);
-        this.plannerAllTodayMap = imdgProvider.getImdg(Map_PlannerAllToday, PlannerAllToday.class);
-        if (plannerAllTodayMap instanceof ImdgHazelcast<PlannerAllToday> plannerAllTodayImdgHazelcast) {
-           plannerAllTodayImdgHazelcast.getMap().addEntryListener(this, true);
-        }
         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
+    protected void updateScheduler() {
-    public void entryAdded(EntryEvent<Long, PlannerAllToday> event) {
+        Collection<PlannerAllToday> schedulerAllTodays = plannerAllTodayMap.getAllValues();
-        PlannerAllToday task = event.getValue();
+        Collection<PlannerAllToday> 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<PlannerNewRequest> 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<PlannerUpdateRequest> 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<CommonDeleteRequest> userRequest) {
+        RequestInfoUpdate requestInfoUpdate = new RequestInfoUpdate();
+        return requestInfoUpdate;
+
+    }
+
+    protected RequestInfoUpdate newPlannerTemplate(BaseRequest<PlannerTemplateNewRequest> userRequest) {
+        RequestInfoUpdate requestInfoUpdate = new RequestInfoUpdate();
+        return requestInfoUpdate;
+
+    }
+
+    protected RequestInfoUpdate updatePlannerTemplate(BaseRequest<PlannerTemplateUpdateRequest> userRequest) {
+        RequestInfoUpdate requestInfoUpdate = new RequestInfoUpdate();
+        return requestInfoUpdate;
+
+    }
+
+    protected RequestInfoUpdate deletePlannerTemplate(BaseRequest<CommonDeleteRequest> userRequest) {
+        RequestInfoUpdate requestInfoUpdate = new RequestInfoUpdate();
+        return requestInfoUpdate;
+
+    }
+
+    protected RequestInfoUpdate newClearingCalendar(BaseRequest<ClearingCalendarNewRequest> userRequest) {
+        RequestInfoUpdate requestInfoUpdate = new RequestInfoUpdate();
+        return requestInfoUpdate;
+
+    }
+
+    protected RequestInfoUpdate updateClearingCalendar(BaseRequest<ClearingCalendarUpdateRequest> userRequest) {
+        RequestInfoUpdate requestInfoUpdate = new RequestInfoUpdate();
+        return requestInfoUpdate;
+
+    }
+
+    protected RequestInfoUpdate deleteClearingCalendar(BaseRequest<CommonDeleteRequest> 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
+    private void updateEntry(PlannerAllToday plannerAllToday) {
-    public void entryUpdated(EntryEvent<Long, PlannerAllToday> event) {
+        PlannerAllToday task = plannerAllToday;
-        PlannerAllToday task = event.getValue();
+        PlannerAllToday oldTask;
-        PlannerAllToday oldTask = event.getOldValue();
+        Optional<PlannerAllToday> 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<Long, PlannerAllToday>,
             } 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<Long, PlannerAllToday> 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<Long, PlannerAllToday>,
         }
     }
 
-    // --- Планирование задач ---
+    private void removePlannerAllToday(PlannerAllToday plannerAllToday) {
-    protected void updateScheduler() {
+        removePlannerAllToday(plannerAllToday.getTask(), plannerAllToday.getTaskTime());
-        Collection<PlannerAllToday> schedulerAllTodays = plannerAllTodayMap.getAllValues();
+    }
-        Collection<PlannerAllToday> sortedSchedulers =
+
-                schedulerAllTodays.stream().sorted((o1, o2) -> (Active.equalsByKey(o1.getTaskStatus()) && Cancel.equalsByKey(o2.getTaskStatus())) ? -1 : 0)
+    private void removePlannerAllToday(String task, LocalTime taskTime) {
-                        .collect(Collectors.toCollection(ArrayList::new));
+        for (PlannerAllToday x : plannerAllTodayMap.getAllValues()) {
-        for (PlannerAllToday schedulerAllToday : sortedSchedulers) {
+            if (x.getTaskTime().equals(taskTime) && x.getTask().equalsIgnoreCase(task)) {
-            processTask(schedulerAllToday);
+                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<RecordMetadata> 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<Long, PlannerAllToday>,
             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());
-        }
-    }
 }

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;
     }

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;
     }