Fixed upstream delivery counting and timer cancelling concurrency bug.

Author: akarnokd

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

@@ -69,6 +69,9 @@ static final class BufferSubscriber<T> extends Subscriber<T> {
         /** Tracks the downstream requested amounts. */
         final AtomicLong requested;
 
+        /** Tracks the upstream requested amounts. */
+        final AtomicLong upstreamRequested;
+        
         /** Holds onto the current timer. */
         final SerialSubscription timer;
 
@@ -92,6 +95,7 @@ public BufferSubscriber(Subscriber<? super List<T>> child, int size,
             this.worker = worker;
             this.timeRemaining = unit.toMillis(time);
             this.requested = new AtomicLong();
+            this.upstreamRequested = new AtomicLong();
             this.timer = new SerialSubscription();
             this.add(timer);
         }
@@ -103,10 +107,12 @@ public void setProducer(Producer producer) {
 
         @Override
         public void onNext(T t) {
-            if (requested.get() == 0) {
+            if (upstreamRequested.get() == 0) {
                 onError(new MissingBackpressureException());
                 return;
             }
+            upstreamRequested.decrementAndGet();
+            
             List<T> list;
             long r;
             long id;
@@ -187,9 +193,6 @@ void timeout() {
         void scheduleTimer(long r, long id, long delay) {
             if (r > 0 && r < Long.MAX_VALUE) {
                 timer.set(worker.schedule(new TimerAction(id), delay, unit));
-            } else 
-            if (r == 0) {
-                timer.set(Subscriptions.unsubscribed());
             }
         }
 
@@ -297,6 +300,17 @@ void handleRequested(long before, long request) {
                     timer.set(worker.schedulePeriodically(new PeriodicAction(), time, time, unit));
                 }
             }
+            for (;;) {
+                long r2 = upstreamRequested.get();
+                long u2 = r2 + elements;
+                if (u2 < 0) {
+                    u2 = Long.MAX_VALUE;
+                }
+                if (upstreamRequested.compareAndSet(r2, u2)) {
+                    break;
+                }
+            }
+            
             Producer p = producer;
             if (p != null) {
                 p.request(elements);

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

@@ -103,5 +103,40 @@ public void testSingleRequestWithTimeout() {
 
         ts.assertValue(Arrays.asList(1));
         ts.assertNoTerminalEvent();
+
+        ps.onNext(2);
+        ps.onNext(3);
+        
+        ts.assertValue(Arrays.asList(1));
+        ts.assertNoTerminalEvent();
+    }
+    @Test
+    public void testUpstreamOverflowBuffer() {
+        TestScheduler scheduler = new TestScheduler();
+        PublishSubject<Integer> ps = PublishSubject.create();
+        Observable<List<Integer>> source = ps.buffer(100, TimeUnit.MILLISECONDS, 10, scheduler);
+        
+        TestSubscriber<List<Integer>> ts = new TestSubscriber<List<Integer>>(0);
+        
+        source.subscribe(ts);
+        
+        ts.requestMore(1);
+        
+        ps.onNext(1);
+        
+        ts.assertNoValues();
+        ts.assertNoTerminalEvent();
+        
+        scheduler.advanceTimeBy(100, TimeUnit.MILLISECONDS);
+        
+        ts.assertValue(Arrays.asList(1));
+        ts.assertNoTerminalEvent();
+
+        for (int i = 2; i <= 11; i++) {
+            ps.onNext(i);
+        }
+        
+        ts.assertValue(Arrays.asList(1));
+        ts.assertError(MissingBackpressureException.class);
     }
 }