001/*
002 * Copyright (C) 2007 The Guava Authors
003 *
004 * Licensed under the Apache License, Version 2.0 (the "License");
005 * you may not use this file except in compliance with the License.
006 * You may obtain a copy of the License at
007 *
008 * http://www.apache.org/licenses/LICENSE-2.0
009 *
010 * Unless required by applicable law or agreed to in writing, software
011 * distributed under the License is distributed on an "AS IS" BASIS,
012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
013 * See the License for the specific language governing permissions and
014 * limitations under the License.
015 */
016
017package com.google.common.eventbus;
018
019import static com.google.common.base.Preconditions.checkNotNull;
020
021import com.google.common.annotations.Beta;
022import com.google.common.annotations.VisibleForTesting;
023import com.google.common.base.Throwables;
024import com.google.common.cache.CacheBuilder;
025import com.google.common.cache.CacheLoader;
026import com.google.common.cache.LoadingCache;
027import com.google.common.collect.HashMultimap;
028import com.google.common.collect.Multimap;
029import com.google.common.collect.SetMultimap;
030import com.google.common.reflect.TypeToken;
031import com.google.common.util.concurrent.UncheckedExecutionException;
032
033import java.lang.reflect.InvocationTargetException;
034import java.util.Collection;
035import java.util.LinkedList;
036import java.util.Map.Entry;
037import java.util.Queue;
038import java.util.Set;
039import java.util.concurrent.locks.ReadWriteLock;
040import java.util.concurrent.locks.ReentrantReadWriteLock;
041import java.util.logging.Level;
042import java.util.logging.Logger;
043
044/**
045 * Dispatches events to listeners, and provides ways for listeners to register
046 * themselves.
047 *
048 * <p>The EventBus allows publish-subscribe-style communication between
049 * components without requiring the components to explicitly register with one
050 * another (and thus be aware of each other).  It is designed exclusively to
051 * replace traditional Java in-process event distribution using explicit
052 * registration. It is <em>not</em> a general-purpose publish-subscribe system,
053 * nor is it intended for interprocess communication.
054 *
055 * <h2>Receiving Events</h2>
056 * <p>To receive events, an object should:<ol>
057 * <li>Expose a public method, known as the <i>event handler</i>, which accepts
058 *     a single argument of the type of event desired;</li>
059 * <li>Mark it with a {@link Subscribe} annotation;</li>
060 * <li>Pass itself to an EventBus instance's {@link #register(Object)} method.
061 *     </li>
062 * </ol>
063 *
064 * <h2>Posting Events</h2>
065 * <p>To post an event, simply provide the event object to the
066 * {@link #post(Object)} method.  The EventBus instance will determine the type
067 * of event and route it to all registered listeners.
068 *
069 * <p>Events are routed based on their type &mdash; an event will be delivered
070 * to any handler for any type to which the event is <em>assignable.</em>  This
071 * includes implemented interfaces, all superclasses, and all interfaces
072 * implemented by superclasses.
073 *
074 * <p>When {@code post} is called, all registered handlers for an event are run
075 * in sequence, so handlers should be reasonably quick.  If an event may trigger
076 * an extended process (such as a database load), spawn a thread or queue it for
077 * later.  (For a convenient way to do this, use an {@link AsyncEventBus}.)
078 *
079 * <h2>Handler Methods</h2>
080 * <p>Event handler methods must accept only one argument: the event.
081 *
082 * <p>Handlers should not, in general, throw.  If they do, the EventBus will
083 * catch and log the exception.  This is rarely the right solution for error
084 * handling and should not be relied upon; it is intended solely to help find
085 * problems during development.
086 *
087 * <p>The EventBus guarantees that it will not call a handler method from
088 * multiple threads simultaneously, unless the method explicitly allows it by
089 * bearing the {@link AllowConcurrentEvents} annotation.  If this annotation is
090 * not present, handler methods need not worry about being reentrant, unless
091 * also called from outside the EventBus.
092 *
093 * <h2>Dead Events</h2>
094 * <p>If an event is posted, but no registered handlers can accept it, it is
095 * considered "dead."  To give the system a second chance to handle dead events,
096 * they are wrapped in an instance of {@link DeadEvent} and reposted.
097 *
098 * <p>If a handler for a supertype of all events (such as Object) is registered,
099 * no event will ever be considered dead, and no DeadEvents will be generated.
100 * Accordingly, while DeadEvent extends {@link Object}, a handler registered to
101 * receive any Object will never receive a DeadEvent.
102 *
103 * <p>This class is safe for concurrent use.
104 * 
105 * <p>See the Guava User Guide article on <a href=
106 * "http://code.google.com/p/guava-libraries/wiki/EventBusExplained">
107 * {@code EventBus}</a>.
108 *
109 * @author Cliff Biffle
110 * @since 10.0
111 */
112@Beta
113public class EventBus {
114
115  /**
116   * A thread-safe cache for flattenHierarchy(). The Class class is immutable. This cache is shared
117   * across all EventBus instances, which greatly improves performance if multiple such instances
118   * are created and objects of the same class are posted on all of them.
119   */
120  private static final LoadingCache<Class<?>, Set<Class<?>>> flattenHierarchyCache =
121      CacheBuilder.newBuilder()
122          .weakKeys()
123          .build(new CacheLoader<Class<?>, Set<Class<?>>>() {
124            @SuppressWarnings({"unchecked", "rawtypes"}) // safe cast
125            @Override
126            public Set<Class<?>> load(Class<?> concreteClass) {
127              return (Set) TypeToken.of(concreteClass).getTypes().rawTypes();
128            }
129          });
130
131  /**
132   * All registered event handlers, indexed by event type.
133   *
134   * <p>This SetMultimap is NOT safe for concurrent use; all access should be
135   * made after acquiring a read or write lock via {@link #handlersByTypeLock}.
136   */
137  private final SetMultimap<Class<?>, EventHandler> handlersByType =
138      HashMultimap.create();
139  private final ReadWriteLock handlersByTypeLock = new ReentrantReadWriteLock();
140
141  /**
142   * Logger for event dispatch failures.  Named by the fully-qualified name of
143   * this class, followed by the identifier provided at construction.
144   */
145  private final Logger logger;
146
147  /**
148   * Strategy for finding handler methods in registered objects.  Currently,
149   * only the {@link AnnotatedHandlerFinder} is supported, but this is
150   * encapsulated for future expansion.
151   */
152  private final HandlerFindingStrategy finder = new AnnotatedHandlerFinder();
153
154  /** queues of events for the current thread to dispatch */
155  private final ThreadLocal<Queue<EventWithHandler>> eventsToDispatch =
156      new ThreadLocal<Queue<EventWithHandler>>() {
157    @Override protected Queue<EventWithHandler> initialValue() {
158      return new LinkedList<EventWithHandler>();
159    }
160  };
161
162  /** true if the current thread is currently dispatching an event */
163  private final ThreadLocal<Boolean> isDispatching =
164      new ThreadLocal<Boolean>() {
165    @Override protected Boolean initialValue() {
166      return false;
167    }
168  };
169
170  /**
171   * Creates a new EventBus named "default".
172   */
173  public EventBus() {
174    this("default");
175  }
176
177  /**
178   * Creates a new EventBus with the given {@code identifier}.
179   *
180   * @param identifier  a brief name for this bus, for logging purposes.  Should
181   *                    be a valid Java identifier.
182   */
183  public EventBus(String identifier) {
184    logger = Logger.getLogger(EventBus.class.getName() + "." + checkNotNull(identifier));
185  }
186
187  /**
188   * Registers all handler methods on {@code object} to receive events.
189   * Handler methods are selected and classified using this EventBus's
190   * {@link HandlerFindingStrategy}; the default strategy is the
191   * {@link AnnotatedHandlerFinder}.
192   *
193   * @param object  object whose handler methods should be registered.
194   */
195  public void register(Object object) {
196    Multimap<Class<?>, EventHandler> methodsInListener =
197        finder.findAllHandlers(object);
198    handlersByTypeLock.writeLock().lock();
199    try {
200      handlersByType.putAll(methodsInListener);
201    } finally {
202      handlersByTypeLock.writeLock().unlock();
203    }
204  }
205
206  /**
207   * Unregisters all handler methods on a registered {@code object}.
208   *
209   * @param object  object whose handler methods should be unregistered.
210   * @throws IllegalArgumentException if the object was not previously registered.
211   */
212  public void unregister(Object object) {
213    Multimap<Class<?>, EventHandler> methodsInListener = finder.findAllHandlers(object);
214    for (Entry<Class<?>, Collection<EventHandler>> entry : methodsInListener.asMap().entrySet()) {
215      Class<?> eventType = entry.getKey();
216      Collection<EventHandler> eventMethodsInListener = entry.getValue();
217
218      handlersByTypeLock.writeLock().lock();
219      try {
220        Set<EventHandler> currentHandlers = handlersByType.get(eventType);
221        if (!currentHandlers.containsAll(eventMethodsInListener)) {
222          throw new IllegalArgumentException(
223              "missing event handler for an annotated method. Is " + object + " registered?");
224        }
225        currentHandlers.removeAll(eventMethodsInListener);
226      } finally {
227        handlersByTypeLock.writeLock().unlock();
228      }
229    }
230  }
231
232  /**
233   * Posts an event to all registered handlers.  This method will return
234   * successfully after the event has been posted to all handlers, and
235   * regardless of any exceptions thrown by handlers.
236   *
237   * <p>If no handlers have been subscribed for {@code event}'s class, and
238   * {@code event} is not already a {@link DeadEvent}, it will be wrapped in a
239   * DeadEvent and reposted.
240   *
241   * @param event  event to post.
242   */
243  public void post(Object event) {
244    Set<Class<?>> dispatchTypes = flattenHierarchy(event.getClass());
245
246    boolean dispatched = false;
247    for (Class<?> eventType : dispatchTypes) {
248      handlersByTypeLock.readLock().lock();
249      try {
250        Set<EventHandler> wrappers = handlersByType.get(eventType);
251
252        if (!wrappers.isEmpty()) {
253          dispatched = true;
254          for (EventHandler wrapper : wrappers) {
255            enqueueEvent(event, wrapper);
256          }
257        }
258      } finally {
259        handlersByTypeLock.readLock().unlock();
260      }
261    }
262
263    if (!dispatched && !(event instanceof DeadEvent)) {
264      post(new DeadEvent(this, event));
265    }
266
267    dispatchQueuedEvents();
268  }
269
270  /**
271   * Queue the {@code event} for dispatch during
272   * {@link #dispatchQueuedEvents()}. Events are queued in-order of occurrence
273   * so they can be dispatched in the same order.
274   */
275  void enqueueEvent(Object event, EventHandler handler) {
276    eventsToDispatch.get().offer(new EventWithHandler(event, handler));
277  }
278
279  /**
280   * Drain the queue of events to be dispatched. As the queue is being drained,
281   * new events may be posted to the end of the queue.
282   */
283  void dispatchQueuedEvents() {
284    // don't dispatch if we're already dispatching, that would allow reentrancy
285    // and out-of-order events. Instead, leave the events to be dispatched
286    // after the in-progress dispatch is complete.
287    if (isDispatching.get()) {
288      return;
289    }
290
291    isDispatching.set(true);
292    try {
293      Queue<EventWithHandler> events = eventsToDispatch.get();
294      EventWithHandler eventWithHandler;
295      while ((eventWithHandler = events.poll()) != null) {
296        dispatch(eventWithHandler.event, eventWithHandler.handler);
297      }
298    } finally {
299      isDispatching.remove();
300      eventsToDispatch.remove();
301    }
302  }
303
304  /**
305   * Dispatches {@code event} to the handler in {@code wrapper}.  This method
306   * is an appropriate override point for subclasses that wish to make
307   * event delivery asynchronous.
308   *
309   * @param event  event to dispatch.
310   * @param wrapper  wrapper that will call the handler.
311   */
312  void dispatch(Object event, EventHandler wrapper) {
313    try {
314      wrapper.handleEvent(event);
315    } catch (InvocationTargetException e) {
316      logger.log(Level.SEVERE,
317          "Could not dispatch event: " + event + " to handler " + wrapper, e);
318    }
319  }
320
321  /**
322   * Flattens a class's type hierarchy into a set of Class objects.  The set
323   * will include all superclasses (transitively), and all interfaces
324   * implemented by these superclasses.
325   *
326   * @param concreteClass  class whose type hierarchy will be retrieved.
327   * @return {@code clazz}'s complete type hierarchy, flattened and uniqued.
328   */
329  @VisibleForTesting
330  Set<Class<?>> flattenHierarchy(Class<?> concreteClass) {
331    try {
332      return flattenHierarchyCache.getUnchecked(concreteClass);
333    } catch (UncheckedExecutionException e) {
334      throw Throwables.propagate(e.getCause());
335    }
336  }
337
338  /** simple struct representing an event and it's handler */
339  static class EventWithHandler {
340    final Object event;
341    final EventHandler handler;
342    public EventWithHandler(Object event, EventHandler handler) {
343      this.event = checkNotNull(event);
344      this.handler = checkNotNull(handler);
345    }
346  }
347}