1
\$\begingroup\$

I have implemented an event bus library for Java - JBus. Anybody please care to review my code? It is an open source project and link is provided above.

Usage

Define events:

public class UserEvent { 
        /* Additional fields if required */ 
    }

Create subscriber:

public class Listener {

        @Subscribe
        private void listen(UserEvent event) {
            /* Your event handling code goes here */
        }

        @Subscribe(async = true)
        public void listenOther(OtherEvent event) {
           /* Your other event handling code goes here */
        }

    }

Register subscriber:

JBus jbus = new JBus();
     jbus.register(new Listener());   
     // or to register with weak reference use jbus.registerWeak(new Listener());

Post event:

jbus.post(new UserEvent());

Some important files are provided here, rest them can be found in the repo itself.

JBus.java

package org.dizitart.jbus;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.util.List;

/**
 * An event bus for java 1.6+. It dispatches event to registered listeners.
 *
 * <p>It is a simple but powerful publish-subscribe event system. It requires object to
 * register themselves with the event bus to receive events. This event bus is safe for
 * concurrent use.</p>
 *
 * <p><b>To receive events</b> from event bus, an object must:
 * <ol>
 *     <li>register itself with the event bus via {@link #register(Object)} or
 *     {@link #registerWeak(Object)} method</li>
 *     <li>have at least one method marked with {@link Subscribe} annotation in its class
 *     hierarchy</li>
 *     <li>subscribed method should accept <b>only one parameter</b> having the type of the event</li>
 * </ol>
 *
 *
 * {@link #register(Object)} will scan the input object for any method which has been marked
 * with {@link Subscribe} annotation and it will keep track of all such methods found. If the
 * subscribing method has more than one parameter, runtime will throw a {@link JBusException}
 * during registration. A subscriber method can have any access modifier.
 * {@link #register(Object)} will scan through full class hierarchy of the input object including
 * any super class and interfaces.
 *
 * <p>Upon successful registration, the runtime will keep track of all subscriber methods
 * found along with a strong reference of the input object for future invocation. To store
 * a weak reference of the input object instead of a strong one, use the {@link #registerWeak(Object)}
 * variant.</p>
 *
 * A developer must {@link #deregister(Object)} the object to stop receiving events. The
 * behavior of {@link #deregister(Object)} is not deterministic in the case of <em>weak
 * registration</em> of the object. As the runtime automatically cleans up any invalid weak
 * references and any subscriber methods associated with it as it goes, so a call to
 * {@link #deregister(Object)} might not do anything if the object has already been
 * garbage collected and the event bus runtime has cleared up its records of subscriber
 * methods already.
 *
 *
 * <p><b>To post an event</b> to the event bus, simply call {@link #post(Object)}
 * passing the event object. Event bus will automatically route the event depending on
 * its type to a handler chain. Handler chain is a collection of registered subscribers
 * of the event. By design, event bus does not support inheritance for the event object.</p>
 *
 * If an event implements {@link HandlerChainAware} then before each invocation, the runtime
 * will check if an interruption has been signalled from the subscriber code via
 * {@link HandlerChain#interrupt()} call. If interrupted, further invocation of the handler
 * chain will be barred until the next {@link #post(Object)} call for the event.
 *
 * <p><b>Subscriber execution</b> mode can be either <em>synchronous</em> or <em>asynchronous</em>
 * depending on the {@link Subscribe} annotation.</p>
 *
 * In case of any error from subscriber code during invocation, the runtime will first search
 * for any {@link ExceptionEvent} handler registered into the system and dispatch the error along with
 * relevant information in {@link ExceptionContext} to the handler if found. If no such error handler
 * is found, runtime will just log the error and move on.
 *
 *
 * @since 1.0
 * @author Anindya Chatterjee.
 * @see Subscribe
 * @see HandlerChain
 * @see HandlerChainAware
 * @see ExceptionEvent
 * @see ExceptionContext
 */
public class JBus {
    private Log logger = LogFactory.getLog(getClass());
    private ListenersRegistry listenersRegistry;
    private EventDispatcher eventDispatcher;
    private String tag = "";

    /**
     * Instantiates a new event bus.
     */
    public JBus() {
        listenersRegistry = new ListenersRegistry();
        eventDispatcher = new EventDispatcher(listenersRegistry);
    }

    /**
     * Registers an event listener to the event bus and keep a strong
     * reference of the listener object for invocation. To listen to events
     * an object must be registered first.
     *
     * @param listener the listener object.
     *
     * @exception  JBusException if the object is already registered or
     * if there is no subscriber method in its class hierarchy.
     * @exception  NullPointerException if the object is null.
     *
     */
    public void register(Object listener) {
        if (listener == null) {
            logger.error("Null object can not be registered.");
            throw new NullPointerException("Null object can not be registered.");
        }
        logger.info("Registering listener " + listener);
        listenersRegistry.register(listener, false);
    }

    /**
     * Registers an event listener to the event bus and keep a weak
     * reference of the listener object for invocation. To listen to events
     * an object must be registered first.
     *
     * @param listener the listener object.
     *
     * @exception  JBusException if the object is already registered or
     * if there is no subscriber method in its class hierarchy.
     * @exception  NullPointerException if the object is null.
     *
     * */
    public void registerWeak(Object listener) {
        if (listener == null) {
            logger.error("Null object can not be registered.");
            throw new NullPointerException("Null object can not be registered.");
        }
        logger.info("Registering listener " + listener);
        listenersRegistry.register(listener, true);
    }

    /**
     * De-registers a listener object that has been registered with the event bus.
     * After de-registration, the object cease to listen to any further event.
     *
     * @param listener the listener object to deregister.
     *
     * @exception  NullPointerException if the object is null.
     *
     * */
    public void deregister(Object listener) {
        if (listener == null) {
            logger.error("Null object can not be de-registered.");
            throw new NullPointerException("Null object can not be de-registered.");
        }
        logger.info("Un-Registering listener " + listener);
        listenersRegistry.deregister(listener);
    }

    /**
     * Posts an event to the event bus.
     *
     * @param event the event to post.
     *
     * @exception NullPointerException if the event is null.
     * @exception JBusException if the invoking subscriber method throws an exception.
     *
     * */
    public void post(Object event) {
        if (event == null) {
            logger.error("Null event posted.");
            throw new NullPointerException("Null event can not be posted.");
        }
        logger.info("Event " + event + " has been posted to the bus " + tag);

        List<ListenerMethod> subscribers = listenersRegistry.getSubscribers(event);
        if (subscribers != null && !subscribers.isEmpty()) {
            if (logger.isDebugEnabled()) {
                logger.debug("Total subscribers found for event " + event + " is = " + subscribers.size());
                logger.debug("Dispatching event " + event);
            }
            DefaultHandlerChain handlerChain = new DefaultHandlerChain(subscribers);
            eventDispatcher.dispatch(event, handlerChain);
        }
    }

    /**
     * Sets a tag to the event bus for identification.
     *
     * @param tag the tag to set.
     * */
    public void setTag(String tag) {
        this.tag = tag;
    }

    /**
     * It registers a JVM shutdown hook for graceful shutdown
     * of event bus.
     *
     * */
    public void addShutdownHook() {
        eventDispatcher.addShutdownHook();
    }
}

EventDispatcher.java

package org.dizitart.jbus;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Internal class for dispatching event to its registered subscribers.
 * It registers itself to a JVM shutdown hook to shutdown gracefully.
 *
 * @since 1.0
 * @author Anindya Chatterjee.
 */
class EventDispatcher {
    private Log logger = LogFactory.getLog(getClass());
    private final ExecutorService executorService = Executors.newCachedThreadPool();
    private ListenersRegistry listenersRegistry;
    private boolean isShutdownInitiated = false;
    private ErrorHandler errorHandler;

    EventDispatcher(ListenersRegistry listenersRegistry) {
        this.listenersRegistry = listenersRegistry;
        errorHandler = new ErrorHandler(listenersRegistry, this);
    }

    /**
     * Dispatches the event to a handler chain.
     *
     * If the submitted event is of type {@link HandlerChainAware}, before each invocation,
     * it will check if an interruption is signalled from the the client code.
     * If signalled then no more invocation will happen for that event.
     *
     * */
    void dispatch(Object event, DefaultHandlerChain handlerChain) {
        // iterate over the subscriber list. If shutdown is initiated already,
        // do not dispatch anything. Otherwise, check the execution mode
        // of the subscriber. If set to async, dispatch it again to the
        // executor service, otherwise invoke the subscriber synchronously.

        HandlerChainAware handlerChainAware;
        if (event instanceof HandlerChainAware) {
            handlerChainAware = (HandlerChainAware) event;
            handlerChainAware.setHandlerChain(handlerChain);
        }

        for (ListenerMethod listenerMethod : handlerChain) {
            if (isShutdownInitiated) {
                logger.debug("Shutdown initiated. No more dispatching.");
                return;
            }

            // check for interruption signal before each invocation. if interrupted,
            // no more invocation will happen from the handler chain.
            if (!handlerChain.interrupt) {
                if (listenerMethod.async) {
                    logger.debug("Executing listener asynchronously " + listenerMethod);
                    dispatchSingleAsync(event, listenerMethod);
                } else {
                    logger.debug("Executing listener " + listenerMethod);
                    dispatchSingle(event, listenerMethod);
                }
            }
        }
    }

    /**
     * Dispatches the subscriber and the event to the executor service for
     * asynchronous execution.
     * */
    private void dispatchSingleAsync(final Object event, final ListenerMethod listenerMethod) {
        executorService.submit(new Runnable() {
            public void run() {
                dispatchSingle(event, listenerMethod);
            }
        });
    }

    /**
     * Executes the subscriber synchronously.
     * */
    private void dispatchSingle(Object event, ListenerMethod listenerMethod) {
        try {
            if (listenerMethod.holdWeakReference) {
                Object listener = listenerMethod.weakListener.get();
                if (listener == null) {
                    // if underlying object is no more, remove it from the runtime and
                    // all of its associations.
                    listenersRegistry.removeWeakListener(listenerMethod.weakListener);
                } else {
                    // invoke synchronously.
                    listenerMethod.method.invoke(listener, event);
                }
            } else {
                listenerMethod.method.invoke(listenerMethod.target, event);
            }
        } catch (Exception e) {
            if (e.getCause() != null) {
                logger.error("Error occurred while invoking " + listenerMethod, e.getCause());
                errorHandler.handle(event, listenerMethod, e.getCause());
            } else {
                logger.error("Error occurred while invoking " + listenerMethod, e);
                errorHandler.handle(event, listenerMethod, e);
            }
        }
    }

    /**
     * Creates a shutdown hook.
     * */
    private Runnable getShutdownHook() {
        return new Runnable() {
            @Override
            public void run() {
                // set a flag to indicate the shutdown has been initiated, so that
                // no more dispatch happens. Then gracefully shutdown the executor.
                isShutdownInitiated = true;
                if (!executorService.isShutdown()) {
                    logger.debug("Shutting down executor, no more event will be dispatched.");
                    executorService.shutdown();
                    logger.debug("Executor has been shutdown gracefully.");
                }
            }
        };
    }

    /**
     * Registers a shutdown hook to the JVM for graceful shutdown of the event bus.
     * */
    void addShutdownHook() {
        // set a shutdown hook to gracefully shutdown executors
        Runtime.getRuntime().addShutdownHook(new Thread(getShutdownHook()));
    }
}

ListenersRegistry.java

package org.dizitart.jbus;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.lang.ref.WeakReference;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

import static org.dizitart.jbus.ReflectionUtil.findSubscribedMethods;

/**
 * A class to hold the records of listeners registered to the event bus runtime.
 *
 * @since 1.0
 * @author Anindya Chatterjee.
 */
class ListenersRegistry {
    private Log logger = LogFactory.getLog(getClass());

    // keep track of event and its registered subscribed methods
    private Map<Class<?>, List<ListenerMethod>> registry =
            new ConcurrentHashMap<Class<?>, List<ListenerMethod>>();
    // cache to keep track of all strong referenced listener object
    private final List<Object> subscriberCache = new CopyOnWriteArrayList<Object>();
    // cache to keep track of all weak referenced listener object
    private final List<WeakReference<Object>> weakSubscriberCache =
            new CopyOnWriteArrayList<WeakReference<Object>>();
    private final Object lock = new Object();

    /**
     * Registers an object in the listener registry. If weak is set,
     * it will create a weak reference of the listener object and register.
     * Otherwise it will store a strong reference of the listener object.
     *
     * */
    void register(Object listener, boolean weak) {
        WeakReference<Object> weakListener = null;

        // synchronize the search in the cache, to check if the listener
        // has already been registered.
        synchronized (lock) {
            if (weak) {
                // if the weak is set, check if a weak reference of the object is
                // kept in the cache or not.
                if (containsWeak(listener)) {
                    logger.error(listener + " has already been registered.");
                    throw new JBusException(listener + " has already been registered.");
                }
                // create a weak reference of the object and add it to the cache
                weakListener = new WeakReference<Object>(listener);
                weakSubscriberCache.add(weakListener);
                logger.debug(listener + " added to the weak subscriber cache.");
            } else {
                if (subscriberCache.contains(listener)) {
                    // if listener is found in the strong referenced cache, throw
                    logger.error(listener + " has already been registered.");
                    throw new JBusException(listener + " has already been registered.");
                }
                // add the object to the strong referenced cache
                subscriberCache.add(listener);
                logger.debug(listener + " added to the subscriber cache.");
            }
        }

        // extract all subscribed methods from the listener and its super class and interfaces.
        // up to this point, we hold a strong reference of the object, beyond this point,
        // if the weak is set, we will not hold any strong reference of the object.
        List<ListenerMethod> subscribedMethods = findSubscribedMethods(listener.getClass());
        if (subscribedMethods == null || subscribedMethods.isEmpty()) {
            logger.error(listener + " does not have any method marked with @Subscribe.");
            throw new JBusException(listener + " does not have any method marked with @Subscribe.");
        }

        for (ListenerMethod listenerMethod : subscribedMethods) {
            if (weak) {
                listenerMethod.weakListener = weakListener;
                listenerMethod.holdWeakReference = true;
            } else {
                listenerMethod.target = listener;
                listenerMethod.holdWeakReference = false;
            }

            Class<?> eventType = listenerMethod.eventType;
            if (registry.containsKey(eventType)) {
                List<ListenerMethod> listenerMethods = registry.get(eventType);

                // check ListenerMethod's equals method
                if (!listenerMethods.contains(listenerMethod)) {
                    listenerMethods.add(listenerMethod);
                    logger.debug(listenerMethod + " has been registered.");
                } else {
                    logger.debug(listenerMethod + " has already been registered.");
                }
            } else {
                List<ListenerMethod> listenerMethods = new CopyOnWriteArrayList<ListenerMethod>();
                listenerMethods.add(listenerMethod);
                registry.put(listenerMethod.eventType, listenerMethods);
                logger.debug(listenerMethod + " has been registered.");
            }
        }
    }

    /**
     * De-registers a listener object.
     * */
    void deregister(Object listener) {
        // synchronize the search. search and remove are put in one place
        // to avoid unnecessary looping.
        synchronized (lock) {
            // we need to check in both the caches, as we don't know how
            // the listener object was registered. If it was a weak reference,
            // there are chances, that underlying object has already been
            // collected by GC. In that case while iterating we will cleanup
            // the cache as we go, if we find such weak reference.
            //
            // But one catch here is that we will never know if we are trying
            // to deregister an object which was never registered before, hence
            // we can not throw such exception.

            for (WeakReference<Object> weakRef : weakSubscriberCache) {
                Object element = weakRef.get();
                if (element == null) {
                    // underlying object is no more. remove it from the cache already.
                    if (weakSubscriberCache.remove(weakRef)) {
                        logger.debug(weakRef + " removed from cache as underlying object does not exist anymore.");
                    }
                } else {
                    if (element.equals(listener)) {
                        // found in weak cache, remove it and break
                        if (weakSubscriberCache.remove(weakRef)) {
                            logger.debug(listener + " removed from the weak subscriber reference cache.");
                        }
                        break;
                    }
                }
            }

            for (Object object : subscriberCache) {
                if (object.equals(listener)) {
                    // found in strong cache, remove it break
                    if (subscriberCache.remove(listener)) {
                        logger.debug(listener + " removed from the subscriber cache.");
                    }
                    break;
                }
            }
        }

        // remove the listener and its subscribed methods from the registry,
        // we are not 100% sure if we are dealing with weak referenced object
        // or not, hence we are passing false.
        removeFromRegistry(listener, false);
    }

    /**
     * Get all registered subscriber information for an event.
     * */
    List<ListenerMethod> getSubscribers(Object event) {
        if (event != null) {
            Class<?> eventType = event.getClass();
            // loop through the registry to get all subscribed method
            if (registry.containsKey(eventType)) {
                return registry.get(eventType);
            }
        }
        return null;
    }

    /**
     * Checks if an object's weak reference is kept in the cache or not.
     * */
    private boolean containsWeak(Object listener) {
        for (WeakReference<Object> weakRef : weakSubscriberCache) {
            Object element = weakRef.get();
            if (element == null) {
                // if the object has already been claimed by the GC,
                // remove it from the cache.
                if (weakSubscriberCache.remove(weakRef)) {
                    logger.debug(weakRef + " removed from cache as underlying object does not exist anymore.");
                }
            } else {
                if (element.equals(listener)) {
                    // if the listener is found equal to any weak referenced object,
                    // return true immediately.
                    return true;
                }
            }
        }
        return false;
    }

    /**
     * Removes a weak referenced listener from the bus runtime.
     * */
    void removeWeakListener(WeakReference<?> weakRef) {
        // first remove it from cache
        if (weakSubscriberCache.remove(weakRef)) {
            logger.debug("Weak reference " + weakRef + " removed from cache.");
        }

        // clean up the registry. Most of the time removeWeakListener is called when
        // underlying object is garbage collected and we want to remove the weak
        // reference from the runtime. So weakRef.get() will always return null
        // in that case.
        //
        // But fortunately, removeFromRegistry method will cleanup the cache while
        // iterating if it finds a weak reference which does not hold any object
        // reference any more.
        removeFromRegistry(weakRef.get(), true);
    }

    /**
     * Removes the listener from the registry. It also cleans up the registry,
     * while iterating, if it finds any weak reference which does not hold any
     * object reference any more, i.e. the underlying object has been garbage collected.
     *
     * @param confirmedWeak if we know listener was registered as a weak reference
     * */
    private void removeFromRegistry(Object listener, boolean confirmedWeak) {
        // iterate the whole registry map
        for (Map.Entry<Class<?>, List<ListenerMethod>> entry : registry.entrySet()) {
            List<ListenerMethod> subscribedMethods = entry.getValue();
            for (ListenerMethod listenerMethod : subscribedMethods) {
                if (confirmedWeak || listenerMethod.holdWeakReference) {
                    // if confirmedWeak is true or listener method holds weak reference,
                    // check if underlying object is still valid.

                    // if not valid clean up. remove the entry from cache and
                    // from the event's subscriber list.
                    Object reference = listenerMethod.weakListener.get();
                    if (reference == null) {
                        // remove from event subscribers list
                        if (subscribedMethods.remove(listenerMethod)) {
                            logger.debug(listenerMethod + " has been un-registered as the " +
                                    "target has been garbage collected.");
                        }
                        // remove that invalid weak reference from cache
                        if (weakSubscriberCache.remove(listenerMethod.weakListener)) {
                            logger.debug(listenerMethod.weakListener +
                                    " removed from cache as underlying object does not exist anymore.");
                        }
                    } else if (reference.equals(listener)) {
                        if (subscribedMethods.remove(listenerMethod)) {
                            logger.debug(listenerMethod + " has been un-registered.");
                        }
                    }
                } else {
                    if (listenerMethod.target.equals(listener)) {
                        if (subscribedMethods.remove(listenerMethod)) {
                            logger.debug(listenerMethod + " has been un-registered.");
                        }
                    }
                }
            }
        }
    }
}

ErrorHandler.java

package org.dizitart.jbus;


import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.util.List;

/**
 * @author Anindya Chatterjee.
 */
class ErrorHandler {
    private Log logger = LogFactory.getLog(getClass());
    private ListenersRegistry listenersRegistry;
    private EventDispatcher eventDispatcher;

    ErrorHandler(ListenersRegistry listenersRegistry, EventDispatcher eventDispatcher) {
        this.listenersRegistry = listenersRegistry;
        this.eventDispatcher = eventDispatcher;
    }

    void handle(Object event, ListenerMethod listenerMethod, Throwable error) {
        // create exception context
        ExceptionContext exceptionContext
                = new ExceptionContext(
                listenerMethod.holdWeakReference ? listenerMethod.weakListener.get() : listenerMethod.target,
                event, listenerMethod.method);
        // create exception event
        ExceptionEvent exceptionEvent
                = new ExceptionEvent(error, exceptionContext);

        // first check if any custom error handler is registered. If found,
        // handle it gracefully, otherwise log it and move on.
        List<ListenerMethod> errorSubscribers = listenersRegistry.getSubscribers(exceptionEvent);
        if (errorSubscribers != null && !errorSubscribers.isEmpty()) {
            logger.debug("Total error handler found for error " + error
                    + " is = " + errorSubscribers.size());
            logger.info("Dispatching error event " + exceptionEvent);
            DefaultHandlerChain errorHandlerChain = new DefaultHandlerChain(errorSubscribers);
            eventDispatcher.dispatch(exceptionEvent, errorHandlerChain);
        } else {
            logger.error("No error handler found for " + error);
        }
    }
}

DefaultHandlerChain.java

package org.dizitart.jbus;

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * An implementation of {@link HandlerChain}. It extends from {@link CopyOnWriteArrayList} to hold
 * the subscriber records.
 *
 * @since 1.0
 * @author Anindya Chatterjee.
 */
class DefaultHandlerChain extends CopyOnWriteArrayList<ListenerMethod> implements HandlerChain {
    volatile boolean interrupt;

    DefaultHandlerChain(List<ListenerMethod> subscribers) {
        super(subscribers);
    }

    @Override
    public void interrupt() {
        interrupt = true;
    }
}

ExceptionContext.java

package org.dizitart.jbus;

import java.lang.reflect.Method;

/**
 * A class to provide relevant information about
 * an exception thrown from subscriber code during event
 * handling.
 *
 * @since 1.0
 * @author Anindya Chatterjee.
 * @see ExceptionEvent
 */
public class ExceptionContext {
    private Object listener;
    private Object event;
    private Method subscribedMethod;

    /**
     * Instantiate a new exception context.
     *
     * @param listener listener object.
     * @param event the event which was being handled
     * @param subscribedMethod the subscribed method which has thrown the exception
     * */
    public ExceptionContext(Object listener, Object event, Method subscribedMethod) {
        this.listener = listener;
        this.event = event;
        this.subscribedMethod = subscribedMethod;
    }

    /**
     * Gets the subscribed method which has thrown the exception.
     *
     * @return the subscribed method.
     * */
    public Method getSubscribedMethod() {
        return subscribedMethod;
    }

    /**
     * Gets the event which was being handled when the exception was thrown.
     *
     * @return the event which was being handled.
     * */
    public Object getEvent() {
        return event;
    }

    /**
     * Gets the listener object.
     *
     * @return the listener object.
     * */
    public Object getListener() {
        return listener;
    }

    @Override
    public String toString() {
        return "[" +
                "method = " +
                subscribedMethod.getName() +
                ", listener = " +
                listener +
                ", event = " +
                event +
                "]";
    }
}

ExceptionEvent.java

package org.dizitart.jbus;

/**
 * Represents the event triggered when an exception is thrown from
 * a subscriber code.
 *
 * @since 1.0
 * @author Anindya Chatterjee.
 * @see ExceptionContext
 */
public class ExceptionEvent implements HandlerChainAware {
    private Throwable exception;
    private ExceptionContext exceptionContext;
    private HandlerChain handlerChain;

    /**
     * Instantiate the exception event.
     *
     * @param exception the exception that was thrown from the subscriber code.
     * @param exceptionContext contextual information about the {@code exception}.
     * */
    public ExceptionEvent(Throwable exception, ExceptionContext exceptionContext) {
        this.exception = exception;
        this.exceptionContext = exceptionContext;
    }

    /**
     * Gets the contextual information about the exception triggering this event.
     *
     * @return the exception context.
     * */
    public ExceptionContext getExceptionContext() {
        return exceptionContext;
    }

    /**
     * Gets the exception triggering this event.
     *
     * @return the exception.
     * */
    public Throwable getException() {
        return exception;
    }

    @Override
    public void setHandlerChain(HandlerChain handlerChain) {
        this.handlerChain = handlerChain;
    }

    @Override
    public HandlerChain getHandlerChain() {
        return handlerChain;
    }

    @Override
    public String toString() {
        return "[" +
                "exception = " +
                exception.toString() +
                ", context = " +
                exceptionContext +
                "]";
    }
}

HandlerChain.java

package org.dizitart.jbus;

/**
 * A mechanism to control the invocation of event handlers. It provides
 * a mechanism to interrupt the execution of handler chain of an event.
 *
 * @since 1.0
 * @author Anindya Chatterjee.
 */
public interface HandlerChain {
    /**
     * Interrupts the execution of handlers of an event. It can be
     * called from the subscriber code to interrupt further execution
     * of other subscriber methods in the handler chain.
     *
     * */
    void interrupt();
}

HandlerChainAware.java

package org.dizitart.jbus;

/**
 * Interface to be implemented by event that wish to be aware of
 * the {@link HandlerChain}.
 *
 * @since 1.0
 * @author Anindya Chatterjee.
 */
public interface HandlerChainAware {
    /**
     * Callback that supplies the instance of the {@link HandlerChain}
     * to the event.
     *
     * @param handlerChain instance of the event handler chain.
     * */
    void setHandlerChain(HandlerChain handlerChain);

    /**
     * Returns the {@link HandlerChain} instance associated with the event.
     *
     * @return {@link HandlerChain} instance.
     * */
    HandlerChain getHandlerChain();
}

ListenerMethod.java

package org.dizitart.jbus;

import org.apache.commons.lang3.builder.HashCodeBuilder;

import java.lang.ref.WeakReference;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;

/**
 * A class to hold all information of a subscriber method.
 *
 * @since 1.0
 * @author Anindya Chatterjee.
 */
class ListenerMethod {
    Object target;
    WeakReference<?> weakListener;

    Method method;
    Class<?> eventType;

    boolean async;
    boolean holdWeakReference;


    ListenerMethod(Method method, Class<?> eventType) {
        this.method = method;
        this.eventType = eventType;
    }

    @Override
    public boolean equals(Object obj) {
        if (this == obj)
            return true;
        if (obj == null)
            return false;
        if (getClass() != obj.getClass())
            return false;
        ListenerMethod other = (ListenerMethod) obj;


        /*
        * Equality condition:
        *
        * 1. Method name should be equal.
        * 2. Method are not overridden. Overridden methods are ignored. It will only consider
        *    the method from the first child class. Overridden methods from super class are
        *    considered equal.
        *    private method can not be overridden, so it is considered as different method.
        *    private methods with same name are allowed in class hierarchy here.
        * 3. Method should have same argument type.
        * 4. Target class is a super class of the other.
        * 6. Target points to the same object.
        * 5. Listener should have same execution mode (sync/async)
        * */

        if (holdWeakReference && other.holdWeakReference) {
            Object t1 = weakListener.get();
            Object t2 = other.weakListener.get();

            if ((t1 == null && t2 == null) || (t1 != null && t2 != null)) {
                if (method.getName().equals(other.method.getName())
                        && method.getModifiers() != Modifier.PRIVATE
                        && other.method.getModifiers() != Modifier.PRIVATE
                        && eventType.equals(other.eventType)
                        && async == other.async) {
                    return t1 == null || (t1.equals(t2) &&
                            (t1.getClass().isAssignableFrom(t2.getClass())
                                    || t2.getClass().isAssignableFrom(t1.getClass())));
                }
            }
        } else {
            return other.method.getName().equals(method.getName())
                    && other.method.getModifiers() != Modifier.PRIVATE
                    && method.getModifiers() != Modifier.PRIVATE
                    && eventType.equals(other.eventType)
                    && async == other.async
                    && target != null
                    && other.target != null
                    && target.equals(other.target)
                    && (target.getClass().isAssignableFrom(other.target.getClass())
                    || other.target.getClass().isAssignableFrom(target.getClass()));
        }
        return false;
    }

    @Override
    public int hashCode() {
        return new HashCodeBuilder()
                .append(holdWeakReference)
                .append(async)
                .append(method)
                .append(holdWeakReference ? weakListener : target)
                .append(eventType)
                .toHashCode();
    }

    @Override
    public String toString() {
        return "[" +
                "method = " +
                method.getName() +
                ", async = " +
                async +
                ", weak = " +
                holdWeakReference +
                ", target = " +
                (holdWeakReference ? weakListener.get() : target) +
                ", event = " +
                eventType.getName() +
                "]";
    }
}

ReflectionUtil.java

package org.dizitart.jbus;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;

/**
 * A reflection utility class to extract information about subscriber methods
 * from a registering listener object.
 *
 * @since 1.0
 * @author Anindya Chatterjee.
 */
class ReflectionUtil {
    private static Log logger = LogFactory.getLog(ReflectionUtil.class);

    /**
     * Finds all subscriber methods in the whole class hierarchy of {@code subscribedClass}.
     *
     * */
    static List<ListenerMethod> findSubscribedMethods(Class<?> subscribedClass) {
        List<ListenerMethod> listenerMethodList = new ArrayList<ListenerMethod>();
        if (subscribedClass != null) {
            Method[] declaredMethods = subscribedClass.getDeclaredMethods();
            for (Method method : declaredMethods) {
                if (method.isAnnotationPresent(Subscribe.class) && !method.isBridge() && !method.isSynthetic()) {
                    if (method.getParameterTypes().length != 1) {
                        logger.error(method.getName() + " has @Subscribe annotation, " +
                                "but it should have exactly 1 parameter.");
                        throw new JBusException(method.getName() + " has @Subscribe annotation, " +
                                "but it should have exactly 1 parameter.");
                    }

                    Class<?> parameterType = method.getParameterTypes()[0];
                    if (parameterType.isArray() || method.isVarArgs()) {
                        logger.error(method.getName() + " has @Subscribe annotation, " +
                                "but its parameter should not be an array or varargs.");
                        throw new JBusException(method.getName() + " has @Subscribe annotation, " +
                                "but its parameter should not be an array or varargs.");
                    }

                    method.setAccessible(true);
                    Subscribe subscribe = method.getAnnotation(Subscribe.class);
                    boolean async = subscribe.async();

                    ListenerMethod listenerMethod = new ListenerMethod(method, method.getParameterTypes()[0]);
                    listenerMethod.async = async;
                    listenerMethodList.add(listenerMethod);
                }
            }

            if (subscribedClass.getSuperclass() != null && !subscribedClass.getSuperclass().equals(Object.class)) {
                if (logger.isDebugEnabled() && !subscribedClass.getSuperclass().equals(Object.class)) {
                    logger.debug("Super class found. searching for listener methods in super class "
                            + subscribedClass.getSuperclass().getName());
                }
                List<ListenerMethod> subscribedMethods = findSubscribedMethods(subscribedClass.getSuperclass());
                listenerMethodList.addAll(subscribedMethods);
            }

            if (subscribedClass.getInterfaces() != null && subscribedClass.getInterfaces().length > 0) {
                if (logger.isDebugEnabled()) {
                    logger.debug("Interface found. searching for listener methods in interfaces.");
                }
                for (Class<?> interfaceClass : subscribedClass.getInterfaces()) {
                    List<ListenerMethod> subscribedMethods = findSubscribedMethods(interfaceClass);
                    listenerMethodList.addAll(subscribedMethods);
                }
            }
        }
        return listenerMethodList;
    }
}

Subscribe.java

package org.dizitart.jbus;

import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

/**
 * Makes a method eligible to respond to an event.
 *
 * <p>It should be applied to a method (with any access modifier) which has
 * only one argument with the type of the event. If it is applied on a method
 * having no parameter or more than one parameters, during registration, event bus
 * runtime will throw a {@link JBusException}.
 * </p>
 *
 * If a method in an interface or an abstract class is marked with this annotation,
 * all the overridden methods will be eligible for subscription.
 *
 * @since 1.0
 * @author Anindya Chatterjee.
 * @see JBus#register(Object)
 * @see JBus#registerWeak(Object)
 */
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.METHOD})
public @interface Subscribe {

    /**
     * Declares whether the subscribed method will be invoked asynchronously.
     *
     * <p>Defaults to {@code false}</p>.
     *
     * @return async flag.
     * */
    boolean async() default false;
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.