Difference Between Flux.create and Flux.generate
Create:
- Accepts a
Consumer<FluxSink<T>>
- Consumer is invoked only once per subscriber
- Consumer can emit 0..N elements immediately
- Publisher is not aware of downstream state. So we need to provide Overflow strategy as an additional parameter
- We can get the reference of FluxSink using which we could keep on emitting elements as and when required using multiple threads.
Generate:
- Accepts a
Consumer<SynchronousSink<T>>
- Consumer is invoked again and again based on the downstream demand
- Consumer can emit only one element at the max with an optional complete/error signal.
- Publisher produces elements based on the downstream demand
- We can get the reference of SynchronousSink. But it might not be really useful as we could emit only one element
Check this blog for more details.
In short:
Flux::create
doesn't react to changes in the state of the app whileFlux::generate
does.
The long version
Flux::create
You will use it when you want to calculate multiple (0...infinity) values which are not influenced by the state of your app and the state of your pipeline (your pipeline == the chain of operations which comes after Flux::create
== downstream).
Why? Because the method which you sent to Flux::create
keeps calculating elements (or none). The downstream will determine how many elements (elements == next signals) it wants and if he can't keep up, those elements which are already emitted will be removed/buffered in some strategy (by default they will be buffered until the downstream will ask for more).
The first and easiest use case is for emitting values which you, theoretically, could sum to a collection and only then take each element and do something with it:
Flux<String> articlesFlux = Flux.create((FluxSink<String> sink) -> {
/* get all the latest article from a server and emit them one by one to downstream. */
List<String> articals = getArticalsFromServer();
articals.forEach(sink::next);
});
As you can see, Flux.create
is used for interaction between blocking method (getArticalsFromServer
) to asynchronous code.
I'm sure there are other use cases for Flux.create
.
Flux::generate
Flux.generate((SynchronousSink<Integer> synchronousSink) -> {
synchronousSink.next(1);
})
.doOnNext(number -> System.out.println(number))
.doOnNext(number -> System.out.println(number + 4))
.subscribe();
The output will be 1 5 1 5 1 5................forever
In each invocation of the method you sent to Flux::generate
, synchronousSink
can only emits: onSubscribe onNext? (onError | onComplete)?
.
It means that Flux::generate
will calculate and emit values on demand. When should you use it? In cases where it's too expensive to calculate elements which may not be used downstream or the events which you emit are influenced by the state of the app or from the state of your pipeline (your pipeline == the chain of operations which comes after Flux::create
== downstream).
For example, if you are building a torrent application then you are receiving blocks of data in real time. You could use Flux::generate
to give tasks (blocks to download) to multiple threads and you will calculate the block you want to download inside Flux::generate
only when some thread is asking. So you will emit only blocks you don't have. The same algorithm with Flux::create
will fail because Flux::create
will emit all the blocks we don't have and if some blocks failed to be downloaded then we have a problem. because Flux::create
doesn't react to changes in the state of the app while Flux::generate
does.