diff --git a/language-adaptors/rxjava-groovy/src/test/groovy/rx/lang/groovy/ObservableTests.groovy b/language-adaptors/rxjava-groovy/src/test/groovy/rx/lang/groovy/ObservableTests.groovy
index ceeb25dcf6..9c6e244149 100644
--- a/language-adaptors/rxjava-groovy/src/test/groovy/rx/lang/groovy/ObservableTests.groovy
+++ b/language-adaptors/rxjava-groovy/src/test/groovy/rx/lang/groovy/ObservableTests.groovy
@@ -345,6 +345,143 @@ def class ObservableTests {
           assertEquals(6, count);
     }
     
+    @Test
+    public void testToMap1() {
+        Map actual = new HashMap();
+        
+        Observable.from("a", "bb", "ccc", "dddd")
+        .toMap({String s -> s.length()})
+        .toBlockingObservable()
+        .forEach({s -> actual.putAll(s); });
+        
+        Map expected = new HashMap();
+        expected.put(1, "a");
+        expected.put(2, "bb");
+        expected.put(3, "ccc");
+        expected.put(4, "dddd");
+        
+        assertEquals(expected, actual);
+    }
+
+    @Test
+    public void testToMap2() {
+        Map actual = new HashMap();
+        
+        Observable.from("a", "bb", "ccc", "dddd")
+        .toMap({String s -> s.length()}, {String s -> s + s})
+        .toBlockingObservable()
+        .forEach({s -> actual.putAll(s); });
+        
+        Map expected = new HashMap();
+        expected.put(1, "aa");
+        expected.put(2, "bbbb");
+        expected.put(3, "cccccc");
+        expected.put(4, "dddddddd");
+        
+        assertEquals(expected, actual);
+    }
+
+    @Test
+    public void testToMap3() {
+        Map actual = new HashMap();
+        
+        LinkedHashMap last3 = new LinkedHashMap() {
+            public boolean removeEldestEntry(Map.Entry e) {
+                return size() > 3;
+            }
+        };
+        
+        Observable.from("a", "bb", "ccc", "dddd")
+        .toMap({String s -> s.length()}, {String s -> s + s}, { last3 })
+        .toBlockingObservable()
+        .forEach({s -> actual.putAll(s); });
+        
+        Map expected = new HashMap();
+        expected.put(2, "bbbb");
+        expected.put(3, "cccccc");
+        expected.put(4, "dddddddd");
+        
+        assertEquals(expected, actual);
+    }
+    @Test
+    public void testToMultimap1() {
+        Map actual = new HashMap();
+        
+        Observable.from("a", "b", "cc", "dd")
+        .toMultimap({String s -> s.length()})
+        .toBlockingObservable()
+        .forEach({s -> actual.putAll(s); });
+        
+        Map expected = new HashMap();
+        
+        expected.put(1, Arrays.asList("a", "b"));
+        expected.put(2, Arrays.asList("cc", "dd"));
+        
+        assertEquals(expected, actual);
+    }
+
+    @Test
+    public void testToMultimap2() {
+        Map actual = new HashMap();
+        
+        Observable.from("a", "b", "cc", "dd")
+        .toMultimap({String s -> s.length()}, {String s -> s + s})
+        .toBlockingObservable()
+        .forEach({s -> actual.putAll(s); });
+        
+        Map expected = new HashMap();
+        
+        expected.put(1, Arrays.asList("aa", "bb"));
+        expected.put(2, Arrays.asList("cccc", "dddd"));
+        
+        assertEquals(expected, actual);
+    }
+
+    @Test
+    public void testToMultimap3() {
+        Map actual = new HashMap();
+        
+        LinkedHashMap last1 = new LinkedHashMap() {
+            public boolean removeEldestEntry(Map.Entry e) {
+                return size() > 1;
+            }
+        };
+        
+        Observable.from("a", "b", "cc", "dd")
+        .toMultimap({String s -> s.length()}, {String s -> s + s}, { last1 })
+        .toBlockingObservable()
+        .forEach({s -> actual.putAll(s); });
+        
+        Map expected = new HashMap();
+        
+        expected.put(2, Arrays.asList("cccc", "dddd"));
+        
+        assertEquals(expected, actual);
+    }
+
+    @Test
+    public void testToMultimap4() {
+        Map actual = new HashMap();
+        
+        LinkedHashMap last1 = new LinkedHashMap() {
+            public boolean removeEldestEntry(Map.Entry e) {
+                return size() > 2;
+            }
+        };
+        
+        Observable.from("a", "b", "cc", "dd", "eee", "eee")
+        .toMultimap({String s -> s.length()}, {String s -> s + s}, { last1 }, 
+            {i -> i == 2 ? new ArrayList() : new HashSet() })
+        .toBlockingObservable()
+        .forEach({s -> actual.putAll(s); });
+        
+        Map expected = new HashMap();
+        
+        expected.put(2, Arrays.asList("cccc", "dddd"));
+        expected.put(3, new HashSet(Arrays.asList("eeeeee")));
+        
+        assertEquals(expected, actual);
+    }
 
     def class AsyncObservable implements OnSubscribeFunc {
 
diff --git a/rxjava-core/src/main/java/rx/Observable.java b/rxjava-core/src/main/java/rx/Observable.java
index 06a08ef576..1f33710e95 100644
--- a/rxjava-core/src/main/java/rx/Observable.java
+++ b/rxjava-core/src/main/java/rx/Observable.java
@@ -19,8 +19,10 @@
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Comparator;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
@@ -87,6 +89,8 @@
 import rx.operators.OperationTimeInterval;
 import rx.operators.OperationTimeout;
 import rx.operators.OperationTimestamp;
+import rx.operators.OperationToMap;
+import rx.operators.OperationToMultimap;
 import rx.operators.OperationToObservableFuture;
 import rx.operators.OperationToObservableIterable;
 import rx.operators.OperationToObservableList;
@@ -5943,6 +5947,7 @@ public static <R> Observable<R> when(Plan0<R> p1, Plan0<R> p2, Plan0<R> p3, Plan
     public static <R> Observable<R> when(Plan0<R> p1, Plan0<R> p2, Plan0<R> p3, Plan0<R> p4, Plan0<R> p5, Plan0<R> p6, Plan0<R> p7, Plan0<R> p8, Plan0<R> p9) {
         return create(OperationJoinPatterns.when(p1, p2, p3, p4, p5, p6, p7, p8, p9));
     }
+
     /**
      * Correlates the elements of two sequences based on overlapping durations.
      * @param right The right observable sequence to join elements for.
@@ -5964,5 +5969,128 @@ public <TRight, TLeftDuration, TRightDuration, R> Observable<R> join(Observable<
             Func2<T, TRight, R> resultSelector) {
         return create(new OperationJoin<T, TRight, TLeftDuration, TRightDuration, R>(this, right, leftDurationSelector, rightDurationSelector, resultSelector));
     }    
+    
+    /**
+     * Return an Observable that emits a single HashMap containing all items
+     * emitted by the source Observable, mapped by the keys returned by the
+     * {@code keySelector} function.
+     * 
+     * If a source item maps to the same key, the HashMap will contain the latest
+     * of those items.
+     * 
+     * @param keySelector the function that extracts the key from the source items
+     *                    to be used as keys in the HashMap.
+     * @return an Observable that emits a single HashMap containing the mapped
+     *         values of the source Observable
+     * @see <a href='http://msdn.microsoft.com/en-us/library/hh229137(v=vs.103).aspx'>MSDN: Observable.ToDictionary</a>
+     */
+    public <K> Observable<Map<K, T>> toMap(Func1<? super T, ? extends K> keySelector) {
+        return create(OperationToMap.toMap(this, keySelector));
+    }
+    
+    /**
+     * Return an Observable that emits a single HashMap containing elements with
+     * key and value extracted from the values emitted by the source Observable.
+     * 
+     * If a source item maps to the same key, the HashMap will contain the latest
+     * of those items.
+     * 
+     * @param keySelector the function that extracts the key from the source items
+     *                    to be used as key in the HashMap
+     * @param valueSelector the function that extracts the value from the source items
+     *                      to be used as value in the HashMap
+     * @return an Observable that emits a single HashMap containing the mapped
+     *         values of the source Observable
+     * @see <a href='http://msdn.microsoft.com/en-us/library/hh212075(v=vs.103).aspx'>MSDN: Observable.ToDictionary</a>
+     */
+    public <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 an Observable that emits a single Map, returned by the mapFactory function,
+     * containing key and value extracted from the values emitted by the source Observable.
+     * 
+     * @param keySelector the function that extracts the key from the source items
+     *                    to be used as key in the Map
+     * @param valueSelector the function that extracts the value from the source items
+     *                      to be used as value in the Map
+     * @param mapFactory the function that returns an Map instance to be used
+     * @return an Observable that emits a single Map containing the mapped
+     *         values of the source Observable
+     */
+    public <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 an Observable that emits a single HashMap containing an ArrayList of elements,
+     * emitted by the source Observable and keyed by the keySelector function.
+     * 
+     * @param keySelector the function that extracts the key from the source items
+     *                    to be used as key in the HashMap
+     * @return an Observable that emits a single HashMap containing an ArrayList of elements
+     *         mapped from the source Observable
+     * @see <a href='http://msdn.microsoft.com/en-us/library/hh212098(v=vs.103).aspx'>MSDN: Observable.ToLookup</a>
+     */
+    public <K> Observable<Map<K, Collection<T>>> toMultimap(Func1<? super T, ? extends K> keySelector) {
+        return create(OperationToMultimap.toMultimap(this, keySelector));
+    }
+    
+    /**
+     * Return an Observable that emits a single HashMap containing an ArrayList of values,
+     * extracted by the valueSelector function, emitted by the source Observable
+     * and keyed by the keySelector function.
+     * 
+     * @param keySelector the function that extracts the key from the source items
+     *                    to be used as key in the HashMap
+     * @param valueSelector the function that extracts the value from the source items
+     *                      to be used as value in the Map
+     * @return an Observable that emits a single HashMap containing an ArrayList of elements
+     *         mapped from the source Observable
+     * 
+     * @see <a href='http://msdn.microsoft.com/en-us/library/hh229101(v=vs.103).aspx'>MSDN: Observable.ToLookup</a>
+     */
+    public <K, V> Observable<Map<K, Collection<V>>> toMultimap(Func1<? super T, ? extends K> keySelector, Func1<? super T, ? extends V> valueSelector) {
+        return create(OperationToMultimap.toMultimap(this, keySelector, valueSelector));
+    }
+    
+    /**
+     * Return an Observable that emits a single Map, returned by the mapFactory function,
+     * containing an ArrayList of values, extracted by the valueSelector function,
+     * emitted by the source Observable and keyed by the 
+     * keySelector function.
+     * 
+     * @param keySelector the function that extracts the key from the source items
+     *                    to be used as key in the Map
+     * @param valueSelector the function that extracts the value from the source items
+     *                      to be used as value in the Map
+     * @param mapFactory the function that returns an Map instance to be used
+     * @return an Observable that emits a single Map containing the list of mapped values
+     *         of the source observable.
+     */
+    public <K, V> Observable<Map<K, Collection<V>>> toMultimap(Func1<? super T, ? extends K> keySelector, Func1<? super T, ? extends V> valueSelector, Func0<? extends Map<K, Collection<V>>> mapFactory) {
+        return create(OperationToMultimap.toMultimap(this, keySelector, valueSelector, mapFactory));
+    }
+
+    /**
+     * Return an Observable that emits a single Map, returned by the mapFactory function,
+     * containing a custom collection of values, extracted by the valueSelector function,
+     * emitted by the source Observable and keyed by the 
+     * keySelector function.
+     * 
+     * @param keySelector the function that extracts the key from the source items
+     *                    to be used as key in the Map
+     * @param valueSelector the function that extracts the value from the source items
+     *                      to be used as value in the Map
+     * @param mapFactory the function that returns an Map instance to be used
+     * @param collectionFactory the function that returns a Collection instance for 
+     *                          a particular key to be used in the Map
+     * @return an Observable that emits a single Map containing the collection of mapped values
+     *         of the source observable.
+     */
+    public <K, V> Observable<Map<K, Collection<V>>> toMultimap(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) {
+        return create(OperationToMultimap.toMultimap(this, keySelector, valueSelector, mapFactory, collectionFactory));
+    } 
 }
 
diff --git a/rxjava-core/src/main/java/rx/operators/OperationToMap.java b/rxjava-core/src/main/java/rx/operators/OperationToMap.java
new file mode 100644
index 0000000000..c52b0244f2
--- /dev/null
+++ b/rxjava-core/src/main/java/rx/operators/OperationToMap.java
@@ -0,0 +1,159 @@
+/**
+ * Copyright 2013 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;
+import rx.Observable.OnSubscribeFunc;
+import rx.Observer;
+import rx.Subscription;
+import rx.subscriptions.Subscriptions;
+import rx.util.functions.Func0;
+import rx.util.functions.Func1;
+import rx.util.functions.Functions;
+
+/**
+ * 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 class OperationToMap {
+    /**
+     * ToMap with key selector, identity value selector and default HashMap factory.
+     */
+    public static <T, K> OnSubscribeFunc<Map<K, T>> toMap(Observable<T> source,
+            Func1<? super T, ? extends K> keySelector) {
+        return new ToMap<T, K, T>(source, keySelector,
+        Functions.<T>identity(), new DefaultToMapFactory<K, T>());
+    }
+
+    /**
+     * ToMap with key selector, value selector and default HashMap factory.
+     */
+    public static <T, K, V> OnSubscribeFunc<Map<K, V>> toMap(Observable<T> source,
+            Func1<? super T, ? extends K> keySelector,
+            Func1<? super T, ? extends V> valueSelector) {
+        return new ToMap<T, K, V>(source, keySelector,
+        valueSelector, new DefaultToMapFactory<K, V>());
+    }
+
+    /**
+     * ToMap with key selector, value selector and custom Map factory.
+     */
+    public static <T, K, V> OnSubscribeFunc<Map<K, V>> toMap(Observable<T> source,
+            Func1<? super T, ? extends K> keySelector,
+            Func1<? super T, ? extends V> valueSelector,
+            Func0<? extends Map<K, V>> mapFactory) {
+        return new ToMap<T, K, V>(source, keySelector,
+        valueSelector, mapFactory);
+    }
+
+    /** The default map factory. */
+    public static class DefaultToMapFactory<K, V> implements Func0<Map<K, V>> {
+        @Override
+        public Map<K, V> call() {
+            return new HashMap<K, V>();
+        }
+    }
+    /**
+     * Maps the elements of the source observable into a java.util.Map instance
+     * returned by the mapFactory function by using the keySelector and
+     * valueSelector.
+     * @param <T> the source's value type
+     * @param <K> the key type
+     * @param <V> the value type
+     */
+    public static class ToMap<T, K, V> implements OnSubscribeFunc<Map<K, V>> {
+        /** The source. */
+        private final Observable<T> source;
+        /** Key extractor. */
+        private final Func1<? super T, ? extends K> keySelector;
+        /** Value extractor. */
+        private final Func1<? super T, ? extends V> valueSelector;
+        /** Map factory. */
+        private final Func0<? extends Map<K, V>> mapFactory;
+        public ToMap(
+                Observable<T> source,
+                Func1<? super T, ? extends K> keySelector,
+                Func1<? super T, ? extends V> valueSelector,
+                Func0<? extends Map<K, V>> mapFactory
+                ) {
+            this.source = source;
+            this.keySelector = keySelector;
+            this.valueSelector = valueSelector;
+            this.mapFactory = mapFactory;
+            
+        }
+        @Override
+        public Subscription onSubscribe(Observer<? super Map<K, V>> t1) {
+            Map<K, V> map;
+            try {
+                map = mapFactory.call();
+            } catch (Throwable t) {
+                t1.onError(t);
+                return Subscriptions.empty();
+            }
+            return source.subscribe(new ToMapObserver<K, V, T>(
+                    t1, keySelector, valueSelector, map));
+        }
+        /**
+         * Observer that collects the source values of T into
+         * a map.
+         */
+        public static class ToMapObserver<K, V, T> implements Observer<T> {
+            /** The map. */
+            Map<K, V> map;
+            /** Key extractor. */
+            private final Func1<? super T, ? extends K> keySelector;
+            /** Value extractor. */
+            private final Func1<? super T, ? extends V> valueSelector;
+            /** The observer who is receiving the completed map. */
+            private final Observer<? super Map<K, V>> t1;
+            
+            public ToMapObserver(
+                    Observer<? super Map<K, V>> t1, 
+                    Func1<? super T, ? extends K> keySelector,
+                    Func1<? super T, ? extends V> valueSelector,
+                    Map<K, V> map) {
+                this.map = map;
+                this.t1 = t1;
+                this.keySelector = keySelector;
+                this.valueSelector = valueSelector;
+            }
+            @Override
+            public void onNext(T args) {
+                K key = keySelector.call(args);
+                V value = valueSelector.call(args);
+                map.put(key, value);
+            }
+            @Override
+            public void onError(Throwable e) {
+                map = null;
+                t1.onError(e);
+            }
+            @Override
+            public void onCompleted() {
+                Map<K, V> map0 = map;
+                map = null;
+                t1.onNext(map0);
+                t1.onCompleted();
+            }
+        }
+    }
+}
diff --git a/rxjava-core/src/main/java/rx/operators/OperationToMultimap.java b/rxjava-core/src/main/java/rx/operators/OperationToMultimap.java
new file mode 100644
index 0000000000..e9cf8d9413
--- /dev/null
+++ b/rxjava-core/src/main/java/rx/operators/OperationToMultimap.java
@@ -0,0 +1,206 @@
+/**
+ * Copyright 2013 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.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import rx.Observable;
+import rx.Observable.OnSubscribeFunc;
+import rx.Observer;
+import rx.Subscription;
+import rx.subscriptions.Subscriptions;
+import rx.util.functions.Func0;
+import rx.util.functions.Func1;
+import rx.util.functions.Functions;
+
+/**
+ * Maps the elements of the source observable into a multimap 
+ * (Map&lt;K, Collection&lt;V>>) where each
+ * key entry has a collection of the source's values.
+ * 
+ * @see <a href='https://github.com/Netflix/RxJava/issues/97'>Issue #97</a>
+ */
+public class OperationToMultimap {
+    /**
+     * ToMultimap with key selector, identitiy value selector,
+     * default HashMap factory and default ArrayList collection factory.
+     */
+    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>()
+        );
+    }
+
+    /**
+     * ToMultimap with key selector, custom value selector,
+     * default HashMap factory and default ArrayList collection factory.
+     */
+    public static <T, K, V> OnSubscribeFunc<Map<K, Collection<V>>> toMultimap(
+            Observable<T> source,
+            Func1<? super T, ? extends K> keySelector,
+            Func1<? super T, ? extends V> valueSelector
+    ) {
+        return new ToMultimap<T, K, V>(
+                source, keySelector, valueSelector,
+                new DefaultToMultimapFactory<K, V>(),
+                new DefaultMultimapCollectionFactory<K, V>()
+        );
+    }
+    /**
+     * ToMultimap with key selector, custom value selector,
+     * custom Map factory and default ArrayList collection factory.
+     */
+    public static <T, K, V> 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
+    ) {
+        return new ToMultimap<T, K, V>(
+                source, keySelector, valueSelector,
+                mapFactory,
+                new DefaultMultimapCollectionFactory<K, V>()
+        );
+    }
+    /**
+     * ToMultimap with key selector, custom value selector,
+     * custom Map factory and custom collection factory.
+     */
+    public static <T, K, V> 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
+    ) {
+        return new ToMultimap<T, K, V>(
+                source, keySelector, valueSelector,
+                mapFactory,
+                collectionFactory
+        );
+    }
+    /**
+     * The default multimap factory returning a HashMap.
+     */
+    public static class DefaultToMultimapFactory<K, V> implements Func0<Map<K, Collection<V>>> {
+        @Override
+        public Map<K, Collection<V>> call() {
+            return new HashMap<K, Collection<V>>();
+        }
+    }
+    /**
+     * The default collection factory for a key in the multimap returning
+     * an ArrayList independent of the key.
+     */
+    public static class DefaultMultimapCollectionFactory<K, V>
+    implements Func1<K, Collection<V>> {
+        @Override
+        public Collection<V> call(K t1) {
+            return new ArrayList<V>();
+        }
+    }
+    /**
+     * Maps the elements of the source observable int a multimap customized
+     * by various selectors and factories.
+     */
+    public static class ToMultimap<T, K, V> implements OnSubscribeFunc<Map<K, Collection<V>>> {
+        private final Observable<T> source;
+        private final Func1<? super T, ? extends K> keySelector;
+        private final Func1<? super T, ? extends V> valueSelector;
+        private final Func0<? extends Map<K, Collection<V>>> mapFactory;
+        private final Func1<? super K, ? extends Collection<V>> collectionFactory;
+        public 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
+        ) {
+            this.source = source;
+            this.keySelector = keySelector;
+            this.valueSelector = valueSelector;
+            this.mapFactory = mapFactory;
+            this.collectionFactory = collectionFactory;
+        }
+        @Override
+        public Subscription onSubscribe(Observer<? super Map<K, Collection<V>>> t1) {
+            Map<K, Collection<V>> map;
+            try {
+                map = mapFactory.call();
+            } catch (Throwable t) {
+                t1.onError(t);
+                return Subscriptions.empty();
+            }
+            return source.subscribe(new ToMultimapObserver<T, K, V>(
+                    t1, keySelector, valueSelector, map, collectionFactory
+            ));
+        }
+        /**
+         * Observer that collects the source values of Ts into a multimap.
+         */
+        public static class ToMultimapObserver<T, K, V> implements Observer<T> {
+            private final Func1<? super T, ? extends K> keySelector;
+            private final Func1<? super T, ? extends V> valueSelector;
+            private final Func1<? super K, ? extends Collection<V>> collectionFactory;
+            private Map<K, Collection<V>> map;
+            private Observer<? super Map<K, Collection<V>>> t1;
+            public ToMultimapObserver(
+                Observer<? super Map<K, Collection<V>>> t1,
+                Func1<? super T, ? extends K> keySelector,
+                Func1<? super T, ? extends V> valueSelector,
+                Map<K, Collection<V>> map,
+                Func1<? super K, ? extends Collection<V>> collectionFactory
+            ) {
+                this.t1 = t1;
+                this.keySelector = keySelector;
+                this.valueSelector = valueSelector;
+                this.collectionFactory = collectionFactory;
+                this.map = map;
+            }
+            @Override
+            public void onNext(T args) {
+                K key = keySelector.call(args);
+                V value = valueSelector.call(args);
+                Collection<V> collection = map.get(key);
+                if (collection == null) {
+                    collection = collectionFactory.call(key);
+                    map.put(key, collection);
+                }
+                collection.add(value);
+            }
+            @Override
+            public void onError(Throwable e) {
+                map = null;
+                t1.onError(e);
+            }
+            @Override
+            public void onCompleted() {
+                Map<K, Collection<V>> map0 = map;
+                map = null;
+                t1.onNext(map0);
+                t1.onCompleted();
+            }
+        }
+    }
+}
diff --git a/rxjava-core/src/test/java/rx/operators/OperationToMapTest.java b/rxjava-core/src/test/java/rx/operators/OperationToMapTest.java
new file mode 100644
index 0000000000..ab1a9fb9de
--- /dev/null
+++ b/rxjava-core/src/test/java/rx/operators/OperationToMapTest.java
@@ -0,0 +1,215 @@
+/**
+ * Copyright 2013 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.LinkedHashMap;
+import java.util.Map;
+import org.junit.Test;
+import org.junit.Before;
+import org.mockito.*;
+import static org.mockito.Mockito.*;
+import rx.Observable;
+import rx.Observer;
+import rx.util.functions.Func0;
+import rx.util.functions.Func1;
+import rx.util.functions.Functions;
+
+public class OperationToMapTest {
+    @Mock
+    Observer<Object> objectObserver;
+    
+    @Before
+    public void before() {
+        MockitoAnnotations.initMocks(this);
+    }
+    Func1<String, Integer> lengthFunc = new Func1<String, Integer>() {
+        @Override
+        public Integer call(String t1) {
+            return t1.length();
+        }
+    };
+    Func1<String, String> duplicate = new Func1<String, String>() {
+        @Override
+        public String call(String t1) {
+            return t1 + t1;
+        }
+    };
+    @Test
+    public void testToMap() {
+        Observable<String> source = Observable.from("a", "bb", "ccc", "dddd");
+        
+
+        Observable<Map<Integer, String>> mapped = Observable.create(OperationToMap.toMap(source, lengthFunc));
+        
+        Map<Integer, String> expected = new HashMap<Integer, String>();
+        expected.put(1, "a");
+        expected.put(2, "bb");
+        expected.put(3, "ccc");
+        expected.put(4, "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 testToMapWithValueSelector() {
+        Observable<String> source = Observable.from("a", "bb", "ccc", "dddd");
+        
+        Observable<Map<Integer, String>> mapped = Observable.create(OperationToMap.toMap(source, lengthFunc, duplicate));
+        
+        Map<Integer, String> expected = new HashMap<Integer, String>();
+        expected.put(1, "aa");
+        expected.put(2, "bbbb");
+        expected.put(3, "cccccc");
+        expected.put(4, "dddddddd");
+
+        mapped.subscribe(objectObserver);
+
+        verify(objectObserver, never()).onError(any(Throwable.class));
+        verify(objectObserver, times(1)).onNext(expected);
+        verify(objectObserver, times(1)).onCompleted();
+    }
+
+    @Test 
+    public void testToMapWithError() {
+        Observable<String> source = Observable.from("a", "bb", "ccc", "dddd");
+        
+        Func1<String, Integer> lengthFuncErr = new Func1<String, Integer>() {
+            @Override
+            public Integer call(String t1) {
+                if ("bb".equals(t1)) {
+                    throw new RuntimeException("Forced Failure");
+                }
+                return t1.length();
+            }
+        };
+        Observable<Map<Integer, String>> mapped = Observable.create(OperationToMap.toMap(source, lengthFuncErr));
+        
+        Map<Integer, String> expected = new HashMap<Integer, String>();
+        expected.put(1, "a");
+        expected.put(2, "bb");
+        expected.put(3, "ccc");
+        expected.put(4, "dddd");
+        
+        mapped.subscribe(objectObserver);
+        
+        verify(objectObserver, never()).onNext(expected);
+        verify(objectObserver, never()).onCompleted();
+        verify(objectObserver, times(1)).onError(any(Throwable.class));
+
+    }
+
+    @Test
+    public void testToMapWithErrorInValueSelector() {
+        Observable<String> source = Observable.from("a", "bb", "ccc", "dddd");
+        
+        Func1<String, String> duplicateErr = new Func1<String, String>() {
+            @Override
+            public String call(String t1) {
+                if ("bb".equals(t1)) {
+                    throw new RuntimeException("Forced failure");
+                }
+                return t1 + t1;
+            }
+        };
+        
+        Observable<Map<Integer, String>> mapped = Observable.create(OperationToMap.toMap(source, lengthFunc, duplicateErr));
+        
+        Map<Integer, String> expected = new HashMap<Integer, String>();
+        expected.put(1, "aa");
+        expected.put(2, "bbbb");
+        expected.put(3, "cccccc");
+        expected.put(4, "dddddddd");
+        
+        mapped.subscribe(objectObserver);
+        
+        verify(objectObserver, never()).onNext(expected);
+        verify(objectObserver, never()).onCompleted();
+        verify(objectObserver, times(1)).onError(any(Throwable.class));
+
+    }
+
+    @Test
+    public void testToMapWithFactory() {
+        Observable<String> source = Observable.from("a", "bb", "ccc", "dddd");
+        
+        Func0<Map<Integer, String>> mapFactory = new Func0<Map<Integer, String>>() {
+            @Override
+            public Map<Integer, String> call() {
+                return new LinkedHashMap<Integer, String>() {
+                    @Override
+                    protected boolean removeEldestEntry(Map.Entry<Integer, String> eldest) {
+                        return size() > 3;
+                    }
+                };
+            }
+        };
+        
+        Func1<String, Integer> lengthFunc = new Func1<String, Integer>() {
+            @Override
+            public Integer call(String t1) {
+                return t1.length();
+            }
+        };
+        Observable<Map<Integer, String>> mapped = Observable.create(OperationToMap.toMap(source, lengthFunc, Functions.<String>identity(), mapFactory));
+        
+        Map<Integer, String> expected = new LinkedHashMap<Integer, String>();
+        expected.put(2, "bb");
+        expected.put(3, "ccc");
+        expected.put(4, "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 testToMapWithErrorThrowingFactory() {
+        Observable<String> source = Observable.from("a", "bb", "ccc", "dddd");
+        
+        Func0<Map<Integer, String>> mapFactory = new Func0<Map<Integer, String>>() {
+            @Override
+            public Map<Integer, String> call() {
+                throw new RuntimeException("Forced failure");
+            }
+        };
+        
+        Func1<String, Integer> lengthFunc = new Func1<String, Integer>() {
+            @Override
+            public Integer call(String t1) {
+                return t1.length();
+            }
+        };
+        Observable<Map<Integer, String>> mapped = Observable.create(OperationToMap.toMap(source, lengthFunc, Functions.<String>identity(), mapFactory));
+        
+        Map<Integer, String> expected = new LinkedHashMap<Integer, String>();
+        expected.put(2, "bb");
+        expected.put(3, "ccc");
+        expected.put(4, "dddd");
+
+        mapped.subscribe(objectObserver);
+
+        verify(objectObserver, never()).onNext(expected);
+        verify(objectObserver, never()).onCompleted();
+        verify(objectObserver, times(1)).onError(any(Throwable.class));
+    }
+
+}
diff --git a/rxjava-core/src/test/java/rx/operators/OperationToMultimapTest.java b/rxjava-core/src/test/java/rx/operators/OperationToMultimapTest.java
new file mode 100644
index 0000000000..f715e69c61
--- /dev/null
+++ b/rxjava-core/src/test/java/rx/operators/OperationToMultimapTest.java
@@ -0,0 +1,250 @@
+/*
+ * To change this license header, choose License Headers in Project Properties.
+ * To change this template file, choose Tools | Templates
+ * and open the template in the editor.
+ */
+
+package rx.operators;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import org.junit.Test;
+import org.junit.Before;
+import org.mockito.*;
+import static org.mockito.Mockito.*;
+import rx.Observable;
+import rx.Observer;
+import rx.operators.OperationToMultimap.DefaultMultimapCollectionFactory;
+import rx.operators.OperationToMultimap.DefaultToMultimapFactory;
+import rx.util.functions.Func0;
+import rx.util.functions.Func1;
+import rx.util.functions.Functions;
+public class OperationToMultimapTest {
+    @Mock
+    Observer<Object> objectObserver;
+    
+    @Before
+    public void before() {
+        MockitoAnnotations.initMocks(this);
+    }
+    Func1<String, Integer> lengthFunc = new Func1<String, Integer>() {
+        @Override
+        public Integer call(String t1) {
+            return t1.length();
+        }
+    };
+    Func1<String, String> duplicate = new Func1<String, String>() {
+        @Override
+        public String call(String t1) {
+            return t1 + t1;
+        }
+    };
+    @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();
+    } 
+}