Hello Developer, Hope you guys are doing great. Today at Tutorial Guruji Official website, we are sharing the answer of Wait given (reset-able) time for subscribers to subscribe without wasting too much if your time.
The question is published on by Tutorial Guruji team.
The question is published on by Tutorial Guruji team.
In reactor there is a way of waiting for a specific number of subscribers and only then the publisher should start emitting.
I would like to implement a slightly different scenario:
- subscriber subscribes
- publisher awaits another subscription
- subscription comes in less than expected time t
- timer resets and publisher continue to wait for another subscriber
- no subscription in a given time t
- publisher starts to emmit
If the cycle of awaits lasts for too long (longer than time t2) than publisher starts emitting regardless.
How to do it?
Answer
There’s no baked-in operator to do that but you should be able to emulate that with a scheduled task and an AtomicInteger, using the standard publish()–connect() cycle:
@Test
public void deferredConnect() throws InterruptedException {
ConnectableFlux<Integer> connectableFlux = Flux.range(1, 10)
.publish();
AtomicInteger subCount = new AtomicInteger();
Flux<Integer> deferredConnect = connectableFlux
.doOnSubscribe(sub -> {
int current = subCount.incrementAndGet();
Schedulers.parallel().schedule(() -> {
if (subCount.compareAndSet(current, -1)) {
connectableFlux.connect();
}
}, 1, TimeUnit.SECONDS);
});
deferredConnect.subscribe(v -> System.out.println("1: " + v));
Thread.sleep(500);
deferredConnect.subscribe(v -> System.out.println("2: " + v));
Thread.sleep(400);
deferredConnect.subscribe(v -> System.out.println("3: " + v));
Thread.sleep(200);
assertThat(subCount).hasNonNegativeValue();
Thread.sleep(800);
assertThat(subCount).hasNegativeValue();
}
We are here to answer your question about Wait given (reset-able) time for subscribers to subscribe - If you find the proper solution, please don't forgot to share this with your team members.