1.x: fix toMap and toMultimap not handling exceptions of the callbacks

This PR adds the usual try-catch around callback invocations in `toMap`
and `toMultimap`.

Related #3555.

Author: akarnokd

src/main/java/rx/internal/operators/OperatorToMap.java

@@ -21,8 +21,10 @@
 
 import rx.Observable.Operator;
 import rx.Subscriber;
+import rx.exceptions.Exceptions;
 import rx.functions.Func0;
 import rx.functions.Func1;
+import rx.observers.Subscribers;
 
 /**
  * Maps the elements of the source observable into a java.util.Map instance and
@@ -75,9 +77,24 @@ public OperatorToMap(
 
     @Override
     public Subscriber<? super T> call(final Subscriber<? super Map<K, V>> subscriber) {
+        
+        Map<K, V> localMap;
+        
+        try {
+            localMap = mapFactory.call();
+        } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
+            subscriber.onError(ex);
+            Subscriber<? super T> parent = Subscribers.empty();
+            parent.unsubscribe();
+            return parent;
+        }
+        
+        final Map<K, V> fLocalMap = localMap;
+        
         return new Subscriber<T>(subscriber) {
 
-            private Map<K, V> map = mapFactory.call();
+            private Map<K, V> map = fLocalMap;
 
             @Override
             public void onStart() {
@@ -86,8 +103,18 @@ public void onStart() {
 
             @Override
             public void onNext(T v) {
-                K key = keySelector.call(v);
-                V value = valueSelector.call(v);
+                K key;
+                V value;
+
+                try {
+                    key = keySelector.call(v);
+                    value = valueSelector.call(v);
+                } catch (Throwable ex) {
+                    Exceptions.throwIfFatal(ex);
+                    subscriber.onError(ex);
+                    return;
+                }
+                
                 map.put(key, value);
             }
 

src/main/java/rx/internal/operators/OperatorToMultimap.java

@@ -22,9 +22,11 @@
 import java.util.Map;
 
 import rx.Observable.Operator;
+import rx.exceptions.Exceptions;
 import rx.Subscriber;
 import rx.functions.Func0;
 import rx.functions.Func1;
+import rx.observers.Subscribers;
 
 /**
  * Maps the elements of the source observable into a multimap
@@ -103,8 +105,24 @@ public OperatorToMultimap(
 
     @Override
     public Subscriber<? super T> call(final Subscriber<? super Map<K, Collection<V>>> subscriber) {
+        
+        Map<K, Collection<V>> localMap;
+        
+        try {
+            localMap = mapFactory.call();
+        } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
+            subscriber.onError(ex);
+            
+            Subscriber<? super T> parent = Subscribers.empty();
+            parent.unsubscribe();
+            return parent;
+        }
+        
+        final Map<K, Collection<V>> fLocalMap = localMap;
+        
         return new Subscriber<T>(subscriber) {
-            private Map<K, Collection<V>> map = mapFactory.call();
+            private Map<K, Collection<V>> map = fLocalMap;
 
             @Override
             public void onStart() {
@@ -113,11 +131,27 @@ public void onStart() {
 
             @Override
             public void onNext(T v) {
-                K key = keySelector.call(v);
-                V value = valueSelector.call(v);
+                K key;
+                V value;
+
+                try {
+                    key = keySelector.call(v);
+                    value = valueSelector.call(v);
+                } catch (Throwable ex) {
+                    Exceptions.throwIfFatal(ex);
+                    subscriber.onError(ex);
+                    return;
+                }
+                
                 Collection<V> collection = map.get(key);
                 if (collection == null) {
-                    collection = collectionFactory.call(key);
+                    try {
+                        collection = collectionFactory.call(key);
+                    } catch (Throwable ex) {
+                        Exceptions.throwIfFatal(ex);
+                        subscriber.onError(ex);
+                        return;
+                    }
                     map.put(key, collection);
                 }
                 collection.add(value);

src/test/java/rx/internal/operators/OperatorToMapTest.java

@@ -16,24 +16,19 @@
 package rx.internal.operators;
 
 import static org.mockito.Matchers.any;
-import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.*;
 
-import java.util.HashMap;
-import java.util.LinkedHashMap;
-import java.util.Map;
+import java.util.*;
 
-import org.junit.Before;
-import org.junit.Test;
-import org.mockito.Mock;
-import org.mockito.MockitoAnnotations;
+import org.junit.*;
+import org.mockito.*;
 
 import rx.Observable;
 import rx.Observer;
-import rx.functions.Func0;
-import rx.functions.Func1;
+import rx.exceptions.TestException;
+import rx.functions.*;
 import rx.internal.util.UtilityFunctions;
+import rx.observers.TestSubscriber;
 
 public class OperatorToMapTest {
     @Mock
@@ -224,4 +219,66 @@ public Integer call(String t1) {
         verify(objectObserver, times(1)).onError(any(Throwable.class));
     }
 
+    @Test
+    public void testKeySelectorThrows() {
+        TestSubscriber<Object> ts = TestSubscriber.create();
+        
+        Observable.just(1, 2).toMap(new Func1<Integer, Integer>() {
+            @Override
+            public Integer call(Integer v) {
+                throw new TestException();
+            }
+        }).subscribe(ts);
+        
+        ts.assertError(TestException.class);
+        ts.assertNoValues();
+        ts.assertNotCompleted();
+    }
+    
+    @Test
+    public void testValueSelectorThrows() {
+        TestSubscriber<Object> ts = TestSubscriber.create();
+        
+        Observable.just(1, 2).toMap(new Func1<Integer, Integer>() {
+            @Override
+            public Integer call(Integer v) {
+                return v;
+            }
+        }, new Func1<Integer, Integer>() {
+            @Override
+            public Integer call(Integer v) {
+                throw new TestException();
+            }
+        }).subscribe(ts);
+        
+        ts.assertError(TestException.class);
+        ts.assertNoValues();
+        ts.assertNotCompleted();
+    }
+    
+    @Test
+    public void testMapFactoryThrows() {
+        TestSubscriber<Object> ts = TestSubscriber.create();
+        
+        Observable.just(1, 2).toMap(new Func1<Integer, Integer>() {
+            @Override
+            public Integer call(Integer v) {
+                return v;
+            }
+        }, new Func1<Integer, Integer>() {
+            @Override
+            public Integer call(Integer v) {
+                return v;
+            }
+        }, new Func0<Map<Integer, Integer>>() {
+            @Override
+            public Map<Integer, Integer> call() {
+                throw new TestException();
+            }
+        }).subscribe(ts);
+        
+        ts.assertError(TestException.class);
+        ts.assertNoValues();
+        ts.assertNotCompleted();
+    }
 }

src/test/java/rx/internal/operators/OperatorToMultimapTest.java

@@ -36,11 +36,13 @@
 
 import rx.Observable;
 import rx.Observer;
+import rx.exceptions.TestException;
 import rx.functions.Func0;
 import rx.functions.Func1;
 import rx.internal.operators.OperatorToMultimap.DefaultMultimapCollectionFactory;
 import rx.internal.operators.OperatorToMultimap.DefaultToMultimapFactory;
 import rx.internal.util.UtilityFunctions;
+import rx.observers.TestSubscriber;
 
 public class OperatorToMultimapTest {
     @Mock
@@ -269,4 +271,98 @@ public Collection<String> call(Integer t1) {
         verify(objectObserver, never()).onNext(expected);
         verify(objectObserver, never()).onCompleted();
     }
+    
+    @Test
+    public void testKeySelectorThrows() {
+        TestSubscriber<Object> ts = TestSubscriber.create();
+        
+        Observable.just(1, 2).toMultimap(new Func1<Integer, Integer>() {
+            @Override
+            public Integer call(Integer v) {
+                throw new TestException();
+            }
+        }).subscribe(ts);
+        
+        ts.assertError(TestException.class);
+        ts.assertNoValues();
+        ts.assertNotCompleted();
+    }
+    
+    @Test
+    public void testValueSelectorThrows() {
+        TestSubscriber<Object> ts = TestSubscriber.create();
+        
+        Observable.just(1, 2).toMultimap(new Func1<Integer, Integer>() {
+            @Override
+            public Integer call(Integer v) {
+                return v;
+            }
+        }, new Func1<Integer, Integer>() {
+            @Override
+            public Integer call(Integer v) {
+                throw new TestException();
+            }
+        }).subscribe(ts);
+        
+        ts.assertError(TestException.class);
+        ts.assertNoValues();
+        ts.assertNotCompleted();
+    }
+    
+    @Test
+    public void testMapFactoryThrows() {
+        TestSubscriber<Object> ts = TestSubscriber.create();
+        
+        Observable.just(1, 2).toMultimap(new Func1<Integer, Integer>() {
+            @Override
+            public Integer call(Integer v) {
+                return v;
+            }
+        }, new Func1<Integer, Integer>() {
+            @Override
+            public Integer call(Integer v) {
+                return v;
+            }
+        }, new Func0<Map<Integer, Collection<Integer>>>() {
+            @Override
+            public Map<Integer, Collection<Integer>> call() {
+                throw new TestException();
+            }
+        }).subscribe(ts);
+        
+        ts.assertError(TestException.class);
+        ts.assertNoValues();
+        ts.assertNotCompleted();
+    }
+    
+    @Test
+    public void testCollectionFactoryThrows() {
+        TestSubscriber<Object> ts = TestSubscriber.create();
+        
+        Observable.just(1, 2).toMultimap(new Func1<Integer, Integer>() {
+            @Override
+            public Integer call(Integer v) {
+                return v;
+            }
+        }, new Func1<Integer, Integer>() {
+            @Override
+            public Integer call(Integer v) {
+                return v;
+            }
+        }, new Func0<Map<Integer, Collection<Integer>>>() {
+            @Override
+            public Map<Integer, Collection<Integer>> call() {
+                return new HashMap<Integer, Collection<Integer>>();
+            }
+        }, new Func1<Integer, Collection<Integer>>() {
+            @Override
+            public Collection<Integer> call(Integer k) {
+                throw new TestException();
+            }
+        }).subscribe(ts);
+        
+        ts.assertError(TestException.class);
+        ts.assertNoValues();
+        ts.assertNotCompleted();
+    }
 }