OperatorToMap

rxjava-core/src/main/java/rx/Observable.java

@@ -74,7 +74,7 @@
 import rx.operators.OperationThrottleFirst;
 import rx.operators.OperationTimeInterval;
 import rx.operators.OperationTimer;
-import rx.operators.OperationToMap;
+import rx.operators.OperatorToMap;
 import rx.operators.OperationToMultimap;
 import rx.operators.OperationUsing;
 import rx.operators.OperationWindow;
@@ -7175,7 +7175,7 @@ public final Observable<List<T>> toList() {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229137.aspx">MSDN: Observable.ToDictionary</a>
      */
     public final <K> Observable<Map<K, T>> toMap(Func1<? super T, ? extends K> keySelector) {
-        return create(OperationToMap.toMap(this, keySelector));
+        return lift(new OperatorToMap<T, K, T>(keySelector, Functions.<T>identity()));
     }
 
     /**
@@ -7197,7 +7197,7 @@ public final <K> Observable<Map<K, T>> toMap(Func1<? super T, ? extends K> keySe
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh212075.aspx">MSDN: Observable.ToDictionary</a>
      */
     public final <K, V> Observable<Map<K, V>> toMap(Func1<? super T, ? extends K> keySelector, Func1<? super T, ? extends V> valueSelector) {
-        return create(OperationToMap.toMap(this, keySelector, valueSelector));
+        return lift(new OperatorToMap<T, K, V>(keySelector, valueSelector));
     }
 
     /**
@@ -7217,7 +7217,7 @@ public final <K, V> Observable<Map<K, V>> toMap(Func1<? super T, ? extends K> ke
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Mathematical-and-Aggregate-Operators#wiki-tomap-and-tomultimap">RxJava Wiki: toMap()</a>
      */
     public final <K, V> Observable<Map<K, V>> toMap(Func1<? super T, ? extends K> keySelector, Func1<? super T, ? extends V> valueSelector, Func0<? extends Map<K, V>> mapFactory) {
-        return create(OperationToMap.toMap(this, keySelector, valueSelector, mapFactory));
+        return lift(new OperatorToMap<T, K, V>(keySelector, valueSelector, mapFactory));
     }
 
     /**

rxjava-core/src/main/java/rx/operators/OperatorToMap.java

@@ -0,0 +1,104 @@
+/**
+ * Copyright 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.operators;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import rx.Observable.Operator;
+import rx.Subscriber;
+import rx.functions.Func0;
+import rx.functions.Func1;
+
+/**
+ * Maps the elements of the source observable into a java.util.Map instance and
+ * emits that once the source observable completes.
+ * 
+ * @see <a href='https://github.com/Netflix/RxJava/issues/96'>Issue #96</a>
+ */
+public final class OperatorToMap<T, K, V> implements Operator<Map<K, V>, T> {
+
+    /**
+     * The default map factory.
+     */
+    public static final class DefaultToMapFactory<K, V> implements Func0<Map<K, V>> {
+        @Override
+        public Map<K, V> call() {
+            return new HashMap<K, V>();
+        }
+    }
+
+
+    private final Func1<? super T, ? extends K> keySelector;
+
+    private final Func1<? super T, ? extends V> valueSelector;
+
+    private final Func0<? extends Map<K, V>> mapFactory;
+
+
+    /**
+     * ToMap with key selector, value selector and default HashMap factory.
+     */
+    public OperatorToMap(
+            Func1<? super T, ? extends K> keySelector,
+            Func1<? super T, ? extends V> valueSelector) {
+        this(keySelector, valueSelector, new DefaultToMapFactory<K, V>());
+    }
+
+
+    /**
+     * ToMap with key selector, value selector and custom Map factory.
+     */
+    public OperatorToMap(
+            Func1<? super T, ? extends K> keySelector,
+            Func1<? super T, ? extends V> valueSelector,
+            Func0<? extends Map<K, V>> mapFactory) {
+        this.keySelector = keySelector;
+        this.valueSelector = valueSelector;
+        this.mapFactory = mapFactory;
+
+    }
+
+    @Override
+    public Subscriber<? super T> call(final Subscriber<? super Map<K, V>> subscriber) {
+        return new Subscriber<T>(subscriber) {
+
+            private Map<K, V> map = mapFactory.call();
+
+            @Override
+            public void onNext(T v) {
+                K key = keySelector.call(v);
+                V value = valueSelector.call(v);
+                map.put(key, value);
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                map = null;
+                subscriber.onError(e);
+            }
+
+            @Override
+            public void onCompleted() {
+                Map<K, V> map0 = map;
+                map = null;
+                subscriber.onNext(map0);
+                subscriber.onCompleted();
+            }
+        };
+    }
+}

rxjava-core/src/test/java/rx/operators/OperationToMapTest.java → rxjava-core/src/test/java/rx/operators/OperatorToMapTest.java

@@ -35,7 +35,7 @@
 import rx.functions.Func1;
 import rx.functions.Functions;
 
-public class OperationToMapTest {
+public class OperatorToMapTest {
     @Mock
     Observer<Object> objectObserver;
 
@@ -61,7 +61,7 @@ public String call(String t1) {
     public void testToMap() {
         Observable<String> source = Observable.from("a", "bb", "ccc", "dddd");
 
-        Observable<Map<Integer, String>> mapped = Observable.create(OperationToMap.toMap(source, lengthFunc));
+        Observable<Map<Integer, String>> mapped = source.toMap(lengthFunc);
 
         Map<Integer, String> expected = new HashMap<Integer, String>();
         expected.put(1, "a");
@@ -80,7 +80,7 @@ public void testToMap() {
     public void testToMapWithValueSelector() {
         Observable<String> source = Observable.from("a", "bb", "ccc", "dddd");
 
-        Observable<Map<Integer, String>> mapped = Observable.create(OperationToMap.toMap(source, lengthFunc, duplicate));
+        Observable<Map<Integer, String>> mapped = source.toMap(lengthFunc, duplicate);
 
         Map<Integer, String> expected = new HashMap<Integer, String>();
         expected.put(1, "aa");
@@ -108,7 +108,7 @@ public Integer call(String t1) {
                 return t1.length();
             }
         };
-        Observable<Map<Integer, String>> mapped = Observable.create(OperationToMap.toMap(source, lengthFuncErr));
+        Observable<Map<Integer, String>> mapped = source.toMap(lengthFuncErr);
 
         Map<Integer, String> expected = new HashMap<Integer, String>();
         expected.put(1, "a");
@@ -138,7 +138,7 @@ public String call(String t1) {
             }
         };
 
-        Observable<Map<Integer, String>> mapped = Observable.create(OperationToMap.toMap(source, lengthFunc, duplicateErr));
+        Observable<Map<Integer, String>> mapped = source.toMap(lengthFunc, duplicateErr);
 
         Map<Integer, String> expected = new HashMap<Integer, String>();
         expected.put(1, "aa");
@@ -176,7 +176,7 @@ public Integer call(String t1) {
                 return t1.length();
             }
         };
-        Observable<Map<Integer, String>> mapped = Observable.create(OperationToMap.toMap(source, lengthFunc, Functions.<String> identity(), mapFactory));
+        Observable<Map<Integer, String>> mapped = source.toMap(lengthFunc, Functions.<String>identity(), mapFactory);
 
         Map<Integer, String> expected = new LinkedHashMap<Integer, String>();
         expected.put(2, "bb");
@@ -207,7 +207,7 @@ public Integer call(String t1) {
                 return t1.length();
             }
         };
-        Observable<Map<Integer, String>> mapped = Observable.create(OperationToMap.toMap(source, lengthFunc, Functions.<String> identity(), mapFactory));
+        Observable<Map<Integer, String>> mapped = source.toMap(lengthFunc, Functions.<String>identity(), mapFactory);
 
         Map<Integer, String> expected = new LinkedHashMap<Integer, String>();
         expected.put(2, "bb");