1
0
mirror of https://github.com/Rogiel/l2jserver2 synced 2025-12-06 07:32:46 +00:00

WorldEventFuture implementation

Signed-off-by: Rogiel <rogiel@rogiel.com>
This commit is contained in:
2011-05-21 00:35:29 -03:00
parent 33b05c33cf
commit 949bcf64e0
7 changed files with 271 additions and 13 deletions

View File

@@ -102,6 +102,11 @@ public class CachedWorldIDService extends AbstractService implements
loaded = true;
}
@Override
public void unload() {
cache.removeAll();
}
/**
* Load the pre-existing ids
*

View File

@@ -35,6 +35,11 @@ public interface WorldIDService extends Service {
*/
void load();
/**
* Unload all loaded {@link ObjectID}
*/
void unload();
/**
* Tries to resolve an ID based on its raw value
*

View File

@@ -82,6 +82,7 @@ public class WorldServiceImpl extends AbstractService implements WorldService {
protected void doStart() throws ServiceStartException {
objects.clear();
idService.load();
dispatcher.start();
}
@Override
@@ -172,5 +173,7 @@ public class WorldServiceImpl extends AbstractService implements WorldService {
@Override
protected void doStop() throws ServiceStopException {
objects.clear();
idService.unload();
dispatcher.stop();
}
}

View File

@@ -31,10 +31,14 @@ public interface WorldEventDispatcher {
* need to invoke listeners immediately. Dispatching <b>can</b> occur
* concurrently.
*
* @param <E>
* the event type
* @param event
* the event
* @return the future. The future can be used to be notified once the event
* has been dispatched to all listeners.
*/
void dispatch(WorldEvent event);
<E extends WorldEvent> WorldEventFuture<E> dispatch(E event);
/**
* Adds a new global <tt>listener</tt>

View File

@@ -21,10 +21,14 @@ import java.util.Queue;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.util.concurrent.AbstractFuture;
import com.l2jserver.model.id.ObjectID;
import com.l2jserver.model.world.WorldObject;
import com.l2jserver.util.factory.CollectionFactory;
@@ -35,28 +39,51 @@ import com.l2jserver.util.factory.CollectionFactory;
* @author <a href="http://www.rogiel.com">Rogiel</a>
*/
public class WorldEventDispatcherImpl implements WorldEventDispatcher {
/**
* The logger
*/
private static final Logger log = LoggerFactory
.getLogger(WorldEventDispatcherImpl.class);
private final Timer timer = new Timer();
/**
* The execution thread
*/
private Timer timer;
/**
* The list of all global listeners
*/
private Set<WorldListener> globalListeners = CollectionFactory.newSet();
/**
* The {@link Map} containing all listeners for every object
*/
private Map<ObjectID<?>, Set<WorldListener>> listeners = CollectionFactory
.newMap();
// private Queue<ListenerIDPair> listeners = CollectionFactory
// .newConcurrentQueue(ListenerIDPair.class);
private Queue<WorldEvent> events = CollectionFactory.newConcurrentQueue();
/**
* The events pending dispatch
*/
private Queue<EventContainer> events = CollectionFactory
.newConcurrentQueue();
public WorldEventDispatcherImpl() {
public void start() {
timer = new Timer();
timer.scheduleAtFixedRate(new TimerTask() {
@Override
public void run() {
final WorldEvent event = events.poll();
final EventContainer event = events.poll();
if (event == null)
return;
try {
doDispatch(event);
// set state
event.future.running = true;
event.future.complete = false;
// dispatch
if (doDispatch(event))
// the set will update state
event.future.set(event.event);
} catch (Throwable t) {
event.future.setException(t);
log.warn("Exception in WorldEventDispatcher thread", t);
}
}
@@ -64,21 +91,32 @@ public class WorldEventDispatcherImpl implements WorldEventDispatcher {
}
@Override
public void dispatch(WorldEvent event) {
public <E extends WorldEvent> WorldEventFuture<E> dispatch(E event) {
log.debug("Queing dispatch for event {}", event);
events.add(event);
final WorldEventFutureImpl<E> future = new WorldEventFutureImpl<E>();
events.add(new EventContainer(event, future));
return future;
}
public void doDispatch(WorldEvent event) {
/**
* Do the dispatching
*
* @param event
* the event
* @return true if dispatch was not canceled
*/
public boolean doDispatch(EventContainer event) {
log.debug("Dispatching event {}", event);
final ObjectID<?>[] objects = event.getDispatchableObjects();
final ObjectID<?>[] objects = event.event.getDispatchableObjects();
for (ObjectID<?> obj : objects) {
if (obj == null)
continue;
final Set<WorldListener> listeners = getListeners(obj);
for (final WorldListener listener : listeners) {
if (event.future.isCancelled())
return false;
try {
if (!listener.dispatch(event))
if (!listener.dispatch(event.event))
// remove listener if return value is false
listeners.remove(listener);
} catch (ClassCastException e) {
@@ -89,6 +127,7 @@ public class WorldEventDispatcherImpl implements WorldEventDispatcher {
}
}
}
return true;
}
@Override
@@ -134,6 +173,14 @@ public class WorldEventDispatcherImpl implements WorldEventDispatcher {
listeners.remove(id);
}
/**
* Get the {@link Set} of listeners for an given object. Creates a new one
* if does not exists.
*
* @param id
* the object id
* @return the {@link Set}. Never null.
*/
private Set<WorldListener> getListeners(ObjectID<?> id) {
Set<WorldListener> set = listeners.get(id);
if (set == null) {
@@ -142,4 +189,117 @@ public class WorldEventDispatcherImpl implements WorldEventDispatcher {
}
return set;
}
public void stop() {
timer.cancel();
timer = null;
}
/**
* {@link WorldEventFuture} implementation
*
* @param <E>
* the event type
* @author <a href="http://www.rogiel.com">Rogiel</a>
*/
private static class WorldEventFutureImpl<E extends WorldEvent> extends
AbstractFuture<E> implements WorldEventFuture<E> {
private boolean running = false;
private boolean complete = false;
@Override
@SuppressWarnings("unchecked")
protected boolean set(WorldEvent value) {
running = false;
complete = true;
return super.set((E) value);
}
@Override
protected boolean setException(Throwable throwable) {
return super.setException(throwable);
}
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
if (!mayInterruptIfRunning && running)
return false;
if (complete)
return false;
return cancel();
}
@Override
public void await() throws InterruptedException {
try {
super.get();
} catch (ExecutionException e) {
}
}
@Override
public void await(long timeout, TimeUnit unit)
throws InterruptedException, TimeoutException {
try {
super.get(timeout, unit);
} catch (ExecutionException e) {
}
}
@Override
public boolean awaitUninterruptibly() {
try {
super.get();
return true;
} catch (InterruptedException e) {
return false;
} catch (ExecutionException e) {
return false;
}
}
@Override
public boolean awaitUninterruptibly(long timeout, TimeUnit unit) {
try {
super.get(timeout, unit);
return true;
} catch (InterruptedException e) {
return false;
} catch (ExecutionException e) {
return false;
} catch (TimeoutException e) {
return false;
}
}
}
/**
* Simple container that contains an event and a future
*
* @author <a href="http://www.rogiel.com">Rogiel</a>
*/
private static class EventContainer {
/**
* The event
*/
private final WorldEvent event;
/**
* The future
*/
private final WorldEventFutureImpl<? extends WorldEvent> future;
/**
* Creates a new instance
*
* @param event
* the event
* @param future
* the future
*/
public EventContainer(WorldEvent event,
WorldEventFutureImpl<? extends WorldEvent> future) {
this.event = event;
this.future = future;
}
}
}

View File

@@ -0,0 +1,74 @@
/*
* This file is part of l2jserver <l2jserver.com>.
*
* l2jserver is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* l2jserver is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with l2jserver. If not, see <http://www.gnu.org/licenses/>.
*/
package com.l2jserver.service.game.world.event;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
/**
* This is an {@link Future} for {@link WorldEvent}. This {@link Future} can be
* used to receive notifications once an event has been dispatched to all
* listeners.
*
* @author <a href="http://www.rogiel.com">Rogiel</a>
*/
public interface WorldEventFuture<E extends WorldEvent> extends Future<E> {
/**
* Waits until the event is dispatched to all listeners
*
* @throws InterruptedException
* if the thread has been interrupted while waiting
*/
void await() throws InterruptedException;
/**
* Waits until the event is dispatched to all listeners
*
* @param timeout
* the timeout
* @param unit
* the timeout unit
*
* @throws InterruptedException
* if the thread has been interrupted while waiting
* @throws TimeoutException
* if timeout was exceeded
*/
void await(long timeout, TimeUnit unit) throws InterruptedException,
TimeoutException;
/**
* Waits until the event is dispatched to all listeners
*
* @return true if execution ended with no error, false otherwise
*/
boolean awaitUninterruptibly();
/**
* Waits until the event is dispatched to all listeners
*
* @param timeout
* the timeout
* @param unit
* the timeout unit
*
* @return true if execution ended with no error, false otherwise. Please
* note that false will be returned if the timeout has expired too!
*/
boolean awaitUninterruptibly(long timeout, TimeUnit unit);
}