Ability to signal error upstream (nack) without cancelling/completing about reactor-core HOT 7 CLOSED

How about JMS.queue() returns something that let's you communicate back an acknowledgement/rejection?

JMS.queue("in").concatMap(manualMsg -> {
    try {
         Flux<R> result = Flux.just(processMessage(manualMsg));
         manualMsg.acknowledge();
         return result;
    } catch (Throwable ex) {
        manualMsg.reject();
        Exceptions.throwIfFatal(ex);
        return Flux.error(ex);
    }
});

from reactor-core.

@akarnokd That's option 3). But it's not quite so simple I don't think:

• You wouldn't want to return Flux.error(ex) as that is terminal. Not sure what you would return instead though?
• It's ok to nack/reject here, but IMO this is the wrong place to ack. Given you maybe have further processing steps which may fail. e.g. `JMS.queue("in").map(msg -> processMsg(msg)).map(msg -> convertMsg(msg)).consume(System.out::println). You only really want to ack at the end. You'd have to put a subscriber at the end to do it yourself.

This approach is doable, but ideally there would be some spec/framework support for this, if it makes sense and is common.

from reactor-core.

The problem is you have a per-value lifecycle and Reactive-Streams can't deal with it. The same issue appears with pooled ByteBuffers in I/O libraries where someone has to release the buffer once all stages have processed it and goes "out of scope".

The closest thing is what I call umap in Reactive-IO where you provide a transform for some upstream sequence and all participating object's lifecycle ends when the transformation terminates. In this context, you'd have:

Flux<Void> JMS.queue(String topic, Function<Mono<Message>, Publisher<?>> umap);

(Note, however, that fatal exceptions break this too.)

from reactor-core.

That's the kind of pattern we try to implement from https://github.com/reactor/reactor-io/blob/master/reactor-ipc/src/main/java/reactor/io/ipc/Outbound.java#L44 where Netty (or Aeron) might receive a notification of failure. But implementing such contract in memory should be fairly easy.

from reactor-core.

@dfeist should we consider the latest Mono#create addition enough for now and close this issue ?

from reactor-core.

@smaldini No, this is completely unrelated to Mono#create. The solution to this issue is take a different approach to this scenario and not think of sources (at least one that need to know how stream terminates) as Publishers, but as an something that creates a stream and then is passed and applies a transformation to it. (As explained by @akarnokd )

The only thing potentially outstanding here is the discussion we had with @sskrla around an alternative way to able to handle errors without i) requiring a Tuple everywhere, and having skip certain processing steps if error is present ii) causing a terminal error which requires re-subscription. This might be simply not possible though, given it conflicts with other things that are in the spec. Is this already be captured in #49 ?

from reactor-core.

@smaldini We can close this out, this was an issue with the way i was attempting to use stream for request/response.

from reactor-core.