Add version of Observable.just(item) that supports backpressure

[davidmoten]

When I need an Observable for a single value that supports backpressure I normally use Observable.from(Arrays.asList(item)). This carries a bit of overhead and @akarnokd suggested this implementation:

Observable.create(new OnSubscribe<T>() {

            @Override
            public void call(Subscriber<? super T> subscriber) {
                subscriber.setProducer(new SingleProducer<T>(subscriber, defaultValue));
            }});

I'd like to see this in the API and if I assume that the current Observable.just(item) stays as is for performance reasons then what would be a good method name/overload?

Perhaps Observable.just(item, enableBackpressure)? Wont' work because clashes with just(item1, item2).

justWithBackpressure(item) ?

[benjchristensen]

I don't like the idea of specialized "just" implementations like this.

I understand the theoretical purity of this, but I'm curious as to what use case actually can't handle an emission of 1 item. Why is the consumer not ready for the single value when it subscribes?

[akarnokd]

I can't think of an operator that would subscribe to a source but not request anything.

[davidmoten]

Lots of operators can subscribe to a source and not request anything (say map) if downstream has called request(0). We have the option in a Subscriber in onStart to request 0 and then at our leisure after the subscribed stream has been setup to make non-zero requests. One emission can induce a lot of work downstream and a system may want to control when that load happens. The only reliable strategy for the moment is to not subscribe till ready for first emission (perhaps that's enough?).

This discussion makes me wonder about the eagerness of many of our Observable sources and operators. For a bit more control should our operators only request non-zero from upstream once downstream has requested non-zero (for example merge, takeLast)? This would be a major change of course because so many operators and sources would be affected if we thought it desirable.

[akarnokd]

I experimented with a barebone reactive-streams just() implementation and it had 20% less throughput due to the mandatory CAS in the ScalarSubscription.

There is an alternative without CAS that works with synchronous requesters:

final class ScalarLazyProducer extends AtomicBoolean implements Producer {
    @Override
    public void request(long n) {
        if (n > 0) {
            lazySet(true);
        }
    }
}

@Override
public void subscribe(Subscriber<? super T> child) {
    ScalarLazyProducer slp = new ScalarLazyProducer();
    child.setProducer(slp);
    if (slp.get()) {
        child.onNext(value);
        child.onCompleted();
    } else {
        child.onError(new MissingBackpressureException());
    }
}

I haven't benchmarked this solution but I'd expect 5% less throughput than the 1.x version.

[akarnokd]

Do you want to pursue this further with 1.x?

[stevegury]

I agree with @davidmoten here. Even if it's rarely used, for the sake of correctness, we have to respect request(0) properly.
I think that the 5% performance degradation is acceptable.

[akarnokd]

Note also that the scheduleOn optimization (i.e., that using subscribeOn or observeOn on a scalar) is no longer valid. In addition, the scalarFlatMap optimization can no longer unwrap and emit a scalar value. See https://github.com/ReactiveX/RxJava/blob/2.x/src/main/java/io/reactivex/internal/operators/ObservableScalarSource.java for guidance.

[davidmoten]

@benjchristensen

I understand the theoretical purity of this, but I'm curious as to what use case actually can't handle an emission of 1 item. Why is the consumer not ready for the single value when it subscribes?

For a use case see #3466 (combination of switchMap and replay).

[davidmoten]

And now another scenario where just can cause backpressure failure #3494.

[akarnokd]

Fix posted: #3496.