DEV Community: Lovegiver

Reactive Pipeline : the App (part 2)

Lovegiver — Sat, 23 Jul 2022 18:35:00 +0000

The "ReactivePipeline" project is a simple interface, a single class project. You can find link to the Github repository in Part. #1 of this post.

The ReactiveContext class just contains the static methods we need to instantiate the necessary objects for creating a persistent Flux pipeline for your whole application.
We will see each of them in the first part of this document.

Nevertheless, these methods / objects are not the only ones that are required to build a reactive app.
So the second part of this README will present you all the objects needed to make this API complete and working.

How things may happen, how it will work

To start, just think about the fact that among all the operations of a particular process, we can distinguish :

Starting operations which do not take any arguments, thus have no predecessors
Ending operations which take arguments and have no successors
Intermediate operations are the others : they need arguments from their predecessors and produce outputs that will be their successors' inputs

Most of the time, we use to design and build apps containing methods which are triggered sequentially in a very procedural way. This can be represented by a straight line of processing operations : A --> B --> C --> ...

But we can also imagine operations as a tree in which methods A and B are independent, so parallelized, and both producing a result which C deserves. In such a case, the C function will take result_A and result_B as arguments and we'll have to synchronize both operations in order to pass their respective results to C : C(result_A, result_B)
Or, at the opposite, a A function producing a result_A which will be consumed in a parallelized manner by B and C as soon as it will be available : B(result_A) // C(result_A)

To face all of these situations, we need a flexible data-structure where data - thus functions producing these data - will be organized smartly. Wrappers has been used for this to be possible.

The toolbox

Many objects we'll talk about are wrappers. It is important to understand how they interact with each others.

Operation is the corner stone of our model. It is a Functional Interface. Each action, each method, each function, has to be an Operation. An Operation takes a varargs of Flux(es) as arguments and produces a Flux.
Task wraps a single Operation. It is a class with some useful properties and methods. It triggers Operation execution and inject the produced Flux into the next Operation to maintain the reactive behavior.
WorkGroup is a wrapper for a set of Tasks, but you won't use it directly. You will use it only if you decide to create your own Optimizer. If you rely on the default Optimizer, the Pipeline will create WorkGroups for you. All you have to understand about a WorkGroup is that it groups all Tasks involved into the realization of a common final Operation.
Finally, the Pipeline is a wrapper for a set of Tasks (and also for one or more WorkGroups as it will dispatch all the tasks in different work-groups).
A last object to talk about is the DataStreamer. It's aside from preceding objects as it is not a wrapper but the mechanism used to export the state of all Monitorable classes : Task, WorkGroup and Pipeline all have a Monitor property that describes their current state (new, running, done, in error). The DataStreamer produces a Flux containing all the states of all objects within the Pipeline. Each time one's state changes, a tick is triggered by a Notifier to the DataStreamer, which in turn triggers a new Flux that can, for example, be displayed on a web page for monitoring purpose. But it can be whatever you need. This is just the way I've chosen to talk about what is known under the hot stream name.

The global philosophy is :

We create all necessary Operations. Try to think this object as a pure function, doing just one thing.
Each Operation is wrapped into a Task object. To be instantiated, a Task must have a single Operation and a Set of all the previous Tasks whom produced Fluxes are arguments for this Task.
All the Tasks will finally be used as arguments for a Pipeline. The Pipeline, thanks to its Optimizer, create one or more WorkGroups. Once this is made, all WorkGroups will be executed in parallel threads and in an asynchronous manner.

Objects from the ReactiveContext

The Pipeline

The Pipeline class is a wrapper for a set of Tasks. When calling its .execute() method, then all Operations will be executed from the very first starting ones to the ending ones.

You can obtain a Pipeline using :

    static Pipeline createPipeline(String pipelineName, Set<Task> allTasks)

    static Pipeline createPipeline(String pipelineName, Set<Task> allTasks, WorkGroupOptimizer optimizer)

With the second method, you'll have to define your own Optimizer. This means that you define you own logic to group Tasks into WorkGroups. In order to define your own Optimizer, you'll have to implement the following Functional Interface :

    Collection<WorkGroup> optimize(Set<Task> allTasks)

The existing default Optimizer's logic is very basic and may be hugely improved and optimized (it is part of my To-Do list).

First it looks for all ending (final, terminal) Tasks
Then, it groups into a same WorkGroup all the Task involved into the realization of the same final Task

The Task

The Task is the Operation's wrapper. The Operation defines the domain's logic whereas the Task organize the interactions with other Operations.
This is why the Task takes a List<Task> as argument.

    static Task createTask(String taskName, Operation operation, List<Task> predecessors)

When defining a Task T, what you concretely do is :

Referencing the Operation to execute
Referencing the Task(s) whom resulting Flux(es) will be used as argument for the task T's Operation

The DataStreamer

The DataStreamer produces a hot stream, a potentially never ending Flux.
For the sake of demonstration, we used it as an exportable monitoring tool. This means that you can define a REST controller and a GET method returning a Flux<ServerSentEvent> that will be consumed by a web app.

    static Flux<ServerSentEvent<String>> getAllPipelinesStatesFlux()

    static Flux<ServerSentEvent<String>> getSinglePipelineStatesFlux(Pipeline pipeline)

In the case you define multiple Pipelines, the DataStreamer can produces a Flux for all of them or a single one. For a single Pipeline, you'll use the second method.

Other useful objects

Other useful objects are not directly accessible through the ReactiveContext class.

Operation

As already said, Operation is the corner-stone of this API. This interface is used to define you domain's logic. As it is a Functional Interface, you can use it as a Lambda expression.

    Flux<?> process(Flux<?>... inputs) throws TaskExecutionException;

We can take some frustrating examples to show how to use it :

    Operation o1 = inputs -> Flux.range(1,10);
    Operation o2 = inputs -> Flux.range(91,100);
    Operation o3 = inputs -> {  
      Flux<?> int1 = inputs[0];  
      Flux<?> int2 = inputs[1];  
      return Flux.zip(int1, int2, (x, y) -> (int) x + (int) y);  
    };

Operation o1 will produce a Flux<Integer> : 1, 2, 3... 10
Operation o2 will produce a Flux<Integer> : 91, 92, 93... 100
Operation o3 will use each single value from preceding Fluxes by creating tuples that will be processed for the producing of a new result (it is the way to use the .zip operator) :

(1, 91) will produce 92
(2, 92) will produce 94
(3, 93) will produce 96
etc.

Of course, this is possible only if you have created the necessary Tasks objects around your Operations :

    Task t1 = ReactiveContext.createTask("Integer Flux 1", o1, Collections.emptyList());
    Task t2 = ReactiveContext.createTask("Integer Flux 2", o2, Collections.emptyList());
    Task t3 = ReactiveContext.createTask("Sum t1 t2", o3, List.of(t1, t2));

There's many things to say here.

the Operation's single abstract method, process(Flux... inputs), may take 0, 1 or N Flux(es) as argument. This is why the Lambda expression starts this way : inputs -> ... ; In the case of a starting Operation, an Operation without any predecessors, there's no inputs to process but we have to respect the method's signature. In the example above, only the o3 operation has inputs to process and this is done by getting them from the array of Fluxes produced by the varargs argument.
the t1 and t2 Tasks wrap starting Operations, that's why there's no previous Tasks to declare here. But we still have to pass an empty collection as argument.
the t3 Task do have predecessors, respectively the t1 and t2 Tasks which are respectively wrapping o1 and o2 Operations. In that case, we pass a collection made of the t1 and t2 Tasks. This collection is a List, because the order of the argument matters of course.

Notifier / StateNotifier

You theoretically will not have to handle the Notifier (interface) and the StateNotifier (implementation) in charge of notifying the DataStreamer of any change in any Monitorable's inner state.
We here use a Visitor's pattern to delegate the action of notifying to an independent object knowing the DataStreamer and the Pipeline it is reporting about.

Monitorable / Monitor

The Monitor is a class holding the inner state of any Monitorable object. Like we have seen before, Monitorable objects are :

Pipeline
WorkGroup
Task

The Monitorable class is the mother class from which the 3 above objects are derived. Its properties are :

    String name;
    Monitor monitor;
    Notifier notifier;
    Map<Task, Optional<Flux<?>>> inputFluxesMap = Collections.synchronizedMap(new LinkedHashMap<>());

This last property is a central part of our system. We actually use it exclusively for Tasks management, but it also could be used for WorkGroups with an adapted Optimizer.
You may have noticed that our Monitorables are designed like a Composite object. Thus, not only Tasks may receive Fluxes as arguments from predecessors, but also WorkGroups. This is still a work-in-progress that will be solved with a non-basic Optimizer.

Have fun.

Lovegiver

Written with StackEdit.

Reactive Pipeline : a starter (part 1)

Lovegiver — Wed, 20 Jul 2022 18:58:00 +0000

Reactive Pipeline : a starter (part 1)

This article is about an app I've made for learning the basics of Reactive Programming in Java.
It was in reaction to the lack of practical sources that I've decided to make it on my own.
There's a lot of resources already available, and I really thank all the authors for giving me the ABC of this definitely new programming paradigm, but they often describe the same concepts and basic stuff. It is important of course, but the subject is hard when time comes to make it concrete and working.

This post is an introduction. I need it to explain my motivations and convince you that the points I will mention are relevant. Explanations about the app itself will come in the next post that I'm actively writing.

Before you start, you have to know the followings :

this technology is quite recent in the Java world. Spring's engeeniers work hard on it. Things you learn today may change.
this technology just begins to be used in enterprises. People mostly don't know anything about it.
this technology has a hard learning curve

You can find the app on my Github repo.

Why this app.

ReactivePipeline is a simple library of my own, built around the TaskPipeline project.

Its aim is simply to provide a clear single class, named ReactiveContext, with a few static methods providing all necessary objects to build a simple yet working reactive application. Under the hood, there's the Spring's Reactor project, a very little part to be honest.

Why did I do that ? For both the pleasure of learning reactive programming and share that little knowledge with the ones who are interested in it, the ones who would like to learn more but have not enough time, and maybe the ones who, just like I do, remain hungry after reading available tutorials.

If you have ever read some tutorials about reactive programming, maybe did you tell yourself that creating a Flux was not the most difficult part.

Take a simple example from Baeldung's post related to Reactor core :

With such code :

List<Integer> elements = new ArrayList<>();

Flux.just(1, 2, 3, 4)
  .log()
  .subscribe(elements::add);

assertThat(elements).containsExactly(1, 2, 3, 4);

you will produce this result in the console, thanks to the .log() statement :

20:25:19.550 [main] INFO  reactor.Flux.Array.1 - | onSubscribe([Synchronous Fuseable] FluxArray.ArraySubscription)
20:25:19.553 [main] INFO  reactor.Flux.Array.1 - | request(unbounded)
20:25:19.553 [main] INFO  reactor.Flux.Array.1 - | onNext(1)
20:25:19.553 [main] INFO  reactor.Flux.Array.1 - | onNext(2)
20:25:19.553 [main] INFO  reactor.Flux.Array.1 - | onNext(3)
20:25:19.553 [main] INFO  reactor.Flux.Array.1 - | onNext(4)
20:25:19.553 [main] INFO  reactor.Flux.Array.1 - | onComplete()

Of course, the .subscribe() statement does the job and will fill the elements collection with each received value. The assertion confirm that.

But... don't you find this very frustrating ?

As you can see, there's no difficulty in creating a Flux dispatching integer values before putting each of them in a collection.
But it is not sufficient, because for me the main question is :

How to maintain a process as a Flux from the beginning to the end of our processing operations ?

My main objective in writing this app, consequently, was to find out a simple way to keep all operations reactive, from the very first one to the very last one. Because writing a reactive app is just like a relay race : when a runner is about to end his tour, the next runner has already started to run so that there's no waiting time, no speed loss at the very moment of passing the baton.

This is why reactive programming seems a good alternative to imperative paradigm.

Imperative programming

Imperative programming is our everyday-way of making our job. We know it so much that we don't question it and accept its constraints for 2 main reasons :

it is normal to do as we learnt to, but the way we learnt is also induced by the procedural nature of most of the languages
it is the way our brain works : cause -> consequence in a single thread and sequentially

In an imperative programming style, we use to design operations that will be executed one after the other.

methodA(...) --> methodB(...) --> methodC(...)

When methodA(...) is called, it does its stuff and triggers the next method when ending. During this time, methodB(...) is waiting, and so is methodC(...).

Of course, we all know some workarounds like iterating on a collection and parallelizing calls to methodB(...) and then methodC(...). But we have to deal with threads, what is funny in fact but hard to debug in fact too.

Or we can of course exploit all the possibilities of Functional programming, like parallel-streaming collections, but code will quickly become unreadable with functions into functions into functions...

Think about a call to a repository that is returning many objects. You won't be able to process any of these results before the database server has returned its result set. It could be a request taking a long time, and nothing happens in the meanwhile.
At the opposite, reactive programming will send you objects from the repository each time one is found. While the database server is still processing your request, you can start processing what has already been found so far.

A reactive app can run faster not because it is intrinsically faster but just never waits.

Never waiting is what we want to design and build. We want pipes with
liquid data flowing inside. We want a Flux.

ReactivePipeline : disclaimer before we start using it

This is what the ReactivePipeline is about : create a persistent Flux across the whole application.

Does-it mean that there's no reactive code to write ?

No. I will talk to you about the Operation class that is a pure reactive object.

Does-it mean that you won't have to wire all operations in a
continuous flux by yourself ?

Yes. I will talk to you about the Task and Pipeline classes that are dedicated to wire operations together so easily.

Is it the perfect app for making things in a reactive way ?

Well, I guess that once you will understand how this modest project works, you will have all the necessary knowledge to use pure Spring Reactor classes. Just take it as a work-in-progress library, with known defaults, that can be improved in so many ways (I will transparently talk about too).

See you next time for practical things : Reactive Pipeline : the App

Lovegiver

Written with StackEdit.