import rx.functions.Functions; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Test
public void testCollectionSizeDifferentThanFunction() {
    FuncN<String> zipr = Functions.fromFunc(getConcatStringIntegerIntArrayZipr());
    //Func3<String, Integer, int[], String>

    /* define a Observer to receive aggregated events */
    Observer<String> observer = mock(Observer.class);

    @SuppressWarnings("rawtypes")
    Collection ws = java.util.Collections.singleton(Observable.just("one", "two"));
    Observable<String> w = Observable.zip(ws, zipr);
    w.subscribe(observer);

    verify(observer, times(1)).onError(any(Throwable.class));
    verify(observer, never()).onComplete();
    verify(observer, never()).onNext(any(String.class));
}
 

import rx.functions.Functions; //导入依赖的package包/类
private Observable<Boolean> directPoster(Observable<List<Sample>> samples, MetricRegistry metrics) {

        final SampleRepository repository = repository();
        final Timer timer = metrics.timer("writes");
        final Meter completions = metrics.meter("samples-completed");
        

        Func1<List<Sample>, Boolean> insert = new Func1<List<Sample>, Boolean>() {

            @Override
            public Boolean call(List<Sample> s) {
                int sz = s.size();
                try (Context timerCtx = timer.time()) {
                    repository.insert(s);
                    return true;
                } finally {
                    completions.mark(sz);
                }
            }
        };
        
        
        return (m_threadCount == 1 ? samples.map(insert) : parMap(samples, metrics, insert)).all(Functions.<Boolean>identity());
        
        
    }
 

import rx.functions.Functions; //导入依赖的package包/类
public Observable<String> getHeader(String key) {
    return map
    // select MethodLine line
            .filter(byKey("MethodLine"))
            // flatten
            .flatMap(Functions.<Observable<Header>> identity())
            // to value
            .map(toValue)
            // cache result
            .cache();
}
 

import rx.functions.Functions; //导入依赖的package包/类
/**
 * Tests whether two {@code Observable} sequences are identical, emitting {@code true} if both sequences
 * complete without differing, and {@code false} if the two sequences diverge at any point.
 *
 * @param first
 *      the first of the two {@code Observable}s to compare
 * @param second
 *      the second of the two {@code Observable}s to compare
 * @param equality
 *      a function that tests emissions from each {@code Observable} for equality
 * @return an {@code Observable} that emits {@code true} if {@code first} and {@code second} complete
 *         after emitting equal sequences of items, {@code false} if at any point in their sequences the
 *         two {@code Observable}s emit a non-equal item.
 */
public static <T> Observable<Boolean> sequenceEqual(
        Observable<? extends T> first, Observable<? extends T> second,
        final Func2<? super T, ? super T, Boolean> equality) {
    Observable<Object> firstObservable = materializeLite(first);
    Observable<Object> secondObservable = materializeLite(second);

    return zip(firstObservable, secondObservable,
            new Func2<Object, Object, Boolean>() {

                @Override
                @SuppressWarnings("unchecked")
                public Boolean call(Object t1, Object t2) {
                    boolean c1 = t1 == LOCAL_ONCOMPLETED;
                    boolean c2 = t2 == LOCAL_ONCOMPLETED;
                    if (c1 && c2) {
                        return true;
                    }
                    if (c1 || c2) {
                        return false;
                    }
                    // Now t1 and t2 must be 'onNext'.
                    return equality.call((T)t1, (T)t2);
                }

            }).all(Functions.<Boolean> identity());
}
 

import rx.functions.Functions; //导入依赖的package包/类
public static <T1, T2, R> Observable<R> combineLatest(Observable<? extends T1> o1, Observable<? extends T2> o2, Func2<? super T1, ? super T2, ? extends R> combineFunction) {
    return combineLatest(Arrays.asList(new Observable[]{o1, o2}), Functions.fromFunc(combineFunction));
}
 

import rx.functions.Functions; //导入依赖的package包/类
public OperatorZip(Func2 f) {
    this.zipFunction = Functions.fromFunc(f);
}
 

import rx.functions.Functions; //导入依赖的package包/类
public static final <T1, T2, R> Observable<R> combineLatest(Observable<? extends T1> o1, Observable<? extends T2> o2, Func2<? super T1, ? super T2, ? extends R> combineFunction) {
    return combineLatest(Arrays.asList(new Observable[]{o1, o2}), Functions.fromFunc(combineFunction));
}
 

import rx.functions.Functions; //导入依赖的package包/类
@SuppressWarnings({ "unchecked", "rawtypes" })
public OperatorZip(BiFunction f) {
    this.zipFunction = Functions.fromFunc(f);
}
 

import rx.functions.Functions; //导入依赖的package包/类
@SuppressWarnings({ "unchecked", "rawtypes" })
public OperatorZip(Func2 f) {
    this.zipFunction = Functions.fromFunc(f);
}