AsyncOnSubscribe.java 文件源码

@Experimental
public static <S, T> Observable$OnSubscribe<T> createSingleState(Func0<? extends S> generator, final Action3<? super S, Long, ? super Observer<Observable<? extends T>>> next, Action1<? super S> onUnsubscribe) {
    return new AsyncOnSubscribeImpl(generator, new Func3<S, Long, Observer<Observable<? extends T>>, S>() {
        public S call(S state, Long requested, Observer<Observable<? extends T>> subscriber) {
            next.call(state, requested, subscriber);
            return state;
        }
    }, onUnsubscribe);
}