OperationToMultimap { public static <T, K> OnSubscribeFunc<Map<K, Collection<T>>> toMultimap( Observable<T> source, Func1<? super T, ? extends K> keySelector ) { return new ToMultimap<T, K, T>( source, keySelector, Functions.<T> identity(), new DefaultToMultimapFactory<K, T>(), new DefaultMultimapCollectionFactory<K, T>()); }  static OnSubscribeFunc<Map<K, Collection<T>>> toMultimap(
            Observable<T> source,
            Func1<? super T, ? extends K> keySelector
            ); static OnSubscribeFunc<Map<K, Collection<V>>> toMultimap(
            Observable<T> source,
            Func1<? super T, ? extends K> keySelector,
            Func1<? super T, ? extends V> valueSelector
            ); static OnSubscribeFunc<Map<K, Collection<V>>> toMultimap(
            Observable<T> source,
            Func1<? super T, ? extends K> keySelector,
            Func1<? super T, ? extends V> valueSelector,
            Func0<? extends Map<K, Collection<V>>> mapFactory
            ); static OnSubscribeFunc<Map<K, Collection<V>>> toMultimap(
            Observable<T> source,
            Func1<? super T, ? extends K> keySelector,
            Func1<? super T, ? extends V> valueSelector,
            Func0<? extends Map<K, Collection<V>>> mapFactory,
            Func1<? super K, ? extends Collection<V>> collectionFactory
            );  }
@Test public void testToMultimap() { Observable<String> source = Observable.from("a", "b", "cc", "dd"); Observable<Map<Integer, Collection<String>>> mapped = Observable.create(OperationToMultimap.toMultimap(source, lengthFunc)); Map<Integer, Collection<String>> expected = new HashMap<Integer, Collection<String>>(); expected.put(1, Arrays.asList("a", "b")); expected.put(2, Arrays.asList("cc", "dd")); mapped.subscribe(objectObserver); verify(objectObserver, never()).onError(any(Throwable.class)); verify(objectObserver, times(1)).onNext(expected); verify(objectObserver, times(1)).onCompleted(); }
@Test public void testToMultimapWithValueSelector() { Observable<String> source = Observable.from("a", "b", "cc", "dd"); Observable<Map<Integer, Collection<String>>> mapped = Observable.create(OperationToMultimap.toMultimap(source, lengthFunc, duplicate)); Map<Integer, Collection<String>> expected = new HashMap<Integer, Collection<String>>(); expected.put(1, Arrays.asList("aa", "bb")); expected.put(2, Arrays.asList("cccc", "dddd")); mapped.subscribe(objectObserver); verify(objectObserver, never()).onError(any(Throwable.class)); verify(objectObserver, times(1)).onNext(expected); verify(objectObserver, times(1)).onCompleted(); }
@Test public void testToMultimapWithMapFactory() { Observable<String> source = Observable.from("a", "b", "cc", "dd", "eee", "fff"); Func0<Map<Integer, Collection<String>>> mapFactory = new Func0<Map<Integer, Collection<String>>>() { @Override public Map<Integer, Collection<String>> call() { return new LinkedHashMap<Integer, Collection<String>>() { @Override protected boolean removeEldestEntry(Map.Entry<Integer, Collection<String>> eldest) { return size() > 2; } }; } }; Observable<Map<Integer, Collection<String>>> mapped = Observable.create( OperationToMultimap.toMultimap(source, lengthFunc, Functions.<String> identity(), mapFactory, new DefaultMultimapCollectionFactory<Integer, String>())); Map<Integer, Collection<String>> expected = new HashMap<Integer, Collection<String>>(); expected.put(2, Arrays.asList("cc", "dd")); expected.put(3, Arrays.asList("eee", "fff")); mapped.subscribe(objectObserver); verify(objectObserver, never()).onError(any(Throwable.class)); verify(objectObserver, times(1)).onNext(expected); verify(objectObserver, times(1)).onCompleted(); }
@Test public void testToMultimapWithCollectionFactory() { Observable<String> source = Observable.from("cc", "dd", "eee", "eee"); Func1<Integer, Collection<String>> collectionFactory = new Func1<Integer, Collection<String>>() { @Override public Collection<String> call(Integer t1) { if (t1 == 2) { return new ArrayList<String>(); } else { return new HashSet<String>(); } } }; Observable<Map<Integer, Collection<String>>> mapped = Observable.create( OperationToMultimap.toMultimap( source, lengthFunc, Functions.<String> identity(), new DefaultToMultimapFactory<Integer, String>(), collectionFactory)); Map<Integer, Collection<String>> expected = new HashMap<Integer, Collection<String>>(); expected.put(2, Arrays.asList("cc", "dd")); expected.put(3, new HashSet<String>(Arrays.asList("eee"))); mapped.subscribe(objectObserver); verify(objectObserver, never()).onError(any(Throwable.class)); verify(objectObserver, times(1)).onNext(expected); verify(objectObserver, times(1)).onCompleted(); }
@Test public void testToMultimapWithError() { Observable<String> source = Observable.from("a", "b", "cc", "dd"); Func1<String, Integer> lengthFuncErr = new Func1<String, Integer>() { @Override public Integer call(String t1) { if ("b".equals(t1)) { throw new RuntimeException("Forced Failure"); } return t1.length(); } }; Observable<Map<Integer, Collection<String>>> mapped = Observable.create(OperationToMultimap.toMultimap(source, lengthFuncErr)); Map<Integer, Collection<String>> expected = new HashMap<Integer, Collection<String>>(); expected.put(1, Arrays.asList("a", "b")); expected.put(2, Arrays.asList("cc", "dd")); mapped.subscribe(objectObserver); verify(objectObserver, times(1)).onError(any(Throwable.class)); verify(objectObserver, never()).onNext(expected); verify(objectObserver, never()).onCompleted(); }
@Test public void testToMultimapWithErrorInValueSelector() { Observable<String> source = Observable.from("a", "b", "cc", "dd"); Func1<String, String> duplicateErr = new Func1<String, String>() { @Override public String call(String t1) { if ("b".equals(t1)) { throw new RuntimeException("Forced failure"); } return t1 + t1; } }; Observable<Map<Integer, Collection<String>>> mapped = Observable.create(OperationToMultimap.toMultimap(source, lengthFunc, duplicateErr)); Map<Integer, Collection<String>> expected = new HashMap<Integer, Collection<String>>(); expected.put(1, Arrays.asList("aa", "bb")); expected.put(2, Arrays.asList("cccc", "dddd")); mapped.subscribe(objectObserver); verify(objectObserver, times(1)).onError(any(Throwable.class)); verify(objectObserver, never()).onNext(expected); verify(objectObserver, never()).onCompleted(); }
@Test public void testToMultimapWithMapThrowingFactory() { Observable<String> source = Observable.from("a", "b", "cc", "dd", "eee", "fff"); Func0<Map<Integer, Collection<String>>> mapFactory = new Func0<Map<Integer, Collection<String>>>() { @Override public Map<Integer, Collection<String>> call() { throw new RuntimeException("Forced failure"); } }; Observable<Map<Integer, Collection<String>>> mapped = Observable.create( OperationToMultimap.toMultimap(source, lengthFunc, Functions.<String> identity(), mapFactory)); Map<Integer, Collection<String>> expected = new HashMap<Integer, Collection<String>>(); expected.put(2, Arrays.asList("cc", "dd")); expected.put(3, Arrays.asList("eee", "fff")); mapped.subscribe(objectObserver); verify(objectObserver, times(1)).onError(any(Throwable.class)); verify(objectObserver, never()).onNext(expected); verify(objectObserver, never()).onCompleted(); }
@Test public void testToMultimapWithThrowingCollectionFactory() { Observable<String> source = Observable.from("cc", "cc", "eee", "eee"); Func1<Integer, Collection<String>> collectionFactory = new Func1<Integer, Collection<String>>() { @Override public Collection<String> call(Integer t1) { if (t1 == 2) { throw new RuntimeException("Forced failure"); } else { return new HashSet<String>(); } } }; Observable<Map<Integer, Collection<String>>> mapped = Observable.create( OperationToMultimap.toMultimap( source, lengthFunc, Functions.<String> identity(), new DefaultToMultimapFactory(), collectionFactory)); Map<Integer, Collection<String>> expected = new HashMap<Integer, Collection<String>>(); expected.put(2, Arrays.asList("cc", "dd")); expected.put(3, Collections.singleton("eee")); mapped.subscribe(objectObserver); verify(objectObserver, times(1)).onError(any(Throwable.class)); verify(objectObserver, never()).onNext(expected); verify(objectObserver, never()).onCompleted(); }