/** * Copyright one 2014 Netflix, Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package rx.internal.operators; import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; import java.util.Map; import rx.Observable; import rx.Observable.OnSubscribe; import rx.Subscriber; import rx.exceptions.Exceptions; import rx.functions.Func0; import rx.functions.Func1; /** * Maps the elements of the source observable into a multimap * (Map<K, Collection<V>>) where each * key entry has a collection of the source's values. * * @see Issue #97 * @param the value type of the input * @param the multimap-key type * @param the multimap-value type */ public final class OnSubscribeToMultimap implements OnSubscribe>>, Func0>> { private final Func1 keySelector; private final Func1 valueSelector; private final Func0>> mapFactory; private final Func1> collectionFactory; private final Observable source; /** * ToMultimap with key selector, custom value selector, * default HashMap factory and default ArrayList collection factory. * @param source the source Observable instance * @param keySelector the function extracting the map-key from the main value * @param valueSelector the function extracting the map-value from the main value */ public OnSubscribeToMultimap( Observable source, Func1 keySelector, Func1 valueSelector) { this(source, keySelector, valueSelector, null, DefaultMultimapCollectionFactory.instance()); } /** * ToMultimap with key selector, custom value selector, * custom Map factory and default ArrayList collection factory. * @param source the source Observable instance * @param keySelector the function extracting the map-key from the main value * @param valueSelector the function extracting the map-value from the main value * @param mapFactory function that returns a Map instance to store keys and values into */ public OnSubscribeToMultimap( Observable source, Func1 keySelector, Func1 valueSelector, Func0>> mapFactory) { this(source, keySelector, valueSelector, mapFactory, DefaultMultimapCollectionFactory.instance()); } /** * ToMultimap with key selector, custom value selector, * custom Map factory and custom collection factory. * @param source the observable source * @param keySelector the function extracting the map-key from the main value * @param valueSelector the function extracting the map-value from the main value * @param mapFactory function that returns a Map instance to store keys and values into * @param collectionFactory function that returns a Collection for a particular key to store values into */ public OnSubscribeToMultimap( Observable source, Func1 keySelector, Func1 valueSelector, Func0>> mapFactory, Func1> collectionFactory) { this.source = source; this.keySelector = keySelector; this.valueSelector = valueSelector; if (mapFactory == null) { this.mapFactory = this; } else { this.mapFactory = mapFactory; } this.collectionFactory = collectionFactory; } // default map factory @Override public Map> call() { return new HashMap>(); } @Override public void call(final Subscriber>> subscriber) { Map> map; try { map = mapFactory.call(); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); subscriber.onError(ex); return; } new ToMultimapSubscriber( subscriber, map, keySelector, valueSelector, collectionFactory) .subscribeTo(source); } private static final class ToMultimapSubscriber extends DeferredScalarSubscriberSafe>> { private final Func1 keySelector; private final Func1 valueSelector; private final Func1> collectionFactory; ToMultimapSubscriber( Subscriber>> subscriber, Map> map, Func1 keySelector, Func1 valueSelector, Func1> collectionFactory) { super(subscriber); this.value = map; this.hasValue = true; this.keySelector = keySelector; this.valueSelector = valueSelector; this.collectionFactory = collectionFactory; } @Override public void onStart() { request(Long.MAX_VALUE); } @Override public void onNext(T t) { if (done){ return; } try { // any interaction with keySelector, valueSelector, collectionFactory, collection or value // may fail unexpectedly because their behaviour is customisable by the user. For this // reason we wrap their calls in try-catch block. K key = keySelector.call(t); V v = valueSelector.call(t); Collection collection = value.get(key); if (collection == null) { collection = collectionFactory.call(key); value.put(key, collection); } collection.add(v); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); unsubscribe(); onError(ex); } } } /** * The default collection factory for a key in the multimap returning * an ArrayList independent of the key. * @param the key type * @param the value type */ private static final class DefaultMultimapCollectionFactory implements Func1> { private static final DefaultMultimapCollectionFactory INSTANCE = new DefaultMultimapCollectionFactory(); @SuppressWarnings("unchecked") static DefaultMultimapCollectionFactory instance() { return (DefaultMultimapCollectionFactory) INSTANCE; } @Override public Collection call(K t1) { return new ArrayList(); } } }