toMap - prevent multiple terminal events

Author: davidmoten

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

@@ -23,6 +23,7 @@
 import rx.exceptions.Exceptions;
 import rx.functions.*;
 import rx.observers.Subscribers;
+import rx.plugins.RxJavaHooks;
 
 /**
  * Maps the elements of the source observable into a java.util.Map instance and
@@ -100,6 +101,8 @@ public Subscriber<? super T> call(final Subscriber<? super Map<K, V>> subscriber
         return new Subscriber<T>(subscriber) {
 
             private Map<K, V> map = fLocalMap;
+            
+            boolean done;
 
             @Override
             public void onStart() {
@@ -108,14 +111,17 @@ public void onStart() {
 
             @Override
             public void onNext(T v) {
+                if (done) {
+                    return;
+                }
                 K key;
                 V value;
 
                 try {
                     key = keySelector.call(v);
                     value = valueSelector.call(v);
                 } catch (Throwable ex) {
-                    Exceptions.throwOrReport(ex, subscriber);
+                    Exceptions.throwOrReport(ex, this);
                     return;
                 }
 
@@ -124,12 +130,21 @@ public void onNext(T v) {
 
             @Override
             public void onError(Throwable e) {
-                map = null;
-                subscriber.onError(e);
+                if (!done) {
+                    done = true;
+                    map = null;
+                    subscriber.onError(e);
+                } else {
+                    RxJavaHooks.onError(e);
+                }
             }
 
             @Override
             public void onCompleted() {
+                if (done) {
+                    return;
+                }
+                done = true;
                 Map<K, V> map0 = map;
                 map = null;
                 subscriber.onNext(map0);

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

@@ -15,20 +15,35 @@
  */
 package rx.internal.operators;
 
+import static org.junit.Assert.assertEquals;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.verify;
 
-import java.util.*;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
 
-import org.junit.*;
-import org.mockito.*;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
 
 import rx.Observable;
+import rx.Observable.OnSubscribe;
 import rx.Observer;
+import rx.Producer;
+import rx.Subscriber;
 import rx.exceptions.TestException;
-import rx.functions.*;
+import rx.functions.Action1;
+import rx.functions.Func0;
+import rx.functions.Func1;
 import rx.internal.util.UtilityFunctions;
 import rx.observers.TestSubscriber;
+import rx.plugins.RxJavaHooks;
 
 public class OperatorToMapTest {
     @Mock
@@ -281,4 +296,111 @@ public Map<Integer, Integer> call() {
         ts.assertNoValues();
         ts.assertNotCompleted();
     }
+    
+    @Test
+    public void testFactoryFailureDoesNotAllowErrorAndCompletedEmissions() {
+        TestSubscriber<Map<Integer, Integer>> ts = TestSubscriber.create(0);
+        final RuntimeException e = new RuntimeException();
+        Observable.create(new OnSubscribe<Integer>() {
+
+            @Override
+            public void call(final Subscriber<? super Integer> sub) {
+                sub.setProducer(new Producer() {
+
+                    @Override
+                    public void request(long n) {
+                        if (n > 1) {
+                            sub.onNext(1);
+                            sub.onCompleted();
+                        }
+                    }
+                });
+            }
+        }).toMap(new Func1<Integer,Integer>() {
+
+            @Override
+            public Integer call(Integer t) {
+                throw e;
+            }
+        }).unsafeSubscribe(ts);
+        ts.assertNoValues();
+        ts.assertError(e);
+        ts.assertNotCompleted();
+    }
+    
+    @Test
+    public void testFactoryFailureDoesNotAllowTwoErrorEmissions() {
+        try {
+            final List<Throwable> list = new CopyOnWriteArrayList<Throwable>();
+            RxJavaHooks.setOnError(new Action1<Throwable>() {
+
+                @Override
+                public void call(Throwable t) {
+                    list.add(t);
+                }
+            });
+            TestSubscriber<Map<Integer, Integer>> ts = TestSubscriber.create(0);
+            final RuntimeException e1 = new RuntimeException();
+            final RuntimeException e2 = new RuntimeException();
+            Observable.create(new OnSubscribe<Integer>() {
+
+                @Override
+                public void call(final Subscriber<? super Integer> sub) {
+                    sub.setProducer(new Producer() {
+
+                        @Override
+                        public void request(long n) {
+                            if (n > 1) {
+                                sub.onNext(1);
+                                sub.onError(e2);
+                            }
+                        }
+                    });
+                }
+            }).toMap(new Func1<Integer, Integer>() {
+
+                @Override
+                public Integer call(Integer t) {
+                    throw e1;
+                }
+            }).unsafeSubscribe(ts);
+            ts.assertNoValues();
+            assertEquals(Arrays.asList(e1), ts.getOnErrorEvents());
+            assertEquals(Arrays.asList(e2), list);
+            ts.assertNotCompleted();
+        } finally {
+            RxJavaHooks.setOnError(null);
+        }
+    }
+    
+    @Test
+    public void testFactoryFailureDoesNotAllowErrorThenOnNextEmissions() {
+        TestSubscriber<Map<Integer, Integer>> ts = TestSubscriber.create(0);
+        final RuntimeException e = new RuntimeException();
+        Observable.create(new OnSubscribe<Integer>() {
+
+            @Override
+            public void call(final Subscriber<? super Integer> sub) {
+                sub.setProducer(new Producer() {
+
+                    @Override
+                    public void request(long n) {
+                        if (n > 1) {
+                            sub.onNext(1);
+                            sub.onNext(2);
+                        }
+                    }
+                });
+            }
+        }).toMap(new Func1<Integer,Integer>() {
+
+            @Override
+            public Integer call(Integer t) {
+                throw e;
+            }
+        }).unsafeSubscribe(ts);
+        ts.assertNoValues();
+        ts.assertError(e);
+        ts.assertNotCompleted();
+    }
 }