Extending CDI Observer pattern to support global events

CDI introduced a convenient implementation of the observer pattern into the JavaEE world. Using the CDI API, components can emit events, or receive events created by other components (using the @Observes annotation). This allows developers to reduce the coupling between the emitters and the receivers of events.

However, the current API does not allow to handle what we call “global” events. Global events are propagated among all active sessions, i.e., current users and are not restricted to the current session.

This blog post briefly explores the CDI observer pattern and explains how it can be extended to support global events.

CDI Observer Pattern

Let’s take an application where all modifications to data have to be logged. Additionally, you need to notify a (stateful) web controller about the change, because it will affect its current view. Instead of calling a logging service EJB and informing the controller explicitly, you can simply emit an event. All interested observers receive the event.

Emitting an event is done using using an injected Event object. Note the ArticleCreateEvent acting as the event’s topic.

@Inject private Event&lt;ArticleCreatedEvent&gt; _articleCreatedEvent;
...
_artikelCreatedEvent.fire(new ArticleCreatedEvent(article));

In our logger component, the reception method has a parameter of the ArticleCreatedEvent type annotated with the @Observes annotation. This instructs CDI as to which method should be called and by which kind of event the method is interested in.

public void observeArticleCreatedEvent(@Observes final ArticleCreatedEvent event) {
  _loggingService.addLogEntry(
    event.getArtikel().getId(),
    EntityAction.CREATED,
    _currentUser.getUserId());
}

Finally, the web controller will display the newly created article carried by the event:

public void observeArticleCreatedEvent(@Observes final ArticleCreatedEvent event) {
  showNewArticle(event.getArticle());
}

Lack of global events

Even if this approach is elegant, it has one major drawback: with the existing API you can not send “global” events. For instance, you cannot send events to all active clients (sessions) of your web application.

This would be very useful in several situations: users always like to be informed if someone else has changed their data, or if some relevant system event has occurred. If someone has created a new article, for example, you could inform all controllers that their article data needs to be reloaded.

There are several solutions addressing this problem, but it would be most elegant to extend the existing CDI observer mechanism to inform all existing instances of your controllers.

You might argue that there is a already a scope to hold global data: classes annotated with @ApplicationScoped. But firing an event using an application scoped class will not have the desired effect; it will still inform the current session only. The reason for this behavior is simple: all session scopes know the application scope, but not vice versa.

The rest of this post will explain how we can extend the CDI Observer pattern to support global events.

Solution part 1: Session controller

The solution consists of several pieces. First we need a GlobalevHttpSessionController that registers and handles all HTTP sessions and delegates all events to the sessions:

@ApplicationScoped
public class GlobalevHttpSessionController {
  public static final String 
    EVENT_ATTRIBUTE_NAME = &quot;HttpSessionControllerEvent&quot;;

  private final List _httpSessions = new ArrayList();

  public List getHttpSessions() {
    return new ArrayList(_httpSessions);
  }

  public void addSession(final HttpSession httpSession) {
    _httpSessions.add(httpSession);
  }

  public void removeSession(final HttpSession httpSession) {
    _httpSessions.remove(httpSession);
  }

  public void fireEvent(final GlobalEvent eventObject) {
    for (final HttpSession session : _httpSessions) {
      fireEvent(session, eventObject);
    }
  }

  private void fireEvent(final HttpSession session, final GlobalEvent eventObject) {
    try {
      final List globalEvents = getGlobalEvents(session);

      globalEvents.add(eventObject);
    } catch (final Exception e) {
      throw new IllegalStateException(&quot;fireEvent&quot;, e);
    }
  }

  private synchronized List getGlobalEvents(final HttpSession session) {
    List globalEvents = (List) session.getAttribute(EVENT_ATTRIBUTE_NAME);

    if (globalEvents == null) {
      globalEvents = new ArrayList();
      session.setAttribute(EVENT_ATTRIBUTE_NAME, globalEvents);
    }

    return globalEvents;
  }
}

Where GlobalEvent is just a simple serializable super class for all global events:

public abstract class GlobalEvent implements Serializable {
  private static final long serialVersionUID = 1L;
}

Solution part 2: HTTP Session Listener

Next, we need an HTTP listener to add and remove active client sessions:

@WebListener
public class GlobalevHttpSessionListener implements HttpSessionListener {
  @Inject
  private GlobalevHttpSessionController _httpSessionController;

  public void sessionCreated(final HttpSessionEvent se) {
    _httpSessionController.addSession(se.getSession());
  }

  public void sessionDestroyed(final HttpSessionEvent se) {
    _httpSessionController.removeSession(se.getSession());
  }
}

Solution part 3: JSF Phase Listener

So far so good. By now, we can fire events using the session controller and queue them into a special attribute of all HTTP sessions.

But how do we dispatch them to the local clients? You could, for instance, use some asynchronous polling mechanisms. This approach has obvious drawbacks, since asynchronous code is error-prone and complicated in most cases. You also have to be careful not to cause too much overhead load with constant polling.

Instead, we chose to look up incoming events using a JSF PhaseListener. Other solutions, for example using Filters, are also possible.

The listener looks for global events and dispatches them to the local session using the CDI bean manager. The manager instance is obtained via a JNDI lookup (by specification the bean manager has to be bound by the container under java:comp/BeanManager). We cannot use CDI injection here, unfortunately, since the phase listener is not instantiated by CDI, but by the Java Server Faces (JSF) framework. The framework however does supply all the information we need to access the HTTP session via the FacesContext object.

public class GlobalevEventPhaseListener implements PhaseListener {
  public void beforePhase(final PhaseEvent event) {
    final FacesContext facesContext = event.getFacesContext();
    final HttpSession httpSession =   
        JSFUtil.getHttpSession(facesContext);

    if (httpSession != null) {
      final List globalEvents = getGlobalEvents(httpSession);

      if (!globalEvents.isEmpty()) {
        fireEvents(globalEvents);
      }
    }
  }

  private void fireEvents(final List globalEvents) {
    final BeanManager beanManager = lookBeanManager();

    if (beanManager != null) {
      try {
        for (final GlobalEvent devaGlobalEvent : globalEvents) {
          beanManager.fireEvent(devaGlobalEvent);
        }
      } catch (final Exception e) {
        throw new IllegalStateException(&quot;fireEvents&quot;, e);
      }
    }
  }

  @Override
  public PhaseId getPhaseId() {
    return PhaseId.RENDER_RESPONSE; // RESTORE_VIEW;
  }

  private BeanManager lookBeanManager() {
    try {
      final Object obj = 
        new InitialContext().lookup(&quot;java:comp/BeanManager&quot;);

      return (BeanManager) obj;
    } catch (final Exception e) {
           throw new 
            IllegalStateException(&quot;Lookup bean manager&quot;, e);
    }

    return null;
  }

  private synchronized List getGlobalEvents(final HttpSession httpSession) {
    final List events = (List) httpSession.getAttribute(
        GlobalevHttpSessionController.EVENT_ATTRIBUTE_NAME);
    final List result = new ArrayList();

    if (events != null) {
      result.addAll(events);
      events.clear();
    }

    return result;
  }
}

Finally, we need to register our listener in faces-config.xml:

&lt;lifecycle&gt;
  &lt;phase-listener&gt;
de.akquinet.jbosscc.globalev.listener.GlobalevEventPhaseListener
  &lt;/phase-listener&gt;
&lt;/lifecycle&gt;

Conclusion

Using this approach you can now easily notify all clients in an elegant way, without direct coupling between the emitters and receivers in different sessions.

There is a sample project at GitHub which contains the code described above. It has been developed and tested using JBoss AS7. After deployment you can access the application with two different web browsers at http://localhost:8080/jbosscc-globalev and send messages to each other.

Suggestions and comments are welcome to the author Markus Dahm.