Brief Intro to Reactive Streams with Project Reactor
Tags: java, programming, architecture, reactivestreams
Published: 2023-01-02 05:20:29

Source code of the examples [on Github](https://github.com/mcwiise/dev.to/tree/main/java-reactive-reactor/src/main/java/org/mcwiise/cap)

Traditional client/server communication is done by requests, where the server sets a thread to attend every request, and then the client waits for the server to complete a bunch of tasks before getting either a successful or failed response, This is what we know as _**synchronous and blocking communication**_.

Regardless the scalability strategy implemented by the server (horizontal or vertical), this sync-blocking communication could be an overhead for the app's performance, so we can use _**reactive programming**_ to handle threads in a more efficient way.

Let's explore the four communication scenarios using a call center as study case.

## Synchronous and Blocking
As we saw above this is the traditional way, where the client will be blocked until getting a response from the server.

![sync-blocking](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/etap8t9cqg0te20t8rmi.jpg)

## Asynchronous
The client will delegate the communication to the server to another thread, thus the client is released to execute other tasks while the server responses.

![async](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/iah34wsl9x7rzttaow2h.jpg)

## Non-Blocking
The client does not delegate the communication to a new thread, but registers a **_callback function_** where will receive the response after the server processes the request. This scenario also releases the client (main thread) to perform other tasks.

![non blocking](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/ncmrvbbm2rgl3z88f9cj.jpg)

## Non-Blocking Async
the client delegates the communication to a new thread, so the new thread registers the _**callback function**_ where will receive the response once the server processes the request. This scenario also releases the client (main thread) to perform other tasks.

![non blocking async](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/t3pna61t1y5g8n26iifp.jpg)

## Publisher Subscriber Pattern
Reactive programming relies on the Publisher and Subscriber pattern, where the communication between components that emit messages (publishers) and the ones which consume them (subscribers) relies on an external component called broker. 

![pub sub pattern](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/md3a3r3pfpqih9ob57at.jpg)

Publishers and subscribers are agnostic each other. More about this pattern in further posts about Apache Kafka.

## The Reactive Streams API
The [reactive streams API](https://www.reactive-streams.org/) provides the specification for **_non-blocking async_** streams processing with back pressure mechanism, and [Project Reactor](https://projectreactor.io/) is an implementation written in java.

### What is Back Pressure?
Having a look at the pub-sub pattern diagram, _**back pressure**_ is a strategy the reactive streams provide to control the number of messages (items) a subscriber is able to consume from a publisher.

Let's review the 4th interfaces provided by the API, and keep in mind the difference between these 2 concepts:

- _**item:**_ this is the element or message sent by a publisher and consumed by a subscriber. 
- **_signal:_** a flag emitted by the publisher after an item or a set of items are sent.

### Publisher Interface
- Responsible for emitting items along with a _**success**_ or _**failure**_ signals.
- It also provides a method for the subscribers to register a _**subscription**_.

### Subscription Interface
- This is a _**channel**_ where the subscriber is able to request more items to the publisher (back pressure mechanism).
- The subscriber can also **_cancel the subscription_** using the same channel.

### Subscriber Interface
A subscriber _**reacts to the items and signals**_ emitted by the publisher by the implementation of the following methods:

- **_onComplete()_** The publisher notifies the subscriber that there are no more items to emit, then the publisher is closed.
- **_onError(Throwable)_** The publisher notifies the subscriber that there was an error when emitting the item, then the publisher is closed.
- _**onNext()**_ The publisher notifies the subscriber that a next item was just emitted.
- _**onSubscribe()**_ This is the initial state, where the publisher notifies the subscriber that the subscription was accepted.

### Processor Interface
This is an intermediary stage, a processor can consume an item and emit a new one.

## [The Project Reactor](https://projectreactor.io/)
As implementation of reactive streams API, reactor is based on two types of publishers: `Mono` and `Flux`. Be aware that reactive streams are lazy evaluated, it means that nothing happens until a subscription is created.

### Mono<T>
A `Mono` is a publisher able to emit from **_zero to 1 items._** followed by an **_onComplete()_** or _**onError()**_ signals. Let's explore some basic functions to get started.

**Emit And Subscribe**
```java
1. public static void emitAndSubscribe(){
2.   var mono = Mono.just("A");
3.   mono.subscribe(
4.     item -> System.out.println(item),
5.     error -> System.out.println(error.getMessage()),
6.     () -> System.out.println("Completed!")
7.   );
8. }
```
- _line 2:_ We create mono publisher which emits 1 item.
- _line 3:_ We create a subscription which receives 3 functions: 2 Consumer<T> and 1 Runnable
- _line 4:_ the consumer that accepts the item just emitted.
- _line 5:_ the consumer that accepts an exception in case it is thrown by the publisher.
- _line 6:_ the runnable which reacts to _**onComplete()**_ signal.

_Outcome:_
> A
> Completed!

### Flux<T>
A `Flux` is a publisher able to emit from **_zero to many_** items, followed by an **_onComplete()_** or _**onError()**_ signals. Let's explore some basic functions to get started:

**Emit And Subscribe**
```java
1. public static void emitAndSubscribe(){
2.   var flux = Flux.fromArray(new String[]{"A", "B", "C"});
3.   flux.subscribe(
4.     item -> System.out.println(item),
5.     error -> System.out.println(error.getMessage()),
6.     () -> System.out.println("Completed!")
7.   );
8. }
```
- _line 2:_ We create flux publisher which emits 3 item.
- _line 3:_ We create a subscription which receives 3 functions: 2 Consumer<T> and 1 Runnable
- _line 4:_ the consumer that accepts the item just emitted.
- _line 5:_ the consumer that accepts an exception in case it is thrown by the publisher.
- _line 6:_ the runnable which reacts to _**onComplete()**_ signal.

_Outcome:_
> A
> B
> C
> Completed!

## Conclusion
The [reactor java doc](https://javadoc.io/doc/io.projectreactor/reactor-core/latest/index.html) is plenty of methods to work with Monos and Fluxes, we have just covered some basic ones to understand the concepts of publishing and subscribe to get  started.