OperatorWindowWithStartEndObservable.java 文件源码

public Subscriber<? super T> call(Subscriber<? super Observable<T>> child) {
    CompositeSubscription csub = new CompositeSubscription();
    child.add(csub);
    final SourceSubscriber sub = new SourceSubscriber(child, csub);
    Subscriber<U> open = new Subscriber<U>() {
        public void onStart() {
            request(Long.MAX_VALUE);
        }

        public void onNext(U t) {
            sub.beginWindow(t);
        }

        public void onError(Throwable e) {
            sub.onError(e);
        }

        public void onCompleted() {
            sub.onCompleted();
        }
    };
    csub.add(sub);
    csub.add(open);
    this.windowOpenings.unsafeSubscribe(open);
    return sub;
}