• Java Standard Edition, JDK 8 or above
• IntelliJ IDEA IDE, 2018.1 or above

The GitHub link for this chapter is https://github.com/PacktPublishing/Hands-On-Reactive-Programming-with-Reactor/tree/master/Chapter02.

As we discussed in the previous chapter, the publisher is responsible for the generation of unbounded asynchronous events, and it pushes them to the associated subscribers. It is represented by the org.reactivestreams.Publisher interface, as follows:

public interface Publisher<T> {
    public void subscribe(Subscriber<? super T> s);
}

The interface provides a single subscribe method. The method is invoked by any party that is interested in listening to events published by the publisher. The interface is quite simple, and it can be used to publish any type of event, be it a UI event (like a mouse-click) or a data event.

Since the interface is simple, let's add an implementation for our custom FibonacciPublisher:

public class FibonacciPublisher implements Publisher<Integer> {
    @Override
    public void subscribe(Subscriber<? super Integer...

A subscriber is used to listen to events generated by a publisher. When a subscriber registers to a publisher, it receives events in the following order:

As a result, the subscriber has the following interface to handle all of these events:

public interface Subscriber<T> {
    public void onSubscribe(Subscription s);
    public void onNext(T t);
    public void onError(Throwable t);
    public void onComplete();
}

Let's cover each of these methods in detail, as follows:

• onSubscribe(Subscription s): As soon as a publisher has received a subscriber, it generates a subscription event. The generated subscription event is then received in the specified method.
• onNext (T): All data events generated by a publisher are received by the subscriber in the specified method. A publisher may or may not publish a data event before closing the stream.
• onCompletion()...

Before we jump into Reactor, let's compare the Streams model with some of the existing similar APIs, such as the java.util.Observable interface and the JMS API. We will try to determine the similarities and the key differences between the APIs.

The java.util.Observable interface implements the Observer pattern, which can be co-related here. However, all similarities end here. If we look at the Observable interface, we have the following methods:

public class Observable {
  void addObserver (Observer o);
  void deleteObserver (Observer o);
  void deleteObservers();
  void notifyObservers();
  void notifyObserver(int arg);
  int countObservers();
  boolean hasChanged();
}

Let...

Flux<T> is a general purpose reactive publisher. It represents a stream of asynchronous events with zero or more values, optionally terminated by either a completion signal or an error. It is important to note that a Flux emits the following three events:

• Value refers to the values generated by the publisher
• Completion refers to a normal termination of the stream
• Error refers to an erroneous termination of the stream:

All of the preceding events are optional. This can lead to streams of the following types:

• Infinite stream: A publisher generating only value events, and no terminal events (completion and error)
• Infinite empty stream: A stream generating no value events and no terminating events
• Finite stream: A publisher generating N finite values, followed by a terminal event
• Empty stream: A publisher generating no value events, and only terminal events...

Now that we have covered the Flux API, let's look at Mono. It is capable of generating a maximum of one event. This is a specific use case for Flux, capable of handling one response model, such as data aggregation, HTTP request-response, service invocation response, and so on. It is important to note that a Mono emits the following three events:

• Value refers to the single value generated by the publisher
• Completion refers to a normal termination of the stream
• Error refers to an erroneous termination of the stream

Since Mono is a subset of Flux, it supports a subset of Flux operators. Let's look at how to build a Mono.

Reactor Flux and Mono provide a wide choice of subscribe methods. Reactive publishers raise four types of events, namely subscription, value, completion, and error. Individual functions can be registered for each of the events. We can also register a subscriber without listening to any kind of event. Let's look at all of the possible variants offered, as follows:

fibonacciGenerator.subscribe(); (1)

fibonacciGenerator.subscribe(t -> System.out.println("consuming " + t));   (2)

fibonacciGenerator.subscribe(t -> System.out.println("consuming " + t),
                e -> e.printStackTrace() ); (3)

fibonacciGenerator.subscribe(t -> System.out.println("consuming " + t),
                e -> e.printStackTrace(),
                ()-> System.out.println("Finished")); (4)

fibonacciGenerator.subscribe...

In this chapter, we provided a detailed discussion of the publisher and the subscriber interfaces of Reactive Streams. We attempted to implement these interfaces to illustrate that there are many non-explicit rules for them. These rules have been converted into the Reactive Streams TCK, against which all implementations should be validated. We also compared the publisher-subscriber pattern with the existing Observer and JMS patterns used in Java. Next, we took a detailed look at the Flux and Mono implementations available in Reactor. We looked at methods for creating them, and then subscribed to the generated streams.

In the next chapter, we will look at the operators that can be used to modify the generated streams.

1. How can we validate Reactive Stream publisher and subscriber implementations?
2. How is the Reactive Stream publisher-subscriber model different from the JMS API?
3. How is the Reactive Stream publisher-subscriber model different from the Observer API?
4. What is the difference between Flux and Mono?
5. What is the difference between SynchronousSink and FluxSink?
6. What are the different lifecycle hooks available in Reactor?

• For more information, please refer to the video under the Reactive Streams specification, as this is part of Java 9: https://www.packtpub.com/application-development/reactive-java-9-video
• To gain in-depth knowledge on building microservices using Akka as a Reactive Streams library, please refer to the following video: https://www.packtpub.com/application-development/building-microservice-akka-http-video