Merge pull request #3181 from akarnokd/MergeNotificationFix

MapNotification producer NPE fix

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

@@ -19,16 +19,12 @@
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.atomic.AtomicLong;
 
+import rx.*;
 import rx.Observable.Operator;
-import rx.Producer;
-import rx.Subscriber;
-import rx.Subscription;
-import rx.exceptions.MissingBackpressureException;
-import rx.exceptions.OnErrorThrowable;
-import rx.functions.Func0;
-import rx.functions.Func1;
-import rx.internal.util.unsafe.SpscArrayQueue;
-import rx.internal.util.unsafe.UnsafeAccess;
+import rx.exceptions.*;
+import rx.functions.*;
+import rx.internal.producers.ProducerArbiter;
+import rx.internal.util.unsafe.*;
 
 /**
  * Applies a function of your choosing to every item emitted by an {@code Observable}, and emits the results of
@@ -50,44 +46,60 @@ public OperatorMapNotification(Func1<? super T, ? extends R> onNext, Func1<? sup
 
     @Override
     public Subscriber<? super T> call(final Subscriber<? super R> o) {
-        Subscriber<T> subscriber = new Subscriber<T>() {
-            SingleEmitter<R> emitter;
-            @Override
-            public void setProducer(Producer producer) {
-                emitter = new SingleEmitter<R>(o, producer, this);
-                o.setProducer(emitter);
-            }
-
-            @Override
-            public void onCompleted() {
-                try {
-                    emitter.offerAndComplete(onCompleted.call());
-                } catch (Throwable e) {
-                    o.onError(e);
-                }
-            }
+        final ProducerArbiter pa = new ProducerArbiter();
+
+        MapNotificationSubscriber subscriber = new MapNotificationSubscriber(pa, o);
+        o.add(subscriber);
+        subscriber.init();
+        return subscriber;
+    }
+
+    final class MapNotificationSubscriber extends Subscriber<T> {
+        private final Subscriber<? super R> o;
+        private final ProducerArbiter pa;
+        final SingleEmitter<R> emitter;
+
+        private MapNotificationSubscriber(ProducerArbiter pa, Subscriber<? super R> o) {
+            this.pa = pa;
+            this.o = o;
+            this.emitter = new SingleEmitter<R>(o, pa, this);
+        }
+
+        void init() {
+            o.setProducer(emitter);
+        }
 
-            @Override
-            public void onError(Throwable e) {
-                try {
-                    emitter.offerAndComplete(onError.call(e));
-                } catch (Throwable e2) {
-                    o.onError(e);
-                }
+        @Override
+        public void setProducer(Producer producer) {
+            pa.setProducer(producer);
+        }
+
+        @Override
+        public void onCompleted() {
+            try {
+                emitter.offerAndComplete(onCompleted.call());
+            } catch (Throwable e) {
+                o.onError(e);
             }
+        }
 
-            @Override
-            public void onNext(T t) {
-                try {
-                    emitter.offer(onNext.call(t));
-                } catch (Throwable e) {
-                    o.onError(OnErrorThrowable.addValueAsLastCause(e, t));
-                }
+        @Override
+        public void onError(Throwable e) {
+            try {
+                emitter.offerAndComplete(onError.call(e));
+            } catch (Throwable e2) {
+                o.onError(e);
             }
+        }
 
-        };
-        o.add(subscriber);
-        return subscriber;
+        @Override
+        public void onNext(T t) {
+            try {
+                emitter.offer(onNext.call(t));
+            } catch (Throwable e) {
+                o.onError(OnErrorThrowable.addValueAsLastCause(e, t));
+            }
+        }
     }
     static final class SingleEmitter<T> extends AtomicLong implements Producer, Subscription {
         /** */

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

@@ -0,0 +1,55 @@
+/**
+ * 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.internal.operators;
+
+import org.junit.Test;
+
+import rx.Observable;
+import rx.functions.*;
+import rx.observers.TestSubscriber;
+
+public class OperatorMapNotificationTest {
+    @Test
+    public void testJust() {
+        TestSubscriber<Object> ts = TestSubscriber.create();
+        Observable.just(1)
+        .flatMap(
+                new Func1<Integer, Observable<Object>>() {
+                    @Override
+                    public Observable<Object> call(Integer item) {
+                        return Observable.just((Object)(item + 1));
+                    }
+                },
+                new Func1<Throwable, Observable<Object>>() {
+                    @Override
+                    public Observable<Object> call(Throwable e) {
+                        return Observable.error(e);
+                    }
+                },
+                new Func0<Observable<Object>>() {
+                    @Override
+                    public Observable<Object> call() {
+                        return Observable.never();
+                    }
+                }
+        ).subscribe(ts);
+
+        ts.assertNoErrors();
+        ts.assertNotCompleted();
+        ts.assertValue(2);
+    }
+}