aalehin
parent
9830cc8805
commit
b225e0d3d9
3 changed files with 124 additions and 101 deletions
|
|
@ -13,7 +13,7 @@ public enum TaskStatuses {
|
|||
this.name = name;
|
||||
}
|
||||
|
||||
static TaskStatuses getEnumById(String name) {
|
||||
public static TaskStatuses getEnumByName(String name) {
|
||||
for (TaskStatuses e : TaskStatuses.values()) {
|
||||
if (Objects.equals(name, e.name()))
|
||||
return e;
|
||||
|
|
@ -21,6 +21,10 @@ public enum TaskStatuses {
|
|||
return null;
|
||||
}
|
||||
|
||||
public Boolean equalsByName(String name) {
|
||||
return this.name.equalsIgnoreCase(name);
|
||||
}
|
||||
|
||||
public String getName() {
|
||||
return this.name;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,16 +1,30 @@
|
|||
package ru.spcex.clearing.scheduler.enums;
|
||||
|
||||
public enum Tasks implements IEnumWithLongValue {
|
||||
SOME_STATUS(1L);//TODO CLEARIFY
|
||||
import java.util.Objects;
|
||||
|
||||
private final Long id;
|
||||
public enum Tasks {
|
||||
SOME_STATUS("STATUS");
|
||||
|
||||
Tasks(Long id) {
|
||||
this.id = id;
|
||||
private String name;
|
||||
|
||||
Tasks(String name) {
|
||||
this.name = name;
|
||||
}
|
||||
|
||||
public Long getId() {
|
||||
return this.id;
|
||||
public static Tasks getEnumByName(String name) {
|
||||
for (Tasks e : Tasks.values()) {
|
||||
if (Objects.equals(name, e.name()))
|
||||
return e;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
public Boolean equalsByName(String name) {
|
||||
return this.name.equalsIgnoreCase(name);
|
||||
}
|
||||
|
||||
public String getName() {
|
||||
return this.name;
|
||||
}
|
||||
|
||||
public String elementName() {
|
||||
|
|
|
|||
|
|
@ -11,17 +11,23 @@ import org.slf4j.LoggerFactory;
|
|||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.scheduling.TaskScheduler;
|
||||
import ru.clearing.classes.statics.data.scheduler.PlannerAllToday;
|
||||
import ru.spcex.clearing.scheduler.enums.TaskStatuses;
|
||||
import ru.spcex.clearing.scheduler.enums.Tasks;
|
||||
|
||||
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.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ScheduledFuture;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import static ru.spcex.clearing.imdg.IMDGDistributedNames.Map_PlannerAllToday;
|
||||
import static ru.spcex.clearing.scheduler.enums.TaskStatuses.*;
|
||||
|
||||
/**
|
||||
* Планировщик задач, расписание берёт из Hazelcast map.
|
||||
|
|
@ -67,12 +73,11 @@ public abstract class TaskManager implements EntryAddedListener<Long, PlannerAll
|
|||
public void entryAdded(EntryEvent<Long, PlannerAllToday> event) {
|
||||
PlannerAllToday task = event.getValue();
|
||||
|
||||
|
||||
//todo define how to get task status Tasks taskType = getEnumById(Tasks.class, task.getParentId());
|
||||
// if (taskType == null) {
|
||||
// log.warn("Task skipped, {} task type not recognized", task.getTaskId());
|
||||
// return;
|
||||
// }
|
||||
Tasks taskType = Tasks.getEnumByName(task.getTask());
|
||||
if (taskType == null) {
|
||||
log.warn("Task skipped, {} task type not recognized", task.getTask());
|
||||
return;
|
||||
}
|
||||
processTask(task);
|
||||
}
|
||||
|
||||
|
|
@ -82,27 +87,27 @@ public abstract class TaskManager implements EntryAddedListener<Long, PlannerAll
|
|||
PlannerAllToday oldTask = event.getOldValue();
|
||||
|
||||
LocalTime oldTime = oldTask.getTaskTime();
|
||||
// todo define TaskStatuses oldStatus
|
||||
// TaskStatuses oldStatus
|
||||
//если таск относится к другому обработчику, пропускаем
|
||||
// if (!getTaskId().equalsById(task.getTaskId()) && !getTaskId().equalsById(oldTask.getTaskId())) {
|
||||
// return;
|
||||
// }
|
||||
// if (!Objects.equals(task.getTaskId(), oldTask.getTaskId())) {
|
||||
// throw new IllegalStateException("changed taskId for SchedulerAllToday in core");
|
||||
// }
|
||||
// if (ACTIVE.equalsById(oldTask.getTaskStatusId())) { //&& ACTIVE.equalsById(task.getTaskStatusId())
|
||||
// if (!removeTask(oldTime)) {
|
||||
// log.debug("cannot cancel task with type {}, time {}", getTaskId().name(), oldTime.toString());
|
||||
// } else {
|
||||
// log.debug("task with type {}, time {} execution cancelled, adding altered task...", getTaskId().name(), oldTime.toString());
|
||||
// }
|
||||
// processTask(task);
|
||||
// } else if (CANCEL.equalsById(oldTask.getTaskStatusId())) { //&& TaskStatuses.CANCEL.equalsById(task.getTaskStatusId())
|
||||
// restorePreviouslyRemovedTask(oldTime, oldTask);
|
||||
// processTask(task);
|
||||
// } else if (BLOCKED.equalsById(oldTask.getTaskStatusId())) {
|
||||
// processTask(task);
|
||||
// }
|
||||
if (!getTask().equalsByName(task.getTask()) && !getTask().equalsByName(oldTask.getTask())) {
|
||||
return;
|
||||
}
|
||||
if (!Objects.equals(task.getTask(), oldTask.getTask())) {
|
||||
throw new IllegalStateException("changed taskId for SchedulerAllToday in core");
|
||||
}
|
||||
if (ACTIVE.equalsByName(oldTask.getTaskStatus())) { //&& ACTIVE.equalsById(task.getTaskStatusId())
|
||||
if (!removeTask(oldTime)) {
|
||||
log.debug("cannot cancel task with type {}, time {}", getTask().name(), oldTime.toString());
|
||||
} else {
|
||||
log.debug("task with type {}, time {} execution cancelled, adding altered task...", getTask().name(), oldTime.toString());
|
||||
}
|
||||
processTask(task);
|
||||
} else if (CANCEL.equalsByName(oldTask.getTaskStatus())) { //&& TaskStatuses.CANCEL.equalsById(task.getTaskStatusId())
|
||||
restorePreviouslyRemovedTask(oldTime, oldTask);
|
||||
processTask(task);
|
||||
} else if (BLOCKED.equalsByName(oldTask.getTaskStatus())) {
|
||||
processTask(task);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -112,24 +117,24 @@ public abstract class TaskManager implements EntryAddedListener<Long, PlannerAll
|
|||
log.debug("no task in removed event");
|
||||
return;
|
||||
}
|
||||
//LocalTime removedTaskTime = timeTypeConvert(taskToRemove.getTaskTime());
|
||||
//
|
||||
//if (ACTIVE.equalsById(taskToRemove.getTaskStatusId())) {
|
||||
// if (removeTask(removedTaskTime))
|
||||
// log.debug("task successfully canceled");
|
||||
//} else if (CANCEL.equalsById(taskToRemove.getTaskStatusId())) {
|
||||
// restorePreviouslyRemovedTask(removedTaskTime, taskToRemove);
|
||||
//} else if (BLOCKED.equalsById(taskToRemove.getTaskStatusId())) {
|
||||
// log.debug("BLOCKED task removed; do nothing");
|
||||
//}
|
||||
LocalTime removedTaskTime = taskToRemove.getTaskTime();
|
||||
|
||||
if (ACTIVE.equalsByName(taskToRemove.getTaskStatus())) {
|
||||
if (removeTask(removedTaskTime))
|
||||
log.debug("task successfully canceled");
|
||||
} else if (CANCEL.equalsByName(taskToRemove.getTaskStatus())) {
|
||||
restorePreviouslyRemovedTask(removedTaskTime, taskToRemove);
|
||||
} else if (BLOCKED.equalsByName(taskToRemove.getTaskStatus())) {
|
||||
log.debug("BLOCKED task removed; do nothing");
|
||||
}
|
||||
}
|
||||
|
||||
private void restorePreviouslyRemovedTask(LocalTime oldTime, PlannerAllToday oldTask) {
|
||||
ScheduledFuture cancelledFuture = scheduledJobs.get(oldTime);
|
||||
if (cancelledFuture != null && cancelledFuture.isCancelled()) {
|
||||
PlannerAllToday schedulerAllToday = new PlannerAllToday();
|
||||
// schedulerAllToday.setTaskId(getTaskId().getId());
|
||||
// schedulerAllToday.setTaskStatusId(ACTIVE.getId());
|
||||
schedulerAllToday.setTask(getTask().getName());
|
||||
schedulerAllToday.setTaskStatus(ACTIVE.getName());
|
||||
schedulerAllToday.setTaskTime(oldTask.getTaskTime());
|
||||
log.debug("CANCEL task updated/removed; restoring previously cancelled task");
|
||||
processTask(schedulerAllToday);
|
||||
|
|
@ -138,69 +143,69 @@ public abstract class TaskManager implements EntryAddedListener<Long, PlannerAll
|
|||
|
||||
// --- Планирование задач ---
|
||||
protected void updateScheduler() {
|
||||
//Collection<PlannerAllToday> schedulerAllTodays = plannerAllTodayMapStore.values();
|
||||
//Collection<PlannerAllToday> sortedSchedulers =
|
||||
// schedulerAllTodays.stream().sorted((o1, o2) -> (ACTIVE.equalsById(o1.getTaskStatusId()) && CANCEL.equalsById(o2.getTaskStatusId())) ? -1 : 0)
|
||||
// .collect(Collectors.toCollection(ArrayList::new));
|
||||
//for (PlannerAllToday schedulerAllToday : sortedSchedulers) {
|
||||
// processTask(schedulerAllToday);
|
||||
//}
|
||||
Collection<PlannerAllToday> schedulerAllTodays = plannerAllTodayMapStore.values();
|
||||
Collection<PlannerAllToday> sortedSchedulers =
|
||||
schedulerAllTodays.stream().sorted((o1, o2) -> (ACTIVE.equalsByName(o1.getTaskStatus()) && CANCEL.equalsByName(o2.getTaskStatus())) ? -1 : 0)
|
||||
.collect(Collectors.toCollection(ArrayList::new));
|
||||
for (PlannerAllToday schedulerAllToday : sortedSchedulers) {
|
||||
processTask(schedulerAllToday);
|
||||
}
|
||||
}
|
||||
|
||||
// --- Работа с задачами ---
|
||||
private void processTask(PlannerAllToday task) {
|
||||
//LocalTime taskTime = task.getTaskTime();
|
||||
//Tasks taskType = getEnumById(Tasks.class, task.getTaskId());
|
||||
//TaskStatuses taskStatus = getEnumById(TaskStatuses.class, task.getTaskStatusId());
|
||||
//if (taskStatus == null) throw new IllegalStateException("task status from core can't be null");
|
||||
//if (!getTaskId().equals(taskType)) {
|
||||
// log.debug("Task skipped - taskTime {}, is not of acceptable type {}", taskTime.toString(), taskType != null ? taskType.name() : "");
|
||||
// return;
|
||||
//}
|
||||
//if (taskTime.isBefore(LocalTime.now())) {
|
||||
// log.debug("Task skipped - taskTime {} is before now, taskType {} ok", taskTime, getTaskId().toString());
|
||||
// return;
|
||||
//}
|
||||
//if (taskStatus.equals(BLOCKED)) {
|
||||
// log.debug("Task skipped - timeTime {} with status {}", taskTime.toString(), BLOCKED.toString());
|
||||
// return;
|
||||
//}
|
||||
//{
|
||||
// ScheduledFuture future = scheduledJobs.get(taskTime);
|
||||
// if (future != null) {
|
||||
// if (TaskStatuses.CANCEL.equals(taskStatus)) {
|
||||
// //случай когда пришел cancel, пытаемся отменить зарегистрированный ранее таск
|
||||
// future.cancel(false);
|
||||
// log.debug("cancelling task time {}", taskTime);
|
||||
// return;
|
||||
// } else if (TaskStatuses.ACTIVE.equals(taskStatus)) {
|
||||
// //случай когда пришел активный таск, и уже был на это время неотмененный
|
||||
// if (!future.isCancelled()) {
|
||||
// log.debug("such task time {} has already been registered", taskTime);
|
||||
// return;
|
||||
// }
|
||||
// }
|
||||
// }
|
||||
//}
|
||||
//if (TaskStatuses.CANCEL.equals(taskStatus)) {
|
||||
// log.debug("task type {} time {} with status CANCEL - no tasks to cancel found", taskType.name(), taskTime);
|
||||
// return;
|
||||
//}
|
||||
////пришел активный таск
|
||||
//log.debug("adding task type {}, time {}", taskType.name(), taskTime);
|
||||
//ScheduledFuture future = taskScheduler.schedule(() -> doJob(task), LocalDateTime.of(LocalDate.now(), taskTime).atZone(ZoneId.systemDefault()).toInstant());
|
||||
//ScheduledFuture oldFuture = scheduledJobs.put(taskTime, future);
|
||||
//if (oldFuture != null && !oldFuture.isCancelled()) { //for synchronization, never
|
||||
// log.warn("tasks were added simultaneously, cancel former one");
|
||||
// boolean success = oldFuture.cancel(false);
|
||||
// log.warn("cancelling task " + (success ? "success" : "fail"));
|
||||
//}
|
||||
LocalTime taskTime = task.getTaskTime();
|
||||
Tasks taskType = Tasks.getEnumByName(task.getTask());
|
||||
TaskStatuses taskStatus = TaskStatuses.getEnumByName(task.getTaskStatus());
|
||||
if (taskStatus == null) throw new IllegalStateException("task status from core can't be null");
|
||||
if (!getTask().equals(taskType)) {
|
||||
log.debug("Task skipped - taskTime {}, is not of acceptable type {}", taskTime.toString(), taskType != null ? taskType.name() : "");
|
||||
return;
|
||||
}
|
||||
if (taskTime.isBefore(LocalTime.now())) {
|
||||
log.debug("Task skipped - taskTime {} is before now, taskType {} ok", taskTime, getTask().toString());
|
||||
return;
|
||||
}
|
||||
if (taskStatus.equals(BLOCKED)) {
|
||||
log.debug("Task skipped - timeTime {} with status {}", taskTime.toString(), BLOCKED.toString());
|
||||
return;
|
||||
}
|
||||
{
|
||||
ScheduledFuture future = scheduledJobs.get(taskTime);
|
||||
if (future != null) {
|
||||
if (TaskStatuses.CANCEL.equals(taskStatus)) {
|
||||
//случай когда пришел cancel, пытаемся отменить зарегистрированный ранее таск
|
||||
future.cancel(false);
|
||||
log.debug("cancelling task time {}", taskTime);
|
||||
return;
|
||||
} else if (TaskStatuses.ACTIVE.equals(taskStatus)) {
|
||||
//случай когда пришел активный таск, и уже был на это время неотмененный
|
||||
if (!future.isCancelled()) {
|
||||
log.debug("such task time {} has already been registered", taskTime);
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if (TaskStatuses.CANCEL.equals(taskStatus)) {
|
||||
log.debug("task type {} time {} with status CANCEL - no tasks to cancel found", taskType.name(), taskTime);
|
||||
return;
|
||||
}
|
||||
//пришел активный таск
|
||||
log.debug("adding task type {}, time {}", taskType.name(), taskTime);
|
||||
ScheduledFuture future = taskScheduler.schedule(() -> doJob(task), LocalDateTime.of(LocalDate.now(), taskTime).atZone(ZoneId.systemDefault()).toInstant());
|
||||
ScheduledFuture oldFuture = scheduledJobs.put(taskTime, future);
|
||||
if (oldFuture != null && !oldFuture.isCancelled()) { //for synchronization, never
|
||||
log.warn("tasks were added simultaneously, cancel former one");
|
||||
boolean success = oldFuture.cancel(false);
|
||||
log.warn("cancelling task " + (success ? "success" : "fail"));
|
||||
}
|
||||
}
|
||||
|
||||
private boolean removeTask(LocalTime taskTime) {
|
||||
ScheduledFuture future = scheduledJobs.remove(taskTime);
|
||||
if (future == null) {
|
||||
log.debug("can't cancel task type {}, time {}, not found", getTaskId().name(), taskTime);
|
||||
log.debug("can't cancel task type {}, time {}, not found", getTask().name(), taskTime);
|
||||
return false;
|
||||
}
|
||||
return future.cancel(false);
|
||||
|
|
@ -213,7 +218,7 @@ public abstract class TaskManager implements EntryAddedListener<Long, PlannerAll
|
|||
*
|
||||
* @return идентификатор для фильтра типов планировщика задачь. Планировать задачи только этого типа.
|
||||
*/
|
||||
protected abstract Tasks getTaskId();
|
||||
protected abstract Tasks getTask();
|
||||
|
||||
protected abstract void doJob(PlannerAllToday taskInfo);
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue