@Test public void testOrElse() { Source<String> source = Source.from("a", "b", "c") .named("my source"); Source<String> empty = Source.<String>empty().named("empty"); Source<String> another = Source.from("d", "e", "f"); ListSink<String> sink = Sink.list(); source.orElse(another).to(sink); assertThat(sink.values()).containsExactly("a", "b", "c"); sink = Sink.list(); empty.orElse(source).to(sink); assertThat(sink.values()).containsExactly("a", "b", "c"); sink = Sink.list(); empty.orElse(another).to(sink); assertThat(sink.values()).containsExactly("d", "e", "f"); sink = Sink.list(); empty.orElse(empty).to(sink); assertThat(sink.values()).isEmpty(); }
@Override public Source<T> orElse(Source<T> alt) { Objects.requireNonNull(alt, "The alternative source must not be `null`"); return new DefaultSource<>(Flowable.fromPublisher(flow).switchIfEmpty(alt), name, attributes); }
DefaultSource implements Source<T> { @Override public Source<T> orElse(Source<T> alt) { Objects.requireNonNull(alt, "The alternative source must not be `null`"); return new DefaultSource<>(Flowable.fromPublisher(flow).switchIfEmpty(alt), name, attributes); } }
DefaultSource implements Source<T> { @Override public Source<T> orElse(Source<T> alt) { Objects.requireNonNull(alt, "The alternative source must not be `null`"); return new DefaultSource<>(Flowable.fromPublisher(flow).switchIfEmpty(alt), name, attributes); } DefaultSource(Publisher<Message<T>> items, String name, Map<String, Object> attr); }
DefaultSource implements Source<T> { @Override public Source<T> orElse(Source<T> alt) { Objects.requireNonNull(alt, "The alternative source must not be `null`"); return new DefaultSource<>(Flowable.fromPublisher(flow).switchIfEmpty(alt), name, attributes); } DefaultSource(Publisher<Message<T>> items, String name, Map<String, Object> attr); @Override Source<T> named(String name); @Override Source<T> unnamed(); @Override Source<T> withAttribute(String key, Object value); @Override Source<T> withoutAttribute(String key); @Override void subscribe(Subscriber<? super Message<T>> s); @Override Source<T> orElse(Source<T> alt); @Override String name(); @Override Optional<T> attr(String key); @Override Source<X> map(Function<Message<T>, Message<X>> mapper); @Override Source<X> mapPayload(Function<T, X> mapper); @Override Source<T> filter(Predicate<Message<T>> filter); @Override Source<T> filterPayload(Predicate<T> filter); @Override Source<T> filterNot(Predicate<Message<T>> filter); @Override Source<T> filterNotPayload(Predicate<T> filter); @Override Source<X> flatMap(Function<Message<T>, Publisher<Message<X>>> mapper); @Override Source<X> concatMap(Function<Message<T>, Publisher<Message<X>>> mapper); @Override Source<X> flatMap(Function<Message<T>, Publisher<Message<X>>> mapper, int maxConcurrency); @Override Source<X> flatMapPayload(Function<T, Publisher<X>> mapper); @Override Source<X> concatMapPayload(Function<T, Publisher<X>> mapper); @Override Source<X> flatMapPayload(Function<T, Publisher<X>> mapper, int maxConcurrency); @Override Source<X> scan(Message<X> zero, BiFunction<Message<X>, Message<T>, Message<X>> function); @Override Source<X> scanPayloads(X zero, BiFunction<X, T, X> function); @Override Publisher<GroupedDataStream<K, T>> groupBy(Function<Message<T>, K> keySupplier); @Override Source<T> log(String loggerName); @Override List<Source<T>> broadcast(int numberOfBranches); @Override Map<String, Source<T>> broadcast(String... names); @Override Pair<Source<T>, Source<T>> branch(Predicate<Message<T>> condition); @Override Pair<Source<T>, Source<T>> branchOnPayload(Predicate<T> condition); @Override Sink<T> to(Sink<T> sink); @Override Flowable<Message<T>> asFlowable(); @Override Source<Pair<T, O>> zipWith(Publisher<Message<O>> source); @Override Source<Tuple> zipWith(Source... sources); Source<Tuple> zipWith(Publisher<Message>... sources); @Override Source<T> mergeWith(Publisher<Message<T>> source); @Override Source<T> mergeWith(Publisher<Message<T>>... sources); @Override Source<X> compose(Function<Publisher<Message<T>>, Publisher<Message<X>>> mapper); @Override Source<X> composeFlowable(Function<Flowable<Message<T>>, Flowable<Message<X>>> mapper); @Override Source<X> composePayloadFlowable(Function<Flowable<T>, Flowable<X>> function); }
DefaultSource implements Source<T> { @Override public Source<T> orElse(Source<T> alt) { Objects.requireNonNull(alt, "The alternative source must not be `null`"); return new DefaultSource<>(Flowable.fromPublisher(flow).switchIfEmpty(alt), name, attributes); } DefaultSource(Publisher<Message<T>> items, String name, Map<String, Object> attr); @Override Source<T> named(String name); @Override Source<T> unnamed(); @Override Source<T> withAttribute(String key, Object value); @Override Source<T> withoutAttribute(String key); @Override void subscribe(Subscriber<? super Message<T>> s); @Override Source<T> orElse(Source<T> alt); @Override String name(); @Override Optional<T> attr(String key); @Override Source<X> map(Function<Message<T>, Message<X>> mapper); @Override Source<X> mapPayload(Function<T, X> mapper); @Override Source<T> filter(Predicate<Message<T>> filter); @Override Source<T> filterPayload(Predicate<T> filter); @Override Source<T> filterNot(Predicate<Message<T>> filter); @Override Source<T> filterNotPayload(Predicate<T> filter); @Override Source<X> flatMap(Function<Message<T>, Publisher<Message<X>>> mapper); @Override Source<X> concatMap(Function<Message<T>, Publisher<Message<X>>> mapper); @Override Source<X> flatMap(Function<Message<T>, Publisher<Message<X>>> mapper, int maxConcurrency); @Override Source<X> flatMapPayload(Function<T, Publisher<X>> mapper); @Override Source<X> concatMapPayload(Function<T, Publisher<X>> mapper); @Override Source<X> flatMapPayload(Function<T, Publisher<X>> mapper, int maxConcurrency); @Override Source<X> scan(Message<X> zero, BiFunction<Message<X>, Message<T>, Message<X>> function); @Override Source<X> scanPayloads(X zero, BiFunction<X, T, X> function); @Override Publisher<GroupedDataStream<K, T>> groupBy(Function<Message<T>, K> keySupplier); @Override Source<T> log(String loggerName); @Override List<Source<T>> broadcast(int numberOfBranches); @Override Map<String, Source<T>> broadcast(String... names); @Override Pair<Source<T>, Source<T>> branch(Predicate<Message<T>> condition); @Override Pair<Source<T>, Source<T>> branchOnPayload(Predicate<T> condition); @Override Sink<T> to(Sink<T> sink); @Override Flowable<Message<T>> asFlowable(); @Override Source<Pair<T, O>> zipWith(Publisher<Message<O>> source); @Override Source<Tuple> zipWith(Source... sources); Source<Tuple> zipWith(Publisher<Message>... sources); @Override Source<T> mergeWith(Publisher<Message<T>> source); @Override Source<T> mergeWith(Publisher<Message<T>>... sources); @Override Source<X> compose(Function<Publisher<Message<T>>, Publisher<Message<X>>> mapper); @Override Source<X> composeFlowable(Function<Flowable<Message<T>>, Flowable<Message<X>>> mapper); @Override Source<X> composePayloadFlowable(Function<Flowable<T>, Flowable<X>> function); static final String FUNCTION_CANNOT_BE_NULL_MESSAGE; static final String FILTER_CANNOT_BE_NULL_MESSAGE; }