+str Add onErrorComplete operator.

[He-Pin]

#31671

@marcospereira @ncreep Would you like to give some inputs?

Have to duplicate some code, as the ambiguous reference to overloaded definition issue.

Still need to add some tests.

[ncreep]

Hi @He-Pin,

Thanks for the effort, but this feature doesn't really address my original request. One of my requirements is that I didn't want to catch exceptions blindly, but rather avoid throwing them in the first place (as idle-timeout) does.

[He-Pin]

I think we can add a akka's timeout exception,which extends the current TimeoutExpection？

[He-Pin]

@ncreep I want to compose #31640 with this pr, what do you think about it?

[ncreep]

#31640 does resolve my issue with the exception (if you add the recoverWithComplete close enough to the idle component). But I'd rather avoid using exceptions for flow control. Seeing how in my use-case having the timeout is not at all "exceptional", I think it would be nicer to have a non-exceptional way to create completion.

Having said, I think the recoverWithComplete is generally useful, and I don't have any other objections.

[He-Pin]

@ncreep Yes, I think the current idleTimeout can accept an alternative Source when idle. if the alternative source is not provided, then just fail the source.
then idleTimeout(timeout, Source.empty) will address your origin issue.

[ncreep]

@He-Pin, that does sound good. But I wonder whether there's something more general that can be done here. Some implementation that won't be specific to just the idleTimeout use-case. I guess that thinking about this is what got the original issue side-tracked...

[johanandren]

I like this, together with #31640 it nicely covers the various timeout scenarios.

I wonder if we should skip the no-param variation though, handing it Throwable in user code seems concise enough for the catch-all IMO.

Also not sure about the name, should it rather be something like completeOn[IdleTimeoutException]? Recover would be to turn the stream back into something working, but this operator always completes.

[He-Pin]

Cool, I was following the old name, your completeOn is definitely a better name, the name in reactor-core is onErrorComplete.

[He-Pin]

@ncreep I was thinking about that too, but never came out a better idea. Will try to get this merged first:) do you have something in mind?

[He-Pin]

Still on holiday, will try to find more time to get it ready.

[ncreep]

@ncreep I was thinking about that too, but never came out a better idea. Will try to get this merged first:) do you have something in mind?

@He-Pin, no, unfortunately I'm not proficient enough in the underlying mechanics of Akka streams to have something concrete in mind.

But I can imagine that whatever the general solution would be it would have to include some change to the underlying signalling mechanism. So that you can "catch" completion signals and then convert those into exceptions. So on its own, idleTimeout would send a completion signal, and then if you want the current behavior back you wrap that signal and throw a Timeout exception for it. That would require some way to identify the origin of the completion signal, so that you don't accidentally catch signals that you didn't mean to. Though maybe something more local would be better, so that every component would receive an onComplete handler, which can decide what to do with the completion of the underlying component (all this assumes that the direction of complete -> exception is more "natural" than exception -> complete).

I'm probably reinventing the wheel here, so I'll stop babbling...

[He-Pin]

@patriknw Any suggestion? I want to allocate one day to continue work on this.

[johanandren]

LGTM, lets go with the empty param list like this 👍

[He-Pin]

@patriknw I have rebased on to the current main

[patriknw]

I know the naming has been discussed, but this name doesn't include much about that it is about errors. Might be difficult to discover. onErrorComplete is better. I still like recoverWithComplete because it's close to what we already have, but seems like @johanandren didn't like it.

[He-Pin]

There are :

• onErrorComplete(..)
• onErrorContinue(...)
• onErrorMap(...)
• onErrorReturn(...)
• onErrorResume(...)
• onErrorStop(...)

in reactor-core. and should we use completeOn , completeWhen, or recoverWithComplete here? naming is always hard.

[He-Pin]

@octonato What's your input? the recoverWithComplete align better but seems like completeWhen is a great name too.

[octonato]

I agree that onErrorComplete is better for discoverability and it express much better that the method has the intention to be used in case of an error.

About recoverWithComplete, it is less interesting, I think. Internally, we are 'recovering' with empty Source, but that's an implementation detail.

From a user perspective, I don't think recover makes sense because we are not recovering (and continuing) when an error occurs. We are completing gracefully.

That said, maybe a overloaded onErrorComplete (Predicated and PF) will do the job. Clear intention and discoverable.

[He-Pin]

@johanandren I updated the PR to onErrorComplete and the https://github.com/akka/akka-persistence-r2dbc is using Flux too.

And as we here we only matching on error types, and completeOn/completeWhen is more generic.

[johanandren]

Let's go with onErrorComplete

[octonato]

LGTM