Null check for BufferExactBoundedObserver (#6499)

* Null check for BufferExactBoundedObserver

Other variants contain this Null check already, e.g. BufferExactUnboundedObserver
It is causing 0.1% crashes in our production app.

* ObservableBufferTimed.BufferExactBoundedObserver - failing supplier unit test
FlowableBufferTimed.BufferExactBoundedSubscriber - failing supplier fix + unit test

* Better Unit tests for FlowableBufferTimed and BufferExactBoundedSubscriber NPE issue
Reverted

Author: JanKn

src/main/java/io/reactivex/internal/operators/flowable/FlowableBufferTimed.java

@@ -501,13 +501,14 @@ public void onComplete() {
                 buffer = null;
             }
 
-            queue.offer(b);
-            done = true;
-            if (enter()) {
-                QueueDrainHelper.drainMaxLoop(queue, downstream, false, this, this);
+            if (b != null) {
+                queue.offer(b);
+                done = true;
+                if (enter()) {
+                    QueueDrainHelper.drainMaxLoop(queue, downstream, false, this, this);
+                }
+                w.dispose();
             }
-
-            w.dispose();
         }
 
         @Override

src/main/java/io/reactivex/internal/operators/observable/ObservableBufferTimed.java

@@ -504,10 +504,12 @@ public void onComplete() {
                 buffer = null;
             }
 
-            queue.offer(b);
-            done = true;
-            if (enter()) {
-                QueueDrainHelper.drainLoop(queue, downstream, false, this, this);
+            if (b != null) {
+                queue.offer(b);
+                done = true;
+                if (enter()) {
+                    QueueDrainHelper.drainLoop(queue, downstream, false, this, this);
+                }
             }
         }
 

src/test/java/io/reactivex/internal/operators/flowable/FlowableBufferTest.java

@@ -2769,4 +2769,19 @@ public void timedSizeBufferAlreadyCleared() {
 
         sub.run();
     }
+
+    @Test
+    public void bufferExactFailingSupplier() {
+        Flowable.empty()
+                .buffer(1, TimeUnit.SECONDS, Schedulers.computation(), 10, new Callable<List<Object>>() {
+                    @Override
+                    public List<Object> call() throws Exception {
+                        throw new TestException();
+                    }
+                }, false)
+                .test()
+                .awaitDone(1, TimeUnit.SECONDS)
+                .assertFailure(TestException.class)
+        ;
+    }
 }

src/test/java/io/reactivex/internal/operators/observable/ObservableBufferTest.java

@@ -2136,4 +2136,19 @@ public ObservableSource<List<Object>> apply(Observable<Object> o)
             }
         });
     }
+
+    @Test
+    public void bufferExactFailingSupplier() {
+        Observable.empty()
+                .buffer(1, TimeUnit.SECONDS, Schedulers.computation(), 10, new Callable<List<Object>>() {
+                    @Override
+                    public List<Object> call() throws Exception {
+                        throw new TestException();
+                    }
+                }, false)
+                .test()
+                .awaitDone(1, TimeUnit.SECONDS)
+                .assertFailure(TestException.class)
+        ;
+    }
 }