Manage event subscriptions in Android.

EventBus.java

import java.lang.Thread.UncaughtExceptionHandler;
import java.lang.ref.WeakReference;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import android.os.Handler;
import android.os.Looper;
import android.util.Log;

/**
 * Utility class to subscribe to events of any class.
 * @author Brian Clymer
 * @version 1.1
 */
public class EventBus {

	private Map<Class<?>, List<Listener>> mSubcribers;
	private Map<Class<?>, Object> mStickies;
	
	private static final int CORE_POOL_SIZE = 1;
	private static final int MAXIMUM_POOL_SIZE = 128;
	private static final int KEEP_ALIVE = 1;
	private static final AtomicInteger threadId = new AtomicInteger();
	private static UncaughtExceptionHandler mExceptionHandler = new UncaughtExceptionHandler() {
		@Override
		public void uncaughtException(Thread thread, Throwable ex) {
			JsonApplication.reportException(ex);
		}
	};
	private static final ThreadFactory mThreadFactory = new ThreadFactory() {
		public Thread newThread(Runnable r) {
			Thread t = new Thread(r);
			t.setName("EventBus-" + threadId.getAndIncrement());
			t.setUncaughtExceptionHandler(mExceptionHandler);
			return t;
		}
	};
	private static final BlockingQueue<Runnable> mPoolWorkQueue = new LinkedBlockingQueue<Runnable>();
	private static final ThreadPoolExecutor mExecutor = new ThreadPoolExecutor(CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE, TimeUnit.SECONDS, mPoolWorkQueue, mThreadFactory);
	private static Handler mHandler;

	
	private EventBus() {
		mSubcribers = new HashMap<Class<?>, List<Listener>>();
		mStickies = new HashMap<Class<?>, Object>();
		mHandler = new Handler(Looper.getMainLooper());
	}

	private static class EventBusHolder {
		public static final EventBus INSTANCE = new EventBus();
	}

	public static EventBus getInstance() {
		return EventBusHolder.INSTANCE;
	}

	
	/**
	 * Subscribe the listeners to all the passed in classes
	 * @param eventListener Class implementing EventListener to subscribe to events.
	 * @param classes Classes to subscribe to events from.
	 */
	public void subscribe(EventListener eventListener, Class<?>... classes) {
		subscribe(eventListener, ThreadMode.POSTING, false, classes);
	}

	
	/**
	 * Subscribe the listeners to all the passed in classes
	 * @param eventListener Class implementing EventListener to subscribe to events.
	 * @param threadMode Which thread to receive the event on.
	 * @param classes Classes to subscribe to events from.
	 */
	public void subscribe(EventListener eventListener, ThreadMode threadMode, Class<?>... classes) {
		subscribe(eventListener, threadMode, false, classes);
	}
	
	/**
	 * Subscribe the listeners to all the passed in classes
	 * @param eventListener Class implementing EventListener to subscribe to events.
	 * @param threadMode Which thread to receive the event on.
	 * @param sticky Whether this class should immediately receive the most recent message of classes.
	 * @param classes Classes to subscribe to events from.
	 */
	public void subscribe(EventListener eventListener, ThreadMode threadMode, boolean sticky, Class<?>... classes) {
		for (Class<?> clazz : classes) {
			if (!mSubcribers.containsKey(clazz)) {
				mSubcribers.put(clazz, new ArrayList<Listener>(1));
			}
			List<Listener> listenerRefs = mSubcribers.get(clazz);
			for (int i = 0; i < listenerRefs.size(); i++) {
				if (listenerRefs.get(i).eventListener.get() != null && listenerRefs.get(i).eventListener.get().equals(eventListener)) {
					Log.w("EventBus", "Attempt to register the same object twice, object not registered second time.");
					return;
				}
			}
			mSubcribers.get(clazz).add(new Listener(threadMode, eventListener));
			if (sticky && mStickies.containsKey(clazz)) {
				post(mStickies.get(clazz), eventListener, threadMode);
			}
		}
	}
	
	/**
	 * Unsubscribe the listener from all passed in classes.
	 * @param eventListener Class implementing EventListener to unsubscribe from events.
	 * @param classes Classes to unsubscribe from.
	 */
	public void unsubscribe(EventListener eventListener, Class<?>... classes) {
		for (Class<?> clazz : classes) {
			if (!mSubcribers.containsKey(clazz)) {
				Log.w("EventBus", "Attempt to unsubscribe without previously subscribing.");
				continue;
			}
			boolean removed = false;
			List<Listener> listenerRefs = mSubcribers.get(clazz);
			for (int i = 0; i < listenerRefs.size(); i++) {
				if (listenerRefs.get(i).eventListener.get() != null && listenerRefs.get(i).eventListener.get().equals(eventListener)) {
					listenerRefs.remove(i);
					removed = true;
					break;
				}
			}
			if (!removed) {
				Log.w("EventBus", "Attempt to unsubscribe without previously subscribing.");
			}
		}
	}
	
	/**
	 * Post an event to all subscribed objects.
	 * @param event The event to post to all subscribers.
	 */
	public void post(Object event) {
		if (!mSubcribers.containsKey(event.getClass())) {
			return;
		}
		mStickies.put(event.getClass(), event);
		List<Listener> listeners = mSubcribers.get(event.getClass());
		List<Listener> toRemove = new ArrayList<Listener>(listeners.size()); 
		for (Listener listener : listeners) {
			EventListener eventListener = listener.eventListener.get();
			if (eventListener == null) {
				Log.w("EventBus", "Some subscription wasn't unregistered before getting deallocated.");
				toRemove.add(listener);
				continue;
			}
			post(event, eventListener, listener.threadMode);
		}
		listeners.removeAll(toRemove);
	}
	
	private void post(Object event, EventListener eventListener, ThreadMode threadMode) {
		switch (threadMode) {
		case POSTING:
			eventListener.newEvent(event);
			break;
		case MAIN:
			// If already on the main thread, post in now.
			// mHandler.post will wait until the main thread is free
			if (Looper.getMainLooper().getThread() == Thread.currentThread()) {
				eventListener.newEvent(event);
			} else {
				mHandler.post(getRunnableForEvent(event, eventListener));
			}
			break;
		case BACKGROUND:
			// not checking if we're on a background thread already, mainly because
			// the user likely called this so that the event is asynchronous, not purely so
			// it's not on the main thread.
			mExecutor.execute(getRunnableForEvent(event, eventListener));
			break;
		}
	}
	
	private Runnable getRunnableForEvent(final Object event, final EventListener eventListener) {
		return new Runnable() {
			@Override
			public void run() {
				eventListener.newEvent(event);
			}
		};
	}
	
	private class Listener {
		ThreadMode threadMode;
		WeakReference<EventListener> eventListener;
		
		public Listener(ThreadMode threadMode, EventListener eventListener) {
			this.threadMode = threadMode;
			this.eventListener = new WeakReference<EventListener>(eventListener);
		}
	}
	
	public enum ThreadMode {
		POSTING,
		MAIN,
		BACKGROUND;
	}
	
	public interface EventListener {
		public void newEvent(Object event);		
	}
}