Observer pattern

To move things along, we need to remind ourselves about a particular pretty old and well-known design pattern—the Observer pattern. That is one of the twenty-three famous GoF (Gang of Four) design patterns. At first glance, it may appear that the Observer pattern is not related to reactive programming. However, as we will see later, with some small modifications, it defines the foundations of reactive programming.

The Observer pattern involves a subject that holds a list of its dependants, called Observers. The subject notifies its observers of any state changes, usually by calling one of their methods. This pattern is essential when implementing systems based on event handling. The Observer pattern is a vital part of the MVC (Model-View-Controller) pattern. Consequently, almost all UI libraries apply it internally.

To simplify this, let's use an analogy from an everyday situation. We can apply this pattern to the newsletter subscription from one of the technical portals. We have to register our email address somewhere on the site of our interest, and then it will send us notifications in the form of newsletters, as shown in the following diagram:

The Observer pattern makes it possible to register one-to-many dependencies between objects at runtime. At the same time, it does this without knowing anything about the component implementation details (to be type safe, an observer may be aware of a type of incoming event). That gives us the ability to decouple application pieces even though these parts actively interact. Such communication is usually one directional and helps efficiently distribute events through the system, as shown in the following diagram:

As the preceding diagram shows, a typical Observer pattern consists of two interfaces, Subject and Observer. Here, an Observer is registered in Subject and listens for notifications from it. A Subject may generate events on its own or may be called by other components. Let's define a Subject interface in Java:

public interface Subject<T> {
   void registerObserver(Observer<T> observer);
   void unregisterObserver(Observer<T> observer);
   void notifyObservers(T event);
}

This generic interface is parametrized with the event type T, which improves the type safety of our programme. It also contains methods for managing subscriptions (the registerObserver, unregisterObserver, and notifyObservers methods) that trigger an event's broadcasting. In turn, the Observer interface may look like the following:

public interface Observer<T> {
   void observe(T event);
}

The Observer is a generic interface, which parametrized with the T type . In turn, it has only one observe method in place to handle events. Both the Observer and Subject do not know about each other more than described in these interfaces. The Observer implementation may be responsible for the subscription procedure, or the Observer instance may not be aware of the existence of the Subject at all. In the latter case, a third component may be responsible for finding all of the instances of the Subject and all registration procedures. For example, such a role may come into play with the Dependency Injection container. This scans the classpath for each Observer with the @EventListener annotation and the correct signature. After that, it registers the found components to the Subject.

Now, let's implement two very simple observers that simply receive String messages and print them to the output stream:

public class ConcreteObserverA implements Observer<String> {
   @Override
   public void observe(String event) {
      System.out.println("Observer A: " + event);
   }
}
public class ConcreteObserverB implements Observer<String> {
   @Override
   public void observe(String event) {
      System.out.println("Observer B: " + event);
   }
}

We also need to write an implementation of the Subject<String>, which produces String events, as shown in the following code:

public class ConcreteSubject implements Subject<String> {
   private final Set<Observer<String>> observers =                 // (1)
           new CopyOnWriteArraySet<>();

   public void registerObserver(Observer<String> observer) {
      observers.add(observer);
   }

   public void unregisterObserver(Observer<String> observer) {
      observers.remove(observer);
   }

   public void notifyObservers(String event) {                     // (2)
      observers.forEach(observer -> observer.observe(event));      // (2.1)
   }
}

As we can see from the preceding example, the implementation of the Subject holds the Set of observers (1) that are interested in receiving notifications. In turn, a modification (subscription or cancellation of the subscription) of the mentioned Set<Observer> is possible with the support of the registerObserver and unregisterObserver methods. To broadcast events, the Subject has a notifyObservers method (2) that iterates over the list of observers and invokes the observe() method with the actual event (2.1) for each Observer. To be secure in the multithreaded scenario, we use CopyOnWriteArraySet, a thread-safe Set implementation that creates a new copy of its elements each time the update operation happens. It is relatively expensive to update the contents of the CopyOnWriteArraySet, especially when the container holds a lot of elements. However, the list of subscribers does not usually change often, so it is a fairly reasonable option for the thread-safe Subject implementation.