### Question:
DefaultSource implements Source<T> { @Override public <X> Source<X> compose(Function<Publisher<Message<T>>, Publisher<Message<X>>> mapper) { return new DefaultSource<>(asFlowable().compose(mapper::apply), 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; }### Answer:
@Test public void testCompose() { ListSink<Integer> list = Sink.list(); Source.from(1, 2, 3, 4, 5) .composeFlowable(flow -> flow.map(Message::payload).map(i -> i + 1).map(Message::new)) .to(list); assertThat(list.values()).containsExactly(2, 3, 4, 5, 6); }