java类rx.functions.FuncN的实例源码

public Zip(Subscriber<? super R> child, FuncN<? extends R> zipFunction) {
    this.child = child;
    this.zipFunction = zipFunction;
    child.add(this.childSubscription);
}

public OperatorZip(FuncN<? extends R> f) {
    this.zipFunction = f;
}

public static final <T, R> Observable<R> combineLatest(List<? extends Observable<? extends T>> sources, FuncN<? extends R> combineFunction) {
    return create(new OnSubscribeCombineLatest(sources, combineFunction));
}

public static final <R> Observable<R> zip(Observable<? extends Observable<?>> ws, FuncN<? extends R> zipFunction) {
    return ws.toList().map(new 4()).lift(new OperatorZip(zipFunction));
}

private static final <T, R> Observable<R> combineSinglesDelayError(List<? extends Observable<? extends T>> sources, FuncN<? extends R> combineFunction) {
    return combineLatestDelayError(sources, combineFunction);
}

private Observable example3() {
    Observable<String> observable1 = Observable.create(new Observable.OnSubscribe<String>() {
        @Override
        public void call(final Subscriber<? super String> subscriber) {
            final Scheduler.Worker worker = Schedulers.io().createWorker();
            subscriber.add(worker);
            worker.schedulePeriodically(new Action0() {
                long counter = 0;

                @Override
                public void call() {
                    if (counter == 8) {
                        subscriber.onError(new Throwable());
                    }
                    try {
                        subscriber.onNext("Observable1 " + counter++ * 10);
                    } catch (Throwable e) {
                        try {
                            worker.unsubscribe();
                        } finally {
                            Exceptions.throwOrReport(e, subscriber);
                        }
                    }
                }
            }, 1000, 1000, TimeUnit.MILLISECONDS);
        }
    });
    Observable<String> observable2 = Observable.interval(1, TimeUnit.SECONDS)
            .map(new Func1<Long, String>() {
                @Override
                public String call(Long position) {
                    return "observable2 " + position * 10;
                }
            }).take(10);
    List<Observable<String>> observableList = new ArrayList<>();
    observableList.add(observable1);
    observableList.add(observable2);

    return Observable.combineLatestDelayError(observableList, new FuncN<Object>() {
        @Override
        public Object call(Object... args) {
            List<String> results = new ArrayList<>();
            for (Object object : args) {
                results.add((String) object);
            }
            return results;
        }
    });
}

public SingleSourceProducer(final Subscriber<? super R> child, Observable<? extends T> source, FuncN<? extends R> combinator) {
    this.source = source;
    this.child = child;
    this.combinator = combinator;
    this.subscriber = new SingleSourceRequestableSubscriber<T, R>(child, combinator);
}

SingleSourceRequestableSubscriber(Subscriber<? super R> child, FuncN<? extends R> combinator) {
    super(child);
    this.child = child;
    this.combinator = combinator;
}

public OperatorZip(FuncN<? extends R> f) {
    this.zipFunction = f;
}

public Zip(final Subscriber<? super R> child, FuncN<? extends R> zipFunction) {
    this.child = child;
    this.zipFunction = zipFunction;
    child.add(childSubscription);
}