Define semantics of subscribers erroring in onNext().

[ldaley]

I can't find this defined in the spec. Apologies if I've just missed it.

Equivalent in Rx semantics would be the Publisher must catch anything thrown by onNext() and terminate with an onError(). For processors, the upstream subscription would also have to be cancelled.

[rkuhn]

There was a lengthy discussion of this which led to rule 2.13, my understanding was that onNext should be treated the same way but that seems to have been lost in the editing (or I my memory is wrong). Unfortunately I cannot find the exact comment anymore, @viktorklang @benjchristensen ? (this is why I’d prefer all issues to be as targeted as this one ;-) )

[viktorklang]

I've addressed this in: #110

[smaldini]

IllegalStateException for onNext() ? I'm not a fan for wrapping up the exception in this specific case, it should just call onError with the source exception innit ?

[viktorklang]

@smaldini I think it is important to signal that it is not acceptable behavior, if we just pass the exception back nobody will ever know that it's not well-behaved spec-wise. Thoughts?

[benjchristensen]

I agree it should just pass the onError and not do any further wrapping. User errors can happen at any time in any user-provided function. Forcing them to all do try/catch within their functions is highly unwanted, and that's what this rule would effectively force to be "compliant".

[benjchristensen]

For history, related discussions exist in #52 (comment) and #68

Particularly:

If an onNext throws an Exception it is breaking the contract in Rx, but it can happen. Users can write buggy code or JVM-level errors can occur. Thus, an Rx implementation must account for "out of contract" behavior.

My stance on this discussion depends on the nuance of what is being restricted. If it is an operator implementation, then I'm okay with saying it shouldn't throw (except for the non-recoverable errors as discussed elsewhere). If user provided functions are trying to be prevented from throwing, then I disagree.

Consider this:

stream.map(t -> {
  ... do stuff ...
  return someComputation();
})

That user provided function should not need to be written like this:

stream.map(t -> {
try {
  ... do stuff ...
  return someComputation();
}catch(Exception e) {
  // now what do I do?
  return what???;
}
})

In particular, there is no mechanism for returning an Exception as there is in flatMap so the only legit approach is to throw an Exception which can then be propagated via onError by the map operator.

stream.map(t -> {
  ... do stuff that might fail ...
  return someComputation();
}).handleError(t -> {
  log(t);
  return defaultValue;
})

I also don't think we should force wrapping into an IllegalStateException even if an Operator implementation fails. If it fails it is a bug, the stacktrace will show it and it will be fixed. Forcing every single Operator to catch and wrap anything that fails underneath it is unnecessary.

[rkuhn]

@benjchristensen I think we’re all in full agreement here: it is the obligation of the map operator to wrap the user code in such a fashion that errors are propagated downstream (via onError) instead of upstream to the caller of onNext—which most of the time should be a non-issue as we should be dispatching asynchronously. The question raised here was not related to that user function, it was only related to the direct implementation of the code calling onNext, and that one should flag exceptions emanating from this method as illegal states, unless you know of an exception type that better captures this kind of spec violating behavior.

[viktorklang]

@rkuhn yes, this is exactly what I meant. Thanks for elaborating!

[benjchristensen]

Glad we agree on the user function failure scenario and its impact on an onNext.

it was only related to the direct implementation of the code calling onNext, and that one should flag exceptions emanating from this method as illegal states

Since operators/processors are generally going to be composed, wouldn't compliance with this rule mean that every processor would need to have a try/catch that wraps in as an IllegalStateException? Also due to composition, the code that catches the exception would not be aware of what caused it.

So I don't feel like I'm understanding what you want code to do.

Take this example:

@Override
public void onNext(T t) {
    try {
        o.onNext(userFunction.call(t));
    } catch (Throwable e) {
        onError(e);
    }
}

Here the try/catch is intended to capture problems from the userFunction, which we have agreed upon. But it can also incidentally catch errors from the onNext failing, but it won't wrap them in anything or distinguish between where it happened.

Are you saying you want all code that calls onNext to do something like this?

@Override
public void onNext(T t) {
    R r = null;
    try {
        r = transformer.call(t);
    } catch (Throwable e) {
        onError(e);
    }
    try {
        o.onNext(r);
    } catch(Throwable e) {
        onError(new IllegalStateException("Rule 2.13", e));
    }
}

[ldaley]

@benjchristensen my interpretation is that your latter example is the intention, and is most desirable in my opinion (with the additional provisor that the upstream subscription should also be cancelled).

[rkuhn]

Yes, the last sample is what I meant, and it will predominantly library code that implements RS in this fashion (though mostly not directly from onNext itself). There is no danger of double-wrapping since the error does not escalate up the call stack but forward in the Subscriber chain. One thing to consider is that an exception emanating from onError is covered by 2:14 which prescribes a mechanism “appropriate for the runtime environment”; if you choose to re-throw then wrapping is probably a bad idea but you are free to do what you deem appropriate.

[benjchristensen]

I don't know that I'm a fan of requiring every single processor/operator having to treat onNext as if it can blow up and then put in specific error messages.

If onNext blows up it is breaking the contract and will throw. If the processor/operator is synchronous (like map often is) then it will throw up to whoever is above it.

An async processor/operator will generally want to be overly cautious and wrap everything, but in the dozens of operators implementations of RxJava we only need a handful to try/catch around onNext - the async operators and some of the plumbing (such as lift and subscribe).

In other words, it feels like too strong a statement to require every single invocation of onNext to be wrapped in a try/catch. The point of the spec is to say that onNext doesn't throw, and therefore people shouldn't have to try/catch every time they use it.

[jbrisbin]

If we wanted to "force" implementors to handle an Exception emanating from onNext then the method should have been declared with throws Exception. As such, it seems unwise (and sometimes downright impractical) to impose try/catch around every onNext invocation.