Programmatically Creating Sequences with Project Reactor

1. Overview

In this tutorial, we’ll use Project Reactor basics to learn a few techniques for creating Fluxes.

2. Maven Dependencies

Let’s get started with a couple of dependencies. We’ll need reactor-core and reactor-test:

<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-core</artifactId>
    <version>3.6.0</version>
</dependency>
<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-test</artifactId>
    <version>3.6.0</version>
    <scope>test</scope>
</dependency>

3. Synchronous Emission

The simplest way to create a Flux is Flux#generate. This method relies on a generator function to produce a sequence of items.

But first, let’s define a class to hold our methods illustrating the generate method:

public class SequenceGenerator {
    // methods that will follow
}

3.1. Generator With New States

Let’s see how we can generate the Fibonacci sequence with Reactor:

public Flux<Integer> generateFibonacciWithTuples() {
    return Flux.generate(
            () -> Tuples.of(0, 1),
            (state, sink) -> {
                sink.next(state.getT1());
                return Tuples.of(state.getT2(), state.getT1() + state.getT2());
            }
    );
}

It’s not hard to see this generate method takes two functions as its arguments – a Callable and a BiFunction:

• The Callable function sets up the initial state for the generator – in this case, it’s a Tuples with elements 0 and 1
• The BiFuntion function is a generator, consuming a SynchronousSink, then emitting an item in each round with the sink’s next method and the current state

As its name suggests, a SynchronousSink object works synchronously. However, notice that we cannot call this object’s next method more than once per generator calling.

Let’s verify the generated sequence with StepVerifier:

@Test
public void whenGeneratingNumbersWithTuplesState_thenFibonacciSequenceIsProduced() {
    SequenceGenerator sequenceGenerator = new SequenceGenerator();
    Flux<Integer> fibonacciFlux = sequenceGenerator.generateFibonacciWithTuples().take(5);

    StepVerifier.create(fibonacciFlux)
      .expectNext(0, 1, 1, 2, 3)
      .expectComplete()
      .verify();
}

In this example, the subscriber requests just five items, hence the generated sequence ends with number 3.

As we can see, the generator returns a new state object to be used in the next pass. It’s not necessary to do so, though. We can reuse a state instance for all the invocations of the generator instead.

3.2. Generator With Mutable State

Suppose we want to generate the Fibonacci sequence with a recycled state. To demonstrate this use case, let’s first define a class:

public class FibonacciState {
    private int former;
    private int latter;

    // constructor, getters and setters
}

We’ll use an instance of this class to hold the generator’s state. The two properties of this instance, former and latter, store two consecutive numbers in the sequence.

If we modify our initial example, we’ll now be using mutable state with generate:

public Flux<Integer> generateFibonacciWithCustomClass(int limit) {
    return Flux.generate(
      () -> new FibonacciState(0, 1),
      (state, sink) -> {
        sink.next(state.getFormer());
        if (state.getLatter() > limit) {
            sink.complete();
        }
        int temp = state.getFormer();
        state.setFormer(state.getLatter());
        state.setLatter(temp + state.getLatter());
        return state;
    });
}

Similar to the previous example, this generate variant has state supplier and generator parameters.

The state supplier of type Callable simply creates a FibonacciState object with the initial properties of 0 and 1. This state object will be reused throughout the lifecycle of the generator.

Just like the SynchronousSink in the Fibonacci-with-Tuples example, the sink over here produces items one by one. However, unlike that example, the generator returns the same state object each time it’s called.

Also notice this time, to avoid an infinite sequence, we instruct the sink to complete when the produced value reaches a limit.

And, let’s again do a quick test to confirm that it works:

@Test
public void whenGeneratingNumbersWithCustomClass_thenFibonacciSequenceIsProduced() {
    SequenceGenerator sequenceGenerator = new SequenceGenerator();

    StepVerifier.create(sequenceGenerator.generateFibonacciWithCustomClass(10))
      .expectNext(0, 1, 1, 2, 3, 5, 8)
      .expectComplete()
      .verify();
}

3.3. Stateless Variant

The generate method has another variant with only one parameter of type Consumer<SynchronousSink>. This variant is only suitable to produce a pre-determined sequence, hence not as powerful. We won’t cover it in detail, then.

4. Asynchronous Emission

Synchronous emission isn’t the only solution to the programmatic creation of a Flux.

Instead, we can use the create and push operators to produce multiple items in a round of emission in an asynchronous manner.

4.1. The create Method

Using the create method, we can produce items from multiple threads. In this example, we’ll collect elements from two different sources into a sequence.

First, let’s see how create is a little different from generate:

public class SequenceCreator {
    public Consumer<List<Integer>> consumer;

    public Flux<Integer> createNumberSequence() {
        return Flux.create(sink -> SequenceCreator.this.consumer = items -> items.forEach(sink::next));
    }
}

Unlike the generate operator, the create method doesn’t maintain a state. And rather than generating items by itself, the emitter passed to this method receives elements from an external source.

Also, we can see that the create operator asks us for a FluxSink instead of a SynchronousSink. With a FluxSink, we can call next() as many times as we need to.

In our case, we’ll call next() for every item we have in the list of items, emitting each one by one. We’ll see how to populate items in just a moment.

Our external source, in this case, is an imaginary consumer field, though this instead could be some observable API.

Let’s put the create operator into action, starting with two sequences of numbers:

@Test
public void whenCreatingNumbers_thenSequenceIsProducedAsynchronously() throws InterruptedException {
    SequenceGenerator sequenceGenerator = new SequenceGenerator();
    List<Integer> sequence1 = sequenceGenerator.generateFibonacciWithTuples().take(3).collectList().block();
    List<Integer> sequence2 = sequenceGenerator.generateFibonacciWithTuples().take(4).collectList().block();

    // other statements described below
}

These sequences, sequence1 and sequence2, will be serving as the sources of items for the generated sequence.

Next, comes two Thread objects that will pour elements into the publisher:

SequenceCreator sequenceCreator = new SequenceCreator();
Thread producingThread1 = new Thread(
  () -> sequenceCreator.consumer.accept(sequence1)
);
Thread producingThread2 = new Thread(
  () -> sequenceCreator.consumer.accept(sequence2)
);

When the accept operator is called, elements start flowing into the sequence source.

And then, we can listen, or subscribe, to our new, consolidated sequence:

List<Integer> consolidated = new ArrayList<>();
sequenceCreator.createNumberSequence().subscribe(consolidated::add);

By subscribing to our sequence, we indicate what should happen with each item emitted by the sequence. Here, it’s to add each item from disparate sources to a consolidated list.

Now, we trigger the whole process that sees items moving on two different threads:

producingThread1.start();
producingThread2.start();
producingThread1.join();
producingThread2.join();

As usual, the last step is to verify the operation’s result:

assertThat(consolidated).containsExactlyInAnyOrder(0, 1, 1, 0, 1, 1, 2);

The first three numbers in the received sequence come from sequence1, while the last four from sequence2. Due to the nature of asynchronous operations, the order of elements from those sequences isn’t guaranteed.

The create method has another variant, taking an argument of type OverflowStrategy. As its name implies, this argument manages back-pressure when the downstream can’t keep up with the publisher. By default, the publisher buffers all elements in such a case.

4.2. The push Method

In addition to the create operator, the Flux class has another static method to emit a sequence asynchronously, namely push. This method works just like create, except that it allows only one producing thread to emit signals at a time.

We could replace the create method in the example we’ve just gone through with push, and the code will still compile

However, sometimes we would see an assertion error, as the push operator keeps FluxSink#next from being called concurrently on different threads. As a result, we should use push only if we don’t intend to use multiple threads.

5. Handling Sequences

All the methods we’ve seen so far are static and allow the creation of a sequence from a given source. The Flux API also provides an instance method, named handle, for handling a sequence produced by a publisher.

This handle operator takes on a sequence, doing some processing and possibly removing some elements. In this regard, we can say the handle operator works just like a map and a filter.

Let’s take a look at a simple illustration of the handle method:

public class SequenceHandler {
    public Flux<Integer> handleIntegerSequence(Flux<Integer> sequence) {
        return sequence.handle((number, sink) -> {
            if (number % 2 == 0) {
                sink.next(number / 2);
            }
        });
    }
}

In this example, the handle operator takes a sequence of numbers, dividing the value by 2 if even. In case the value is an odd number, the operator doesn’t do anything, meaning that such a number is ignored.

Another thing to notice is that, as with the generate method, handle employs a SynchronousSink and enables one-by-one emissions only.

And finally, we need to test things. Let’s use StepVerifier one last time to confirm that our handler works:

@Test
public void whenHandlingNumbers_thenSequenceIsMappedAndFiltered() {
    SequenceHandler sequenceHandler = new SequenceHandler();
    SequenceGenerator sequenceGenerator = new SequenceGenerator();
    Flux<Integer> sequence = sequenceGenerator.generateFibonacciWithTuples().take(10);

    StepVerifier.create(sequenceHandler.handleIntegerSequence(sequence))
      .expectNext(0, 1, 4, 17)
      .expectComplete()
      .verify();
}

There are four even numbers among the first 10 items in the Fibonacci sequence: 0, 2, 8, and 34, hence the arguments we pass to the expectNext method.

6. Conclusion

In this article, we walked through various methods of the Flux API that can be used to produce a sequence in a programmatic way, notably the generate and create operators.