Package org.gcube.common.events

Interface Hub

All Known Implementing Classes:
DefaultHub
public interface Hub
Mediates between producers and observers of events.

Producers fire events through the hub ((cf. fire(Object, String...))), which routes them to all the observers that have previously subscribed to receive them (subscribe(Object))).

Producers, Observers, and Events

Producers, observers, and events are arbitrary objects. Observers are objects with one or more methods marked with the Observes annotation and taking events as their only parameter. Observes methods can have any name and access modifier, and may throw any kind of exception. Any object value may serve as an event. The following example illustrates: 

 
 class Observer {
 
   {@literal @}Observes
   void someMethod(int event) {...}
 
 }
 
  Hub hub = ....;
 
  hub.subscribe(new Observer());
 
  hub.fire(10);
In general, events and Observes methods match events when the type of events is a subtype of the type of the single parameter of the observer methods. Normal java subtyping rule apply, with the following exceptions:
  • observers cannot use primitive types and should use wrapper types instead.
  • parametric types are not directly supported, as Java does not make available type parameters at runtime. Possible solutions are discussed below.
Qualifiers The type matching algorithm can be refined by qualifying events and event types with one ore more labels that observers declare in Observes annotations and produce specify when firing events. An example of using qualifiers is the following: 
 
 class Observer {
 
      {@literal @}Observes({"currency","dollars"}) 
      void onDollarPayment(Integer amount) {...}
      
      {@literal @}Observes({"currency","euro"}) 
      void onEuroPayment(Integer amount) {...}
      
      {@literal @}Observes({"currency"}) 
      void onAnyPayment(Integer amount) {...} 
 }
 
  Hub hub = ....;
 
  hub.subscribe(new Observer());
 
  hub.fire(10, "currency", "dollars");
Here the methods onDollarPayment() and onAnyPayment() receive the event, while the method onEuroPayment() does not. In general, Observes methods are notified if they specify a subset of the qualifiers associated with events, including no qualifiers at all.

Event Grouping

Observers that perform costly operations may wish to process rapid bursts of events at once. Observes methods may then specify the minimal delay in milliseconds between two successive notifications {cf. @link Observes#every()}. All the events produced within this delay are grouped and delivered in a collection when the delay expires and in the order in which they were produced. Observers process the collections as they see fit (e.g. consume the most recent, merge all the events, aggregate data in the events, etc). For example: 

 
 
   {@literal @}Observes(value="change",every=1000) 
   void onPayments(List<Integer> payments) {...}
Any Collection type can be used for the delivery of the events.

Critical, Safe, and Resilient Consumers

Firing events blocks producers until all matching Observes methods that are marked Observes.Kind.critical have been executed (cf. Observes.kind()). Critical Observes methods execute sequentially in an unpredictable order, and any exception they raise is reported to producers. Producers do not block instead for the execution of Observes methods that are marked Observes.Kind.safe or Observes.Kind.resilient. Safe and resilient Observes methods execute in parallel and any exception they raise is logged. Finally, the difference between Observes.Kind.safe and Observes.Kind.resilient handlers is that the former execute if and only if there are no critical failures, while the latter execute in all cases. Parametric Observers and Events

Parametric observers and events can be still be used in either one of two ways:

  • Qualifiers can be used to differentiate different instantiations of a parametric type (qualifiers are discussed below). Like with any other use of qualifiers, convenience is traded off for compile-time safety.
  • Events can be wrapped as instances of the Event class, which is provided to capture type information which is otherwise lost at runtime due to type erasure. This approach is more verbose but also safer.
Consider the following example of event wrapping: 
 
 class Observer {
 
   {@literal @}Observes
   void someMethod(MyType<Integer> event) {...}
 
 }
 
  Hub hub = ....;
 
  hub.subscribe(new Observer());
 
  MyType<Integer> event = ...
  
  hub.fire(new Event<MyType<Integer>>(event){});
where new Event<MyType<Integer>>(event){} instantiates an anonymous Event subclass. The idiom is hardly palatable, but it does circumvent the limitation of type erasure in Java. Currently, the follow limitations apply:
  • type variables cannot be used in both events and observers' parameters;
  • wildcards are supported, but only with upper bounds (? extends ...). Lower bounds are not currently supported.
Author:
Fabio Simeoni, Luca Frosini (ISTI-CNR)
See Also:

  • Method Summary

    Modifier and Type
    Method
    Description
    void
    fire(Object event, String... qualifiers)
    Fires an event to all observers which have subscribed for them, blocking the client until all the critical observers have executed.
    void
    stop()
    Signals that this hub is no longer needed and can release its resources.
    void
    subscribe(Object observer)
    Subscribes an observer for notification of events of a given type.
    boolean
    unsubscribe(Object observer)
    Unsubscribes an observer.
    void
    waitFor(Class<?> eventType)
    Blocks the caller until an event of a given type occurs.
    void
    waitFor(Class<?> eventType, long duration, TimeUnit unit)
    Blocks the caller until an event of a given type occurs or a given length of time elapses.

  • Method Details

    • subscribe

      void subscribe(Object observer)
      Subscribes an observer for notification of events of a given type.

      The single parameter of any method of the observer annotated with Observes identifies the type of the events observed by the observer.

      Parameters:
      observer - the observer
      Throws:
      IllegalArgumentException - if the observer declares no methods annotated with Observes, or such methods do not declare a single parameter
      See Also:

    • unsubscribe

      boolean unsubscribe(Object observer)
      Unsubscribes an observer.
      Parameters:
      observer - the observer
      Returns:
      true if the observer is found and unsubscribed.

    • fire

      void fire(Object event, String... qualifiers)
      Fires an event to all observers which have subscribed for them, blocking the client until all the critical observers have executed.
      Parameters:
      event - the event
      qualifiers - optional event qualifiers

    • waitFor

      void waitFor(Class<?> eventType)
      Blocks the caller until an event of a given type occurs.
      Parameters:
      eventType - the event type
      Throws:
      RuntimeException - if the wait is interrupted

    • waitFor

      void waitFor(Class<?> eventType, long duration, TimeUnit unit)
      Blocks the caller until an event of a given type occurs or a given length of time elapses.
      Parameters:
      eventType - the event type
      duration - the length of time to wait for
      unit - the time unit of the duration
      Throws:
      IllegalArgumentException - if the inputs are null or the duration is less or equal than zero
      RuntimeException - if the wait is interrupted

    • stop

      void stop()
      Signals that this hub is no longer needed and can release its resources.