Made withSpring BootTutorial

Spring Boot WebFlux | Project setup & understanding Mono & Flux Internal Workflow | JavaTechie

J

Java Techie

Creator on the Spring Boot channel

Watch on YouTube ↗

145K viewsApr 1, 202115:26 runtime

Learn how to set up a Spring Boot WebFlux project and dive into the internal workflow of Project Reactor's Mono and Flux publishers. This tutorial guides you through creating a reactive application from scratch using the Spring Initializr in IntelliJ. You'll see practical…

Chapters

#springboot#tutorial#frameworks

Project Setup

To begin working with Spring WebFlux, you create a Spring Boot project and include a key dependency:

Dependency: spring-boot-starter-webflux
This starter transitively includes reactor-core, which provides the Mono and Flux APIs.
The demonstration uses a JUnit test class (MonoFluxTest) to explore reactive concepts without needing to build a full API.

Understanding Publishers: Mono and Flux

Project Reactor provides two core Publisher types for handling asynchronous data streams:

Mono<T>: Represents a stream of 0 or 1 items. It's used for asynchronous operations that will yield a single result or no result at all.
- Example: Mono<String> monoString = Mono.just("javateki");
Flux<T>: Represents a stream of 0 to N items. It's used for handling sequences of multiple data elements.
- Example: Flux<String> fluxString = Flux.just("Spring", "Spring Boot", "Hibernate");

Note
Both Mono and Flux are lazy. Nothing happens until a Subscriber subscribes to the stream.

The Reactive Streams Workflow

The interaction between a Publisher (Mono/Flux) and a Subscriber follows a strict protocol defined by the Reactive Streams specification:

Subscription: The process begins when a Subscriber calls the .subscribe() method on a Publisher.
onSubscribe: The Publisher responds by calling the onSubscribe(Subscription subscription) method on the Subscriber. This passes a Subscription object, establishing the connection.
request(n): The Subscriber uses the Subscription object to request a number of items (n) from the Publisher. This implements backpressure.
onNext(T t): The Publisher sends data items to the Subscriber by calling onNext for each item, up to the number requested.
Termination: The stream completes in one of two ways:
- onComplete(): Called by the Publisher to signal that all data has been sent successfully.
- onError(Throwable t): Called by the Publisher if an error occurs during processing.

Verifying the Workflow with `.log()`

A powerful tool for observing the internal events of a reactive stream is the .log() operator.

Placing .log() in a reactive chain prints every signal (onSubscribe, request, onNext, onComplete, onError) to the console.
This makes it an invaluable debugging tool for visualizing the flow of data and events.

// Example: Logging a Mono stream
Mono.just("javateki")
    .log()
    .subscribe(System.out::println);

Error Handling

Error handling is a critical part of the reactive contract.

Important
Once an onError signal is emitted, the stream is terminated. No further onNext or onComplete events will be sent, even if there are more operators or data sources chained after the error.

An error can be introduced into a stream using Mono.error(new Exception()) or Flux.error(new Exception()).
The .subscribe() method can be overloaded to handle these error signals gracefully.

// Example: Subscribing with an error handler
fluxString
    .log()
    .subscribe(
        data -> System.out.println("Received: " + data),
        error -> System.err.println("Error: " + error.getMessage())
    );

Introduction and Project Setup

Hi everyone, welcome to Java Techy. In this tutorial we'll be exploring about mono and plugs with its internal workflow. Okay, also we'll verify the steps which we discuss in reactive stream workflow using this mono and phlox.

Basically we'll be creating one small project to demonstrate these flux and mono and its internal flow. So let's quickly create one spring boot reactive project. Go to your favorite ID, I will be using IntelliJ ID. Then click on file, click new, click on project, click spring assistant. You should have this plugin in your ID.

Now click on next. Then change the group ID here, so I'll add com.javateki. Then change the artifact ID, I'll name it springboot-webflux-demo. Then change your java version, I will use 8. Let me copy the artifact ID so that I will specify same as my project name. Then change the package here: com.javateki.dot.webflox.

Now click on next. Now you need to add the reactive dependency. So let me type here spring web. If you observe there is two spring web dependency. One is spring web and another is spring reactive web. Okay, since we are doing the reactive programming I need to choose this spring reactive.

Examining Project Dependencies

Now click on next, finish. Then go inside your project. Now if you will open your pom.xml, you could see the latest version of springboot we are using, 2.4.4 and we added only the dependency spring boot starter wavefox, okay?

And this is the default dependency for test and there is a dependence called reactor test and this is what the maven plugin, right? So only one dependency which you added that is the springboot starter webflox.

Creating a Mono Test Case

Now I'm not going to write any API here, I'll just write one test case to show you the workflow of mono and flux. Now go inside your src test, then I'll create one class. I'll name it MonoFluxTest.

So we'll be begin with mono. So what I'll do, I'll write one method, public void then testMono. Okay so there is multiple way you can create object of mono. So what I will do I will just type Mono.just. This is how you can create object of this mono. Now as we already understood mono can only handle one element. So I will add here any kind of data type, I will add abc, if I want I can type 1. It's up to you which kind of data type we want to use. Okay, so for this example I will use something called string, "javateki".

Now the return type will be Mono of string, okay? I'll name it monoString. Fine, this is how you can create object of mono who can handle only one object. Fine.

Subscribing to a Mono and Initial Run

Now if you will go inside your Mono class, so let me go to this implementation. If you observe the packet structure, reactor.core.publisher, it means this mono and flux will act as a publisher, right? So to access any publisher, first subscriber need to call subscribe method of publisher. So here my publisher is monoString. Now subscriber need to call monoString is my publisher, .subscribe, right?

So if you remember this is what the workflow we understand right? First subscriber need to call subscribe method of publisher, that's what we are doing here. This is my publisher and we are invoking the subscribe method. Then if you observe the subscribe method is one overloaded method. There is a multiple subscribe method. Okay, you can observe here. So what I want to do here, I just want to print the published event. So I will directly use method reference System.out::println. Fine.

Now I just annotate this with @Test. Let's run this piece of code. Run this MonoFluxTest. Yeah, so we are seeing the event here right? Once we call publisher.subscribe method, then immediately publisher start emitting the event. That's how we can see the result.

Logging the Reactive Stream Workflow

Now what I need to verify, I just want to verify whether this mono and flux is really following the reactive stream workflow or not. So for that reason I will add one method called log. Now each of execution will be printed in console. Now let's rerun this piece of code.

Yeah, so if you observe the output there is a onSubscribe method call, then there is a request method call from the subscription interface, then onNext method call from the subscriber interface, then onComplete event, right? So let's go to the diagram once again. The first step, we just need to subscribe to the publisher by calling this subscribe method. Then publisher need to send the subscription event to subscriber. Then if you observe the third request, third need to be the request method from the subscription, right?

So subscriber is invoking this request(n) method to inform to the publisher, give me n number of data. That's how the method argument is n. And then the fourth step will be onNext, right? And if there is n number of event, we will get n number of onNext method call. As you are trying with the mono I cannot show you this n number of event. Once you will try with the flux I can show you this output. Then at end you can observe we are receiving the onComplete event.

Simulating and Handling Mono Errors

Now if you go back to the console, the first method is onSubscribe, then there is a request method from the subscription interface with n number of argument. Then once publisher called this onNext method we are receiving the event. Then as everything is okay there is no error, we are getting onComplete event. Now if you go to the diagram, if there is any failure we should receive onError event right? Now let's verify this onError event.

So what I will forcefully try to throw some exception from this mono. So let me minimize this. Then what we'll do we'll just add here, then I just want to Mono.error, new something called RuntimeException. Just pass the message, "exception occurred" something like that. Okay, just change this to type generic. Fine, because this will return the object and we are mapping here the string.

So now forcefully we are throwing some error while processing the mono. Now let's verify whether it return onError event or not. Let me run this piece of code. We're getting onSubscribe then request then onError, there is no onComplete event. If you observe there is no request onNext event as well, okay. Now we are not handling this exception. So what we can do, as this subscribe is overloaded method I will try to print the exception. System.out... I'll just print this message. Now let's rerun this. It called onSubscribe, request, onError. There is no onNext call and onComplete call. Okay. So the error handling scenario is something different in reactive programming. So we'll cover a separate session for error handling. This is just I was trying to show you the flow of onComplete and onError. Okay.

Creating a Flux Test Case

Now we will just verify if publisher will publishing n number of event whether we are getting n number of onNext call or not. So what we can do I will go with the flux. So I'll write one another test case public void testFlux. Then I just need to create the flux object, Flux.create. I will pass your n number of elements, or I can just name it to Flux.just. Okay, so here I will give something called a spring, spring boot, then I'll add hibernate, some random value okay. So I'll add another string, let's say microservice. Fine. So define the local variable fluxString. Fine.

Now this fluxString will be act as my publisher. So to consume this what I need to do? I just need to call the subscribe method of my publisher. So just call it. Once you call this subscribe method on flux it will immediately start emitting the event. It will emit the four event. Okay, so what I'll do I will just print it. So I'll copy this directly. Fine.

Now let me run this piece of code. So we need to annotate @Test. Let's run this. We can see the result right? Spring, spring boot, hibernate and microservices.

Logging the Flux Workflow and Concatenation

Each of the message or each of the data published as a separate onNext call. So to verify that what we can do we'll just add one log method. Now let's run this code, not entire test case because on the first one we are getting the exception right. I'll run this. Yeah, so if you observe the first call is onSubscribe, then request n number of object from the subscription interface. Then if you observe on the first onNext call we are receiving one event. On the second onNext call we are receiving another event. Like we have four so here four onNext event, right? At end we are getting onComplete because there is no error.

Now in middle if I want to add something let's say I'll just add .concatWith value, I'll add something called aws. Now my publisher will fire five event so there should be five onNext call. Now let me run this. Yeah, so if you observe there is five onNext call.

Simulating Flux Errors and Final Recap

Now let's verify the error scenario. So what I want to do, I just want to forcefully send some error. Okay, so I can write concatWith(Flux.error(...)). I'll return the same exception object. So I will add the message, "exception occurred in flux". Fine. Now if I will run this, we are not handling the exception. So let me add this. We are not handling we are just printing the message. So this will be second argument of subscriber. Now let's run this.

So if you observe here there is onSubscribe call, request call, then 5 onNext event then at finally we are receiving onError event, right? As there is error we are getting onError. If there is everything succeed we should receive onComplete event. Now what we just need to verify after this error if I will add some value to this existing flux whether it will do the next call or not? So what I will do, I will add something called here cloud, so many random string. Now let me run this. So if you observe this output we don't have this cloud onNext event right? Still you can see only five onNext and at end we are getting an error.

So this is how it follow the reactive stream workflow. Once again if you see this diagram, the first we need to subscribe, then publisher need to send the subscription event, then there will be a request n call. So let me remove this. Yeah, so there will be request n method call from the subscriber from the subscription interface. Then there will be onNext event will be published from the publisher to the subscriber. And the number of record you have you will find that many and onNext event, then at end you will find onComplete or onError event. This is what we understand with our code, right? So I believe this concept is clear for you. In our upcoming tutorial we will understand how this asynchronous and non-blocking works in reactive programming with live example. Okay, so once again, don't ignore this theory session, just try to brush up once again so that it will help you to understand the next tutorial. That's all about this particular video guys. Thanks for watching this video. Meet you soon with a new concept.