Difference Between Flux.create and Flux.generate

1. Introduction

Project Reactor provides a fully non-blocking programming foundation for JVM. It offers implementation of Reactive Streams specification and provides composable asynchronous API  such as Flux. A Flux is a Reactive Streams publisher with several reactive operators. It emits 0 to N elements and then completes successfully or with error. It can be created in several different ways depending on our needs.

2. Understanding Flux

A Flux is a Reactive Stream publisher that can emit 0 to N elements. It has several operators which are used to generate, orchestrate, and transform Flux sequences. A Flux can complete successfully or complete with errors.

Flux API provides several static factory methods on Flux to create sources or generate from several callback types. It also provides instance methods and operators to build an asynchronous processing pipeline. This pipeline produces an asynchronous sequence.

In the next sections, let’s see a few usages of the Flux generate() and create() methods.

3. Maven Dependencies

We’ll need the reactor-core and reactor-test Maven dependencies:

<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>

4. Flux Generate

The generate() method of the Flux API provides a simple and straightforward programmatic approach to creating a Flux. The generate() method takes a generator function to produce a sequence of items.

There are three variants of the generate method:

• generate(Consumer<SynchronousSink<T>> generator)
• generate(Callable<S> stateSupplier, BiFunction<S, SynchronousSink<T>, S> generator)
• generate(Callable<S> stateSupplier, BiFunction<S, SynchronousSink<T>, S> generator, Consumer<? super S> stateConsumer)

The generate method calculates and emits the values on demand. It is preferred to use in cases where it is too expensive to calculate elements that may not be used downstream. It can also be used if the emitted events are influenced by the state of the application.

4.1. Example

In this example, let use the generate(Callable<S> stateSupplier, BiFunction<S, SynchronousSink<T>, S> generator) to generate a Flux:

public class CharacterGenerator {
    
    public Flux<Character> generateCharacters() {
        
        return Flux.generate(() -> 97, (state, sink) -> {
            char value = (char) state.intValue();
            sink.next(value);
            if (value == 'z') {
                sink.complete();
            }
            return state + 1;
        });
    }
}

In the generate() method, we are supplying two functions as arguments:

• The first one is a Callable function. This function defines the initial state for the generator with the value 97
• The second one is a BiFunction. This is a generator function that consumes a SynchronousSink. This SynchronousSink returns an item whenever the sink’s next method is invoked

Based on its name, a SynchronousSink instance works synchronously. However, we cannot call the SynchronousSink object’s next method more than once per generator call.

Let’s verify the generated sequence with StepVerifier:

@Test
public void whenGeneratingCharacters_thenCharactersAreProduced() {
    CharacterGenerator characterGenerator = new CharacterGenerator();
    Flux<Character> characterFlux = characterGenerator.generateCharacters().take(3);

    StepVerifier.create(characterFlux)
      .expectNext('a', 'b', 'c')
      .expectComplete()
      .verify();
}

In this example, the subscriber requests just three items. Hence the generated sequence ends by emitting three characters – a,b, and c. The expectNext() expects the elements we are expecting from the Flux. The expectComplete() indicates the completion of element emission from the Flux.

5. Flux Create

The create() method in Flux is used when we want to calculate multiple (0 to infinity) values that are not influenced by the application’s state. This is because the underlying method of the Flux create() method keeps calculating the elements.

Besides, the downstream system determines how many elements it needs. Therefore, if the downstream system is unable to keep up, already emitted elements are either buffered or removed.

By default, the emitted elements are buffered until the downstream system request more elements.

5.1. Example

Let us now demonstrate the example of the create() method:

public class CharacterCreator {
    public Consumer<List<Character>> consumer;

    public Flux<Character> createCharacterSequence() {
        return Flux.create(sink -> CharacterCreator.this.consumer = items -> items.forEach(sink::next));
    }
}

We can notice that the create operator asks us for a FluxSink instead of a SynchronousSink as used in the generate(). In this case, we’ll call next() for every item we have in the list of items, emitting each one by one.

Let us now use the CharacterCreator with two sequences of characters:

@Test
public void whenCreatingCharactersWithMultipleThreads_thenSequenceIsProducedAsynchronously() throws InterruptedException {
    CharacterGenerator characterGenerator = new CharacterGenerator();
    List<Character> sequence1 = characterGenerator.generateCharacters().take(3).collectList().block();
    List<Character> sequence2 = characterGenerator.generateCharacters().take(2).collectList().block();
}

We’ve created two sequences in the above code snippet – sequence1 and sequence2. These sequences serve as the source of character items. Note that we are using the CharacterGenerator instance to get the sequence of characters.

Let us now define an instance of characterCreator and two thread instances:

CharacterCreator characterCreator = new CharacterCreator();
Thread producerThread1 = new Thread(() -> characterCreator.consumer.accept(sequence1));
Thread producerThread2 = new Thread(() -> characterCreator.consumer.accept(sequence2));

We are now creating two thread instances that will provide elements to the publisher. The character element starts flowing into the sequence source when the accept operator is invoked. Next, we subscribe to the new consolidated sequence:

List<Character> consolidated = new ArrayList<>();
characterCreator.createCharacterSequence().subscribe(consolidated::add);

Notice that createCharacterSequence returns a Flux to which we subscribed and consumes the elements in the consolidated list. Next, let us trigger the whole process that sees items moving on two different threads:

producerThread1.start();
producerThread2.start();
producerThread1.join();
producerThread2.join();

Finally, let us verify the operation’s result:

assertThat(consolidated).containsExactlyInAnyOrder('a', 'b', 'c', 'a', 'b');

The first three characters in the received sequence come from sequence1. And the last two characters are from sequence2. Since this is an asynchronous operation, the order of elements from those sequences isn’t guaranteed.

6. Flux Create vs. Flux Generate

Following are a few differences between the create and generate operations:

7. Conclusion

In this article, we discussed the differences between the create and generate methods of Flux API.

First, we introduced the notion of reactive programming and talked about the Flux API. We then discussed the create and generate methods of Flux API. Finally, we provided a list of differences between the create and generate methods of the Flux API.