Fix to future cancellation

[mattrjacobs]

Right now, Observable.from(Future f) emits onError(CancellationException) when f starts out in a cancelled state. This change makes Observable.from(f) emit nothing at all.

[headinthebox]

Mmm, not sure if that is better. I think it makes more sense to call onError in that case. Why do you think being silent is better?

[cloudbees-pull-request-builder]

RxJava-pull-requests #1161 SUCCESS
This pull request looks good

[mattrjacobs]

My understanding of the Rx contract was that an outer unsubscribe should cause downstream emissions to stop whenever possible.

I encountered a case where Observable.zip(Observable.empty(), Observable.from(future), func) was emitting onError(CancellationException). Zip loops through the Observables and, on the onCompleted from the first Observable, unsubscribed from the 2nd Observable (thereby cancelling the underlying future). The onError then gets propagated out of the zip.

[headinthebox]

I think the question is more about the semantics of zip than of from(Future). Zip should stop whenever one of the sources completes. I must admit that the error behavior is underspecified.

In .NET (pseudo notation where ] is successful termination and > error) zipping one sequence that terminates normally with a sequence that terminates with an error will wait until the second sequence produces the next value before terminating:

  [1].zip([1,      >, (x,y) => (x,y))
=
 [(1,1),     >

whereas (using 18.3, for samples I stick to the released version, may be changed in master since there were a lot of changes to zip) this returns [(1,1)]. In 18.3 when you delay successful completion, zip waits for the successful termination before terminating:

  [1          ].zip([1,>, (x,y) => (x,y))
=
 [(1,1),    ]

So there is definitively a difference here, but in any case I'd separate that from from.

[abersnaze]

The CancelationException is not a real error. Its more of an onComplete because it is the only way the get() can stop without returning data.

[mattrjacobs]

I agree with @abersnaze. I don't find it intuitive that calling Observable.from on f (below) could ever emit onError.

Future<Integer> f = new FutureTask<Integer>(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                return 3;
            }
        });

[benjchristensen]

Agreed on this statement.

The catch could still result in an exception being thrown and unsubscribe having happened because it is a race. So we likely need a check inside the catch as well.

[benjchristensen]

An onError should never occur as a result of unsubscribe being called on an Observable.

[cloudbees-pull-request-builder]

RxJava-pull-requests #1177 SUCCESS
This pull request looks good

[benjchristensen]

Can you please rebase this so it can be merged? It conflicts now.

[benjchristensen]

Done in #1291