@Test public void testGroupBy() { String text = "In 1815, M. Charles–Francois-Bienvenu Myriel was Bishop of D He was an old man of about seventy-five " + "years of age; he had occupied the see of D since 1806. Although this detail has no connection whatever with the " + "real substance of what we are about to relate, it will not be superfluous, if merely for the sake of exactness in all" + " points, to mention here the various rumors and remarks which had been in circulation about him from the very moment " + "when he arrived in the diocese. True or false, that which is said of men often occupies as important a place in their" + " lives, and above all in their destinies, as that which they do. M. Myriel was the son of a councillor of the " + "Parliament of Aix; hence he belonged to the nobility of the bar. It was said that his father, destining him to be the" + " heir of his own post, had married him at a very early age, eighteen or twenty, in accordance with a custom which is " + "rather widely prevalent in parliamentary families. In spite of this marriage, however, it was said that Charles Myriel" + " created a great deal of talk. He was well formed, though rather short in stature, elegant, graceful, intelligent; the" + " whole of the first portion of his life had been devoted to the world and to gallantry."; Publisher<GroupedDataStream<Character, String>> publisher = Source.from( text.replaceAll("[^a-zA-Z ]", "").toLowerCase().split("\\s+")) .mapPayload(String::toLowerCase) .groupBy(msg -> msg.payload().charAt(0)); Multimap<Character, String> wordsByFirstLetter = ArrayListMultimap.create(); Flowable.fromPublisher(publisher) .doOnNext(stream -> Flowable.fromPublisher(stream).subscribe(word -> wordsByFirstLetter.put(stream.key(), word.payload()))) .subscribe(); assertThat(wordsByFirstLetter.get('a')).hasSize(25).contains("aix"); assertThat(wordsByFirstLetter.get('y')).hasSize(1).contains("years"); assertThat(wordsByFirstLetter.get('w')).hasSize(21).contains("widely"); assertThat(wordsByFirstLetter.get('j')).isEmpty(); }
@Override public <K> Publisher<GroupedDataStream<K, T>> groupBy(Function<Message<T>, K> keySupplier) { Objects.requireNonNull(keySupplier, "The function computing the key must not be `null`"); return Flowable.fromPublisher(flow) .groupBy(keySupplier::apply) .flatMapSingle(gf -> { GroupedDataStream<K, T> stream = new GroupedDataStream<>(gf.getKey(), gf); return Single.just(stream); }); }
DefaultSource implements Source<T> { @Override public <K> Publisher<GroupedDataStream<K, T>> groupBy(Function<Message<T>, K> keySupplier) { Objects.requireNonNull(keySupplier, "The function computing the key must not be `null`"); return Flowable.fromPublisher(flow) .groupBy(keySupplier::apply) .flatMapSingle(gf -> { GroupedDataStream<K, T> stream = new GroupedDataStream<>(gf.getKey(), gf); return Single.just(stream); }); } }
DefaultSource implements Source<T> { @Override public <K> Publisher<GroupedDataStream<K, T>> groupBy(Function<Message<T>, K> keySupplier) { Objects.requireNonNull(keySupplier, "The function computing the key must not be `null`"); return Flowable.fromPublisher(flow) .groupBy(keySupplier::apply) .flatMapSingle(gf -> { GroupedDataStream<K, T> stream = new GroupedDataStream<>(gf.getKey(), gf); return Single.just(stream); }); } DefaultSource(Publisher<Message<T>> items, String name, Map<String, Object> attr); }
DefaultSource implements Source<T> { @Override public <K> Publisher<GroupedDataStream<K, T>> groupBy(Function<Message<T>, K> keySupplier) { Objects.requireNonNull(keySupplier, "The function computing the key must not be `null`"); return Flowable.fromPublisher(flow) .groupBy(keySupplier::apply) .flatMapSingle(gf -> { GroupedDataStream<K, T> stream = new GroupedDataStream<>(gf.getKey(), gf); return Single.just(stream); }); } 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 <K> Publisher<GroupedDataStream<K, T>> groupBy(Function<Message<T>, K> keySupplier) { Objects.requireNonNull(keySupplier, "The function computing the key must not be `null`"); return Flowable.fromPublisher(flow) .groupBy(keySupplier::apply) .flatMapSingle(gf -> { GroupedDataStream<K, T> stream = new GroupedDataStream<>(gf.getKey(), gf); return Single.just(stream); }); } 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; }