Implemented the AsyncOnSubscribe

[stealthcode]

This includes the first functioning unit tests. This still has a few iterations to go but opening PR for review.

[akarnokd]

What's the purpose of this class?

[stealthcode]

This is the asynchronous complement to the SyncOnSubscribe as described in #3003.

[stealthcode]

The remaining issue with this implementation is that the inner observables are subscribed to an unbounded buffer. This could be improved to a back pressure supporting RxRingBuffer that requests from it's producer only to fill it's buffer capacity and then when it connects, drains the buffer and forwards all requests to its producer.

[akarnokd]

l still don't understand this class. Looks very similar to a SyncOnSubscribe that emits Observables and flatMaps it for you.

[stealthcode]

I understand your concern but I'm not sure about your comment. This operator does not call flatMap. This issue comes from use cases at Netflix with large services and streaming work. I think it's pretty clear to see from the unit tests that this gives a user direct access to the request and allows fulfillment by any observable (which may or may not be asynchronous).

[stealthcode]

This is very similar to the work that you tried with the AbstractProducer. It allows batching and eager subscription.

[stealthcode]

This is supposed to help with creating observables from asynchronous data sources in a non-blocking way. Consider the example below...

public static void main(String[] args) {
    Observable.<Integer>create((Subscriber<? super Integer> s) -> {
        s.setProducer((long requested) -> {
            getData(requested)
                .observe()
                .subscribe(s::onNext, s::onError, s::onCompleted);
        });
    });
}

public static HystrixCommand<Integer> getData(long requested) {
    return null;
}

public static interface HystrixCommand<T> {
    public Observable<T> observe();
}

The use case here is that for each request we want to issue an asynchronous service request and when that request is fulfilled then we want to start onNexting the data for that request. The problem with this naive implementation is that it's very easy and certainly possible to have overlapping requests. You could implement the cas addToRequestAndGet and make sure that only one thread at a time is fulfilling the request at a time but you still have the possibility of onNext events interleaving or overlapping.

[abersnaze]

it doesn't help with truly hot sources but it'll work great for progressive async sources. 👍

[benjchristensen]

I have reviewed this work with @stealthcode as he's been working on it and am quite interested in where this goes. It can greatly help us solve the most difficult part of using RxJava: creating correct sources that support backpressure.

I'll open an issue to summarize the needs for 1.x and 2.x so we can all get aligned on this topic.

I'm +1 on experimenting with this and eventually getting Observable.create overloads for all these helpers.

[benjchristensen]

Actually, there is already an issue: #3003

[akarnokd]

It would be great, if this is to be migrated to 2.x, that user-supplied functions are called in try-catch. Perhaps not here but in call().

[stealthcode]

It's merged into 1.x. Feel free to open a pull request to 2.x if you have time or to 1.x if you feel its necessary.

[akarnokd]

If nextIteration has terminated, the queue may still hold requested values so it might be worth considering calling clear() before quitting.

[stealthcode]

Good point. This would affect resubscriptions.

[davidmoten]

I'm not sure if this class could get used by multiple concurrent subscribers (?) but if it does then I would expect to see an atomic compareAndSet here rather than just !=null.