task manager...
This commit is contained in:
aalehin
parent
91433d2710
commit
9830cc8805
4 changed files with 351 additions and 0 deletions
|
|
@ -0,0 +1,81 @@
|
|||
package ru.spcex.clearing.scheduler.enums;
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.util.Objects;
|
||||
|
||||
/**
|
||||
* Удобный интерфейс для enum при использовании TextErrorService.
|
||||
*/
|
||||
public interface IEnumWithLongValue extends Serializable {
|
||||
|
||||
|
||||
/**
|
||||
* Проверяет что среди данного набора Enum, присутствует элемент с данным id
|
||||
* id может быть null
|
||||
*/
|
||||
static <T extends Enum<T> & IEnumWithLongValue> boolean contains(Long id, T... enumSet) {
|
||||
for (T e : enumSet) {
|
||||
if (Objects.equals(id, e.getId()))
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
static <T extends Enum<T> & IEnumWithLongValue> boolean contains(T e, T... enumSet) {
|
||||
for (T enumEl : enumSet) {
|
||||
if (enumEl.equals(e))
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
static <T extends Enum<T> & IEnumWithLongValue> Long[] toLongArray(T... enumSet) {
|
||||
Long[] enumId = new Long[enumSet.length];
|
||||
for (int i = 0; i < enumSet.length; i++)
|
||||
enumId[i] = enumSet[i].getId();
|
||||
return enumId;
|
||||
}
|
||||
|
||||
/**
|
||||
* Проверяет что среди всех Enum данного класса, присутствует элемент с данным id
|
||||
* id может быть null
|
||||
*/
|
||||
static <T extends Enum<T> & IEnumWithLongValue> boolean contains(Class<T> enumClass, Long id) {
|
||||
for (T e : enumClass.getEnumConstants()) {
|
||||
if (Objects.equals(id, e.getId()))
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Возвращает Enum по id, если в заданном классе такой определен
|
||||
* id может быть null
|
||||
*
|
||||
* @return Enum если нашел, иначе <tt>null</tt>
|
||||
*/
|
||||
static <T extends Enum<T> & IEnumWithLongValue> T getEnumById(Class<T> enumClass, Long id) {
|
||||
for (T e : enumClass.getEnumConstants()) {
|
||||
if (Objects.equals(id, e.getId()))
|
||||
return e;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
static Long getIdOrNull(IEnumWithLongValue enumVal) {
|
||||
return enumVal == null ? null : enumVal.getId();
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* errorCode
|
||||
**/
|
||||
Long getId();
|
||||
|
||||
default boolean equalsById(Long id) {
|
||||
return id != null && getId().equals(id);
|
||||
}
|
||||
|
||||
String elementName();
|
||||
}
|
||||
|
|
@ -0,0 +1,31 @@
|
|||
package ru.spcex.clearing.scheduler.enums;
|
||||
|
||||
import java.util.Objects;
|
||||
|
||||
public enum TaskStatuses {
|
||||
ACTIVE("ACTV"),
|
||||
BLOCKED("CNCL"),
|
||||
CANCEL("BLKD");
|
||||
|
||||
private final String name;
|
||||
|
||||
TaskStatuses(String name) {
|
||||
this.name = name;
|
||||
}
|
||||
|
||||
static TaskStatuses getEnumById(String name) {
|
||||
for (TaskStatuses e : TaskStatuses.values()) {
|
||||
if (Objects.equals(name, e.name()))
|
||||
return e;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
public String getName() {
|
||||
return this.name;
|
||||
}
|
||||
|
||||
public String elementName() {
|
||||
return this.name();
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,20 @@
|
|||
package ru.spcex.clearing.scheduler.enums;
|
||||
|
||||
public enum Tasks implements IEnumWithLongValue {
|
||||
SOME_STATUS(1L);//TODO CLEARIFY
|
||||
|
||||
private final Long id;
|
||||
|
||||
Tasks(Long id) {
|
||||
this.id = id;
|
||||
}
|
||||
|
||||
public Long getId() {
|
||||
return this.id;
|
||||
}
|
||||
|
||||
public String elementName() {
|
||||
return this.name();
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,219 @@
|
|||
package ru.spcex.clearing.scheduler.service;
|
||||
|
||||
import com.hazelcast.core.EntryEvent;
|
||||
import com.hazelcast.core.HazelcastInstance;
|
||||
import com.hazelcast.core.IMap;
|
||||
import com.hazelcast.map.listener.EntryAddedListener;
|
||||
import com.hazelcast.map.listener.EntryRemovedListener;
|
||||
import com.hazelcast.map.listener.EntryUpdatedListener;
|
||||
import org.slf4j.Logger;
|
||||
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.Tasks;
|
||||
|
||||
import java.time.LocalDate;
|
||||
import java.time.LocalDateTime;
|
||||
import java.time.LocalTime;
|
||||
import java.time.ZoneId;
|
||||
import java.util.Date;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ScheduledFuture;
|
||||
|
||||
import static ru.spcex.clearing.imdg.IMDGDistributedNames.Map_PlannerAllToday;
|
||||
|
||||
/**
|
||||
* Планировщик задач, расписание берёт из Hazelcast map.
|
||||
* <p>
|
||||
*/
|
||||
public abstract class TaskManager implements EntryAddedListener<Long, PlannerAllToday>,
|
||||
EntryUpdatedListener<Long, PlannerAllToday>, EntryRemovedListener<Long, PlannerAllToday>,
|
||||
InitializingBean {
|
||||
private static final Logger log = LoggerFactory.getLogger(TaskManager.class);
|
||||
|
||||
protected HazelcastInstance hazelcastInstance;
|
||||
protected IMap<Long, PlannerAllToday> plannerAllTodayMapStore;
|
||||
protected TaskScheduler taskScheduler;
|
||||
|
||||
protected ConcurrentHashMap<LocalTime, ScheduledFuture> scheduledJobs;
|
||||
|
||||
protected TaskManager(TaskScheduler taskScheduler, HazelcastInstance hazelcastInstance) {
|
||||
this.taskScheduler = taskScheduler;
|
||||
this.hazelcastInstance = hazelcastInstance;
|
||||
}
|
||||
|
||||
private static LocalDateTime dateOldTypeConvert(Date oldDate) {
|
||||
return LocalDateTime.ofInstant(oldDate.toInstant(), ZoneId.systemDefault());
|
||||
}
|
||||
|
||||
private static LocalDate dateTypeConvert(Date oldDate) {
|
||||
return dateOldTypeConvert(oldDate).toLocalDate();
|
||||
}
|
||||
|
||||
private static LocalTime timeTypeConvert(Date oldDate) {
|
||||
return dateOldTypeConvert(oldDate).toLocalTime();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void afterPropertiesSet() {
|
||||
plannerAllTodayMapStore = hazelcastInstance.getMap(Map_PlannerAllToday);
|
||||
scheduledJobs = new ConcurrentHashMap<>();
|
||||
updateScheduler();
|
||||
}
|
||||
|
||||
// --- Слушатели Hazelcast Map ---
|
||||
@Override
|
||||
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;
|
||||
// }
|
||||
processTask(task);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void entryUpdated(EntryEvent<Long, PlannerAllToday> event) {
|
||||
PlannerAllToday task = event.getValue();
|
||||
PlannerAllToday oldTask = event.getOldValue();
|
||||
|
||||
LocalTime oldTime = oldTask.getTaskTime();
|
||||
// todo define 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);
|
||||
// }
|
||||
}
|
||||
|
||||
@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 = 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");
|
||||
//}
|
||||
}
|
||||
|
||||
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.setTaskTime(oldTask.getTaskTime());
|
||||
log.debug("CANCEL task updated/removed; restoring previously cancelled task");
|
||||
processTask(schedulerAllToday);
|
||||
}
|
||||
}
|
||||
|
||||
// --- Планирование задач ---
|
||||
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);
|
||||
//}
|
||||
}
|
||||
|
||||
// --- Работа с задачами ---
|
||||
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"));
|
||||
//}
|
||||
}
|
||||
|
||||
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);
|
||||
return false;
|
||||
}
|
||||
return future.cancel(false);
|
||||
}
|
||||
|
||||
// --- Реализация выполнения задач ---
|
||||
|
||||
/**
|
||||
* Рабочий тип запланированных задач.
|
||||
*
|
||||
* @return идентификатор для фильтра типов планировщика задачь. Планировать задачи только этого типа.
|
||||
*/
|
||||
protected abstract Tasks getTaskId();
|
||||
|
||||
protected abstract void doJob(PlannerAllToday taskInfo);
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue