EventStreams.java 文件源码

/**
 * A more general version of {@link #merge(ObservableSet)} for a set of
 * arbitrary element type and a function to obtain an event stream from
 * the element.
 * @param set observable set of elements
 * @param f function to obtain an event stream from an element
 */
public static <T, U> EventStream<U> merge(
        ObservableSet<? extends T> set,
        Function<? super T, ? extends EventStream<U>> f) {
    return new EventStreamBase<U>() {
        @Override
        protected Subscription observeInputs() {
            return Subscription.dynamic(
                    set,
                    t -> f.apply(t).subscribe(this::emit));
        }
    };
}