1

Reactive programming newbie. I have a sequence of calls in my Flux and I need to ensure they're done in order. Eg

Flux<Thing> flux = ...
.doOnNext(this::sendThing)
.doOnNext(this::persistThing)
.doOnError(error -> log.error("", error))
.blockLast();

I need to ensure that sendThing completes before persistThing. I'm unclear, being a reactive newbie, if this is guaranteed.

2 Answers 2

1

According to documentation:

The Consumer is executed first, then the onNext signal is propagated downstream.

Thus, this::sendThing consumer will be completed before the onNext signal be propagated to this::persistThing.

Example:

    Flux.range(1, 3)
        .doOnNext(n -> log.info("doOnNext1"))
        .log()
        .doOnNext(n -> log.info("doOnNext2"))
        .blockLast();

The above snippet prints the following:

  • doOnNext1
  • | onNext(1)
  • doOnNext2
  • doOnNext1
  • | onNext(2)
  • doOnNext2
  • doOnNext1
  • | onNext(3)
  • doOnNext2

onNext signal is always propagated before "doOnNext2" is printed.

Bear in mind that doOn methods should be used only for side-effect operations(e.g. log).

1

Not familiar with this reactive implementation, but looks like you're adding two independent subscribers to a single observable, which does not guarantee an order of execution.

If send and persist are blocking (sync) functions you can just create some 'sendAndSync' function.

Otherwise, you need to make the 'send' into and observable itself, and 'persist' should be its subscriber. You can achieve this by 'send' writing to a PublishSubject once a send is completed, and 'persist' being the subscriber (via doOnNext) of that PublishSubject.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.