First implementation for plugin synchronization
This commit is contained in:
@@ -9,12 +9,14 @@ import mc.core.events.runner.EventExecutorService;
|
|||||||
import mc.core.events.runner.ResourceRunnable;
|
import mc.core.events.runner.ResourceRunnable;
|
||||||
|
|
||||||
import java.lang.reflect.InvocationTargetException;
|
import java.lang.reflect.InvocationTargetException;
|
||||||
|
import java.util.ArrayList;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
|
import java.util.concurrent.locks.Lock;
|
||||||
|
|
||||||
@RequiredArgsConstructor
|
@RequiredArgsConstructor
|
||||||
@Getter
|
@Getter
|
||||||
@Slf4j
|
@Slf4j
|
||||||
public class EventPipeline {
|
public class EventPipelineTask {
|
||||||
private final List<RegisteredEventHandler> handlers;
|
private final List<RegisteredEventHandler> handlers;
|
||||||
private final FullAsyncEventLoop manager;
|
private final FullAsyncEventLoop manager;
|
||||||
private final Event event;
|
private final Event event;
|
||||||
@@ -39,6 +41,11 @@ public class EventPipeline {
|
|||||||
|
|
||||||
RegisteredEventHandler handler = handlers.get(currentIndex);
|
RegisteredEventHandler handler = handlers.get(currentIndex);
|
||||||
if (!event.isCanceled() || !handler.isIgnoreCancelled()) {
|
if (!event.isCanceled() || !handler.isIgnoreCancelled()) {
|
||||||
|
List<Lock> locks = new ArrayList<>();
|
||||||
|
|
||||||
|
if (handler.isPluginSynchronize())
|
||||||
|
locks.add(manager.getResourceManager().getPluginLock(handler.getPlugin()));
|
||||||
|
|
||||||
service.addTask(new ResourceRunnable() {
|
service.addTask(new ResourceRunnable() {
|
||||||
@Override
|
@Override
|
||||||
public void run() {
|
public void run() {
|
||||||
@@ -54,6 +61,11 @@ public class EventPipeline {
|
|||||||
currentIndex++;
|
currentIndex++;
|
||||||
next(service);
|
next(service);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public List<Lock> getLocks() {
|
||||||
|
return locks;
|
||||||
|
}
|
||||||
});
|
});
|
||||||
} else {
|
} else {
|
||||||
currentIndex++;
|
currentIndex++;
|
||||||
@@ -17,10 +17,11 @@ import java.util.concurrent.ConcurrentHashMap;
|
|||||||
public class FullAsyncEventLoop {
|
public class FullAsyncEventLoop {
|
||||||
Map<Class<? extends Event>, List<RegisteredEventHandler>> handlers = new HashMap<>();
|
Map<Class<? extends Event>, List<RegisteredEventHandler>> handlers = new HashMap<>();
|
||||||
// Item leaves this queue only when EventPipeline is fully executed
|
// Item leaves this queue only when EventPipeline is fully executed
|
||||||
private Map<EventQueueOwner, Queue<EventPipeline>> eventQueue = new ConcurrentHashMap<>();
|
private Map<EventQueueOwner, Queue<EventPipelineTask>> eventQueue = new ConcurrentHashMap<>();
|
||||||
@Autowired
|
@Autowired
|
||||||
@Setter
|
@Setter
|
||||||
private EventExecutorService eventExecutorService;
|
private EventExecutorService eventExecutorService;
|
||||||
|
private SharedResourceManager resourceManager = new SharedResourceManager();
|
||||||
|
|
||||||
public void addEventHandler(Plugin plugin, Object object) {
|
public void addEventHandler(Plugin plugin, Object object) {
|
||||||
Map<Method, EventHandler> candidates = getEventHandlerCandidates(object);
|
Map<Method, EventHandler> candidates = getEventHandlerCandidates(object);
|
||||||
@@ -73,8 +74,8 @@ public class FullAsyncEventLoop {
|
|||||||
if (handlers == null)
|
if (handlers == null)
|
||||||
return;
|
return;
|
||||||
|
|
||||||
Queue<EventPipeline> queue = eventQueue.computeIfAbsent(owner, s -> new ArrayDeque<>());
|
Queue<EventPipelineTask> queue = eventQueue.computeIfAbsent(owner, s -> new ArrayDeque<>());
|
||||||
queue.add(new EventPipeline(handlers, this, event, owner));
|
queue.add(new EventPipelineTask(handlers, this, event, owner));
|
||||||
update(owner);
|
update(owner);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -91,22 +92,24 @@ public class FullAsyncEventLoop {
|
|||||||
log.warn("Unable to update pipeline executor: unable to find queue");
|
log.warn("Unable to update pipeline executor: unable to find queue");
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
Queue<EventPipeline> queue = eventQueue.get(owner);
|
Queue<EventPipelineTask> queue = eventQueue.get(owner);
|
||||||
if (queue.isEmpty()) {
|
if (queue.isEmpty()) {
|
||||||
log.warn("Unable to update pipeline executor: queue is empty");
|
log.warn("Unable to update pipeline executor: queue is empty");
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
if (queue.peek().getState() == EventPipeline.PipelineState.FINISHED) {
|
if (queue.peek().getState() == EventPipelineTask.PipelineState.FINISHED) {
|
||||||
queue.poll();
|
queue.poll();
|
||||||
}
|
}
|
||||||
|
|
||||||
EventPipeline pipeline;
|
EventPipelineTask pipeline;
|
||||||
if ((pipeline = queue.peek()) != null
|
if ((pipeline = queue.peek()) != null
|
||||||
&& pipeline.getState() == EventPipeline.PipelineState.IDLE) {
|
&& pipeline.getState() == EventPipelineTask.PipelineState.IDLE) {
|
||||||
pipeline.next(eventExecutorService);
|
pipeline.next(eventExecutorService);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public SharedResourceManager getResourceManager() {
|
||||||
|
return resourceManager;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,17 @@
|
|||||||
|
package mc.core.events;
|
||||||
|
|
||||||
|
import mc.core.events.api.Plugin;
|
||||||
|
|
||||||
|
import java.util.Map;
|
||||||
|
import java.util.concurrent.ConcurrentHashMap;
|
||||||
|
import java.util.concurrent.locks.Lock;
|
||||||
|
import java.util.concurrent.locks.ReentrantLock;
|
||||||
|
|
||||||
|
public class SharedResourceManager {
|
||||||
|
private Map<Plugin, Lock> pluginLocks = new ConcurrentHashMap<>();
|
||||||
|
|
||||||
|
public Lock getPluginLock(Plugin plugin) {
|
||||||
|
return pluginLocks.computeIfAbsent(plugin, s -> new ReentrantLock());
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user