Observer pattern usage example

Now, let's write a simple JUnit test that uses our classes and demonstrates how all of them play together. Also, in the following example, we are using the Mockito library (http://site.mockito.org) in order to verify expectations with the support of the Spies Pattern:

@Test
public void observersHandleEventsFromSubject() {
   // given
   Subject<String> subject = new ConcreteSubject();
   Observer<String> observerA = Mockito.spy(new ConcreteObserverA());
   Observer<String> observerB = Mockito.spy(new ConcreteObserverB());

   // when
   subject.notifyObservers("No listeners");

   subject.registerObserver(observerA);
   subject.notifyObservers("Message for A");

   subject.registerObserver(observerB);
   subject.notifyObservers("Message for A & B");

   subject.unregisterObserver(observerA);
   subject.notifyObservers("Message for B");

   subject.unregisterObserver(observerB);
   subject.notifyObservers("No listeners");

   // then
   Mockito.verify(observerA, times(1)).observe("Message for A");
   Mockito.verify(observerA, times(1)).observe("Message for A & B");
   Mockito.verifyNoMoreInteractions(observerA);

   Mockito.verify(observerB, times(1)).observe("Message for A & B");
   Mockito.verify(observerB, times(1)).observe("Message for B");
   Mockito.verifyNoMoreInteractions(observerB);
}

By running the preceding test, the following output is produced. It shows which messages have been received by which Observer:

Observer A: Message for A
Observer A: Message for A & B
Observer B: Message for A & B
Observer B: Message for B

In the case when we do not need to cancel subscriptions, we may leverage Java 8 features and replace the Observer implementation classes with lambdas. Let's write the corresponding test:

@Test
public void subjectLeveragesLambdas() {
   Subject<String> subject = new ConcreteSubject();

   subject.registerObserver(e -> System.out.println("A: " + e));
   subject.registerObserver(e -> System.out.println("B: " + e));
   subject.notifyObservers("This message will receive A & B");
   ...
}

It is important to mention that the current Subject implementation is based on the CopyOnWriteArraySet, which is not the most efficient one. However, that implementation is thread-safe at least, which means that we are allowed to use our Subject in the multithreaded environment. For example, it may be useful when events are distributed through many independent components, that usually work from multiple threads (it is especially valid nowadays, when most applications are not single threaded). Throughout the course of this book, we will be covering thread safety and other multithreaded concerns.

Do keep in mind that when we have a lot of observers that handle events with some noticeable latency—as introduced by downstream processing—we may parallel message propagation using additional threads or Thread pool. This approach may lead to the next implementation of the notifyObservers method:

private final ExecutorService executorService = 
   Executors.newCachedThreadPool();

public void notifyObservers(String event) {
   observers.forEach(observer ->
           executorService.submit(
                   () -> observer.observe(event)
           )
   );
}

However, with such improvements, we are stepping on the slippery road of homegrown solutions that are usually not the most efficient, and that most likely hide bugs. For example, we may forget to limit the thread pool size, which eventually leads to an OutOfMemoryError. A naively configured ExecutorService may create a growing number of threads in situations where clients ask to schedule tasks more frequently than the executors can finish their current ones. And because each Thread consumes around 1 MB in Java, a typical JVM application has a chance to exhaust all available memory by creating a few thousand threads.

To prevent excessive resource usage, we may restrict the thread pool size and violate the liveness property of the application. Situations such as this arise when all available threads attempt to push some events to the same sluggish Observer. Here, we have just scratched the surface of the potential problems that can occur. Also, as stated in the white-paper: