Upgrading SyncOnSubscribe from experimental to beta

[stealthcode]

No description provided.

[stevegury]

👍

[stealthcode]

@akarnokd is this alright with you?

[akarnokd]

I'm not seeing why the change before 1.2?

[stealthcode]

I think at Netflix we are comfortable using these going forward. Is there any reason to wait? This only communicates our comfort level with this api slightly more strongly.

[akarnokd]

I see the value of SyncOnSubscribe but I'd like some user opinion about AsyncOnSubscribe outside Netflix.

[stealthcode]

Okay, fair. I would also like to see some usage and feedback on AsyncOnSubscribe as well. However I would prefer both of these apis to be in beta so people can reliably attempt to adopt them. I am comfortable (eventually) upgrading SyncOnSubscribe to part of the standard api but have Async in Beta. Is this fair?

[akarnokd]

Fine, but since this is a promotion, I'd like a majority vote just like before 1.1.

I personally would 👍 for SyncOnSubscribe promotion but I'm still not convinced about AsyncOnSubscribe's usability because the unpredictable request count if chained with any reasonable prefetching/replenishing operator (observeOn, flatMap, etc.).

[stealthcode]

Okay. I still would like it if we cleaned up the prefetching behavior so the request 1 patterns which we know are detrimental to performance are resolved.

[stealthcode]

@abersnaze @zsxwing @benjchristensen any concerns with this api upgrade from experimental to beta?

[akarnokd]

Could you open a separate issue with the problem description and example code so we can discuss it and not clutter this PR?

[stealthcode]

Do you mean the prefetching/batching behavior of observeOn and flatMap?

To confirm, you are in agreement that Experimental to Beta is acceptable for both?

[akarnokd]

I still would like it if we cleaned up the prefetching behavior so the request 1 patterns which we know are detrimental to performance are resolved.

[stealthcode]

Cool, Yes will do!

[stealthcode]

So to confirm you do not up vote this PR as is?

[akarnokd]

So to confirm you do not up vote this PR as is?

👍 for SyncOnSubscribe, 👎 for AsyncOnSubscribe. You'd better split the PR.

[kurzweil]

I've used AsyncOnSubcribe in a couple of places with success. One of which made integrating Elasticsearch's pagination api play nicely with Rx while allowing backpressure semantics to just work. Maybe one can use the following real code as an example of AsyncOnSubscribe usage:

import java.util.concurrent.TimeUnit;

import org.elasticsearch.action.ActionRequestBuilder;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchScrollRequestBuilder;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.unit.TimeValue;

import rx.Observable;
import rx.Observable.OnSubscribe;
import rx.observables.AsyncOnSubscribe;

public class ScrollObservable {
    private static class ElasticsearchRequestState {
        private final Client client;
        private final ActionRequestBuilder<?, SearchResponse, ?, ?> request;
        private final int pageNumber;
        private final long hitCount;
        private final TimeValue timeout;
        private ElasticsearchRequestState nextState;

        private ElasticsearchRequestState(Client client, SearchRequestBuilder request, TimeValue timeout) {
            this.client = client;
            this.request = request.setScroll(timeout);
            this.timeout = timeout;
            this.pageNumber = 0;
            this.hitCount = 0;
        }

        private ElasticsearchRequestState(Client client, SearchScrollRequestBuilder request, TimeValue timeout, int pageNumber, long hitCount) {
            this.client = client;
            this.request = request.setScroll(timeout);
            this.timeout = timeout;
            this.pageNumber = pageNumber;
            this.hitCount = hitCount;
        }

        public Observable<SearchResponse> getResponse() {
            return Observable.defer(() -> Observable.just(request.execute().actionGet()))
                    .retryWhen(errors -> errors.flatMap(error -> {
                        if (error instanceof InterruptedException || error.getCause() instanceof InterruptedException) {
                            return Observable.just(null);
                        } else {
                            return Observable.error(error);
                        }
                    }))
                    .doOnNext(r -> {
                        long currentCount = hitCount + r.getHits().getHits().length;
                        if (r.getHits().getHits().length > 0 && currentCount < r.getHits().getTotalHits()) {
                            nextState = new ElasticsearchRequestState(client, client.prepareSearchScroll(r.getScrollId()), timeout, pageNumber + 1, currentCount);
                        }
                    });
        }

        public boolean hasNext() {
            return nextState != null;
        }

        public ElasticsearchRequestState next() {
            return nextState;
        }
    }

    public static Observable<SearchResponse> from(final Client client, final SearchRequestBuilder request, final TimeValue timeout) {
        OnSubscribe<SearchResponse> os = AsyncOnSubscribe.createStateful(() ->
                new ElasticsearchRequestState(client, request, timeout),
                (state, requested, observer) -> {
                    observer.onNext(state.getResponse());
                    if (state.hasNext()) {
                        return state.next();
                    } else {
                        observer.onCompleted();
                        return null;
                    }
                });
        return Observable.create(os);
    }

    public static Observable<SearchResponse> from(Client client, SearchRequestBuilder request, long duration, TimeUnit timeUnit) {
        return from(client, request, new TimeValue(duration, timeUnit));
    }

    public static Observable<SearchResponse> from(Client client, SearchRequestBuilder request) {
        return from(client, request, new TimeValue(1, TimeUnit.MINUTES));
    }
}

[stealthcode]

@kurzweil interesting. It seems like you don't actually use the requested amount in producing an observable to return. I'd recommend changing state.getResponse() to take the long requested and limit the upper bounds of the data events.

[simonbasle]

@stealthcode It was some time ago, but last time I checked I couldn't find a detailed documentation on how to use these onSubscribe, other than the javadoc, after a quick search. Can you suggest a pointer to such resources if they exist? If they don't exist, could promoting these to BETA include some detailed piece of documentation, like a wiki page giving examples on how and when to use the AsyncOnSubscribe and SyncOnSubscribe?

[stealthcode]

@simonbasle Right now the documentation is pretty sparse on these topics. I would certainly welcome more documentation but don't currently have the time myself. @DavidMGross would you have some cycles to work on detailed documentation for the SyncOnSubscribe and AsyncOnSubscribe?

Here is one relevant SO post.

[simonbasle]

thanks @stealthcode

[JakeWharton]

To add some color here, there's currently no stable APIs for creating observables that do backpressure-aware work deferring. Even just for wrapping a synchronous method that returns a scalar value.

This make RxJava use in libraries very challenging since fromCallable and the helpers herein are unavailable.

[akarnokd]

What do you mean by

that do backpressure-aware work deferring

?

[JakeWharton]

A common pattern for deferring work is defer(() -> just(someMethod())), but if the subscriber has requested 0 the defer func is still called immediately on subscription, not on the first non-zero request. fromCallable doesn't suffer from this problem and the heavyweight hitters in this PR don't either, but all of those are non-stable APIs. As a workaround you can do things like empty().startWith(defer(() -> just(someMethod()))) or concat(just(defer(() -> just(someMethod())))) but that's a bit nuts.

[davidmoten]

@JakeWharton would be nice to support this strict deferral of work formally in the API and defer seems like a good place to put this support (so that subscription doesn't happen till first non-zero request). I'd be happy to see this happen in the existing defer operator but it could happen in an overload or elsewhere. Would you like to raise an issue to discuss this one?

[JakeWharton]

Sure!

On Thu, Mar 31, 2016 at 3:35 PM Dave Moten [email protected] wrote:

@JakeWharton https://github.com/JakeWharton would be nice to support
this strict deferral of work formally in the API and defer seems like a
good place to put this support (so that subscription doesn't happen till
first non-zero request). I'd be happy to see this happen in the existing
defer operator but it could happen in an overload or elsewhere. Would you
like to raise an issue to discuss this one?

—
You are receiving this because you were mentioned.

Reply to this email directly or view it on GitHub
#3780 (comment)

[stealthcode]

I have taken out AsyncOnSubscribe @akarnokd

[zsxwing]

👍

[akarnokd]

👍

[DavidMGross]

I'm currently in a holding pattern waiting to see where Netflix wants me to
apply my hours for them this season (they've been picking up the tab for my
documentation work thusfar), but I've filed a documentation issue so that I
or someone else can pick up the trail at some point: ReactiveX/
reactivex.github.io#221

On Mon, Mar 28, 2016 at 4:28 PM, Aaron Tull [email protected]
wrote:

@simonbasle https://github.com/simonbasle Right now the documentation
is pretty sparse on these topics. I would certainly welcome more
documentation but don't currently have the time myself. @DavidMGross
https://github.com/DavidMGross would you have some cycles to work on
detailed documentation for the SyncOnSubscribe and AsyncOnSubscribe?

Here is one SO post
http://stackoverflow.com/questions/32723315/is-unsubscribe-thread-safe-in-rxjava

—
You are receiving this because you were mentioned.
Reply to this email directly or view it on GitHub
#3780 (comment)

David M. Gross
PLP Consulting