How do you implement heartbeat-like functionality using Reactor's Flux?

Let's imagine there is a Flux emitting some payloads:

Flux<Integer> payloads = Flux.range(0, 5)
        .delayElements(Duration.ofSeconds(2));

This one just emits integers between 0 and 4, once per 2 seconds.

The task is to mix another signal into this Flux. That another signal just emits something every N units of time. Something like this:

Flux<Integer> heartbeats = Flux.just(-1)
        .repeat()
        .delayElements(Duration.ofSeconds(1));

1. The resulting Flux should output numbers from 0 to 4 once per 2 seconds, also number -1 each 1 second
2. Also, the resulting Flux must complete as soon as payloads Flux completes
3. It should propagate any error signal that any of the two Fluxes emit

(Please note that the example described above is just for illustration purposes. The real Flux emitting payloads may emit at inpredictable rate: sometimes fast, sometimes slowly, sometimes not emitting at all for long periods of time).

The best I could achieve with the 'out-of-the-box' operators was

Flux<Integer> flux = payloads.mergeWith(heartbeats);

But this violates requirement 2 as the merge result only completes when all the merge components complete, and the heartbeats does not ever complete.

(Actually, this could be made work if there was a flag that is set to true when the payloads completes, and repeat(BooleanSupplier) variant was used instead of repeat() on the heartbeats, but this would delay the completion of the resulting Flux till the moment of the next heartbeat.)

The next thing to try is to write my own Publisher implementation that would work in the way that is required for my task, but I'd like to avoid this as implementing a Publisher+Subscription correctly seems to be a tricky task.

Is there an easier way?

FluxSink API might be able to help to implement your requirements. You can access the API through Flux::create and use the FluxSink emitter to send the emitted signals downstream.

Here is a simple implementation in Kotlin:

val flux = Flux.create<Int> { emitter ->
    payloads.subscribe(
        { emitter.next(it) },
        { emitter.error(it) },
        { emitter.complete() },
    )

    heartbeats.subscribe(
        { emitter.next(it) },
        { emitter.error(it) },
    )
}

Please note, that this implementation turns payloads and heartbeats to hot Fluxes in a generic case.

Building on the answer of @zokni, here is what I was able to achieve using Flux.create():

Flux<Integer> flux = Flux.create(sink -> {
    AtomicReference<Subscription> payloadSubscription = new AtomicReference<>();
    AtomicReference<Subscription> heartbeatsSubscription = new AtomicReference<>();

    payloads.subscribe(
            t -> {
                sink.next(t);
                payloadSubscription.get().request(1);
            },
            sink::error,
            () -> {
                sink.complete();
                heartbeatsSubscription.get().cancel();
            },
            subscription -> {
                payloadSubscription.set(subscription);
                subscription.request(1);
            }
    );
    heartbeats.subscribe(sink::next, sink::error, () -> {}, subscription -> {
        heartbeatsSubscription.set(subscription);
        subscription.request(Long.MAX_VALUE);
    });

    sink.onCancel(() -> {
        payloadSubscription.get().cancel();
        heartbeatsSubscription.get().cancel();
    });
});

This solution is pretty verbose, but it has the following handy properties:

1. The completion of the payloads publisher makes the merged publisher complete as well
2. Both payloads's and heartbeats' subscriptions are cancelled on the cancellation of the merged publisher subscription
3. heartbeats' subscription is cancelled properly on payloads completion
4. Data is requested from payloads one by one element, instead of an unbounded demand
5. Repetitive subscription to the merged publisher (using Flux.repeat()) produce the expected results

I'm still not sure whether it's ok to ask for Long.MAX_VALUE from heartbeats, maybe it makes sense to do the same there: ask another element only after the previous was consumed, one by one.

Here is a solution with only out-of-the-box operators:

Flux<Integer> payloadsShared = payloads.publish().refCount(2);
Flux<Integer> flux = Flux.merge(payloadsShared,
        heartbeats.takeUntilOther(payloadsShared.ignoreElements()));

I'm using Flux.takeUntilOther(Publisher) to take elements from the heartbeat Flux until the payloads Flux emits an onComplete signal. Since takeUntilOther completes on the first received signal (onComplete or onNext), I drop the onNext signals with Flux.ignoreElements().

The original payloads Flux still has to be merged to this modified heartbeat. To allow the Flux to be processed by two Flux pipelines, I use Flux.publish().refCount(2).

"seems to be a tricky task" - yes it is tricky unless you use the right tool for this. The solution you need is an actor with one output port and ability to delay, taken from my library df4j:

import org.df4j.core.port.OutFlow;
import org.df4j.core.dataflow.Actor;

class DelayActor extends Actor {
    OutFlow<Integer> out = new OutFlow<>(this) ;
    int period = 1000; // ms
    int cnt = 0;
    int maxValue = 4;

    @Override
    protected void runAction() throws Throwable {
        out.onNext(-1);
        if (cnt % 2 == 0) {
            int value = cnt / 2;
            out.onNext(value);
            if (value == maxValue) {
                complete();
                out.onComplete();
                return;
            }
        }
        cnt++;
        delay(period);
    }
}

Port OutFlow out implements org.reactivestreams.Publisher and can accept all org.reactivestreams.Subscribers. If Flux is needed, it can be obtained as Flux.from(actor.out).

So, we'd like to subscribe same subscriber to two different publishers. For some reasons, it is prohibited by the specification of reactive streams. Luckily, my asynchronous library DF4J implements, among others, communication protocol which I named ReverseFlow, where multiple producers can feed the same consumer - just like many threads can write to the same BlockingQueue, but in asynchronous fashion.

import java.time.Duration;
import org.df4j.core.communicator.AsyncArrayBlockingQueue;
import org.df4j.core.dataflow.Actor;
import org.df4j.core.port.InpFlow;
import org.df4j.core.port.OutChannel;
import org.junit.Test;
import reactor.core.publisher.Flux;
/** converts reactive stream into ReverseFlow.Producer */
class Adapter<T> extends Actor {
public InpFlow<T> inp = new InpFlow<>(this);
public OutChannel<T> out = new OutChannel<>(this);
{start();}
@Override
protected void runAction() throws Throwable {
if (inp.isCompletedExceptionally()) {
out.onError(inp.getCompletionException());
} else if (inp.isCompleted()) {
out.onComplete();
} else {
out.onNext(inp.remove());
}
}
}
@Test
public void mergeTest() throws InterruptedException {
Flux<Integer> payloads = Flux.range(0, 5)
.delayElements(Duration.ofSeconds(2));
Adapter<Integer> adapter1 = new Adapter<>();
payloads.subscribe(adapter1.inp);
Flux<Integer> heartbeats = Flux.just(-1)
.repeat()
.delayElements(Duration.ofSeconds(1));
Adapter<Integer> adapter2 = new Adapter<>();
heartbeats.subscribe(adapter2.inp);
/* we use a queue with asynchronous interfaces on both ends */
AsyncArrayBlockingQueue<Integer> queue = new AsyncArrayBlockingQueue<>();
queue.feedFrom(adapter1.out);
queue.feedFrom(adapter2.out);
/* on the output end, AsyncArrayBlockingQueue is an ordinary Publisher */
Flux<Integer> merged = Flux.from(queue);
merged.subscribe(System.out::println);
Thread.sleep(13000);
}
<dependencies>
<dependency>
<groupId>org.df4j</groupId>
<artifactId>df4j-core</artifactId>
<version>8.3</version>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-core</artifactId>
<version>3.3.2.RELEASE</version>
</dependency>
</dependencies>