From 47eb3061b7da7fab6e1c87b242fa604a6c0df0e7 Mon Sep 17 00:00:00 2001
From: akarnokd <akarnokd@gmail.com>
Date: Tue, 29 Mar 2016 17:33:24 +0200
Subject: [PATCH] 1.x: fix merge/flatMap crashing on an inner scalar of null

---
 .../rx/internal/operators/OperatorMerge.java  |  2 +-
 .../internal/operators/OperatorMergeTest.java | 21 +++++++++++++++++--
 2 files changed, 20 insertions(+), 3 deletions(-)

diff --git a/src/main/java/rx/internal/operators/OperatorMerge.java b/src/main/java/rx/internal/operators/OperatorMerge.java
index a9c7b86b09..2aede6d9d7 100644
--- a/src/main/java/rx/internal/operators/OperatorMerge.java
+++ b/src/main/java/rx/internal/operators/OperatorMerge.java
@@ -475,7 +475,7 @@ protected void queueScalar(T value) {
                 }
                 this.queue = q;
             }
-            if (!q.offer(value)) {
+            if (!q.offer(nl.next(value))) {
                 unsubscribe();
                 onError(OnErrorThrowable.addValueAsLastCause(new MissingBackpressureException(), value));
                 return;
diff --git a/src/test/java/rx/internal/operators/OperatorMergeTest.java b/src/test/java/rx/internal/operators/OperatorMergeTest.java
index 2c40ac53d3..b93f32b580 100644
--- a/src/test/java/rx/internal/operators/OperatorMergeTest.java
+++ b/src/test/java/rx/internal/operators/OperatorMergeTest.java
@@ -28,10 +28,10 @@
 import org.mockito.*;
 
 import rx.*;
-import rx.Observable.OnSubscribe;
-import rx.Scheduler.Worker;
 import rx.Observable;
+import rx.Observable.OnSubscribe;
 import rx.Observer;
+import rx.Scheduler.Worker;
 import rx.functions.*;
 import rx.internal.util.RxRingBuffer;
 import rx.observers.TestSubscriber;
@@ -1353,4 +1353,21 @@ public void zeroMaxConcurrent() {
             assertEquals("maxConcurrent > 0 required but it was 0", e.getMessage());
         }
     }
+    
+    @Test
+    public void mergeJustNull() {
+        TestSubscriber<Integer> ts = new TestSubscriber<Integer>(0);
+        
+        Observable.range(1, 2).flatMap(new Func1<Integer, Observable<Integer>>() {
+            @Override
+            public Observable<Integer> call(Integer t) {
+                return Observable.just(null);
+            }
+        }).subscribe(ts);
+        
+        ts.requestMore(2);
+        ts.assertValues(null, null);
+        ts.assertNoErrors();
+        ts.assertCompleted();
+    }
 }