Merge pull request #3028 from akarnokd/DelayErrorCutaheadFix

Delay: error cut ahead was not properly serialized

Author: akarnokd

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

@@ -49,22 +49,36 @@ public Subscriber<? super T> call(final Subscriber<? super T> child) {
         final Worker worker = scheduler.createWorker();
         child.add(worker);
         return new Subscriber<T>(child) {
-
+            // indicates an error cut ahead
+            // accessed from the worker thread only
+            boolean done;
             @Override
             public void onCompleted() {
                 worker.schedule(new Action0() {
 
                     @Override
                     public void call() {
-                        child.onCompleted();
+                        if (!done) {
+                            done = true;
+                            child.onCompleted();
+                        }
                     }
 
                 }, delay, unit);
             }
 
             @Override
-            public void onError(Throwable e) {
-                child.onError(e);
+            public void onError(final Throwable e) {
+                worker.schedule(new Action0() {
+                    @Override
+                    public void call() {
+                        if (!done) {
+                            done = true;
+                            child.onError(e);
+                            worker.unsubscribe();
+                        }
+                    }
+                });
             }
 
             @Override
@@ -73,7 +87,9 @@ public void onNext(final T t) {
 
                     @Override
                     public void call() {
-                        child.onNext(t);
+                        if (!done) {
+                            child.onNext(t);
+                        }
                     }
 
                 }, delay, unit);

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

@@ -798,4 +798,27 @@ public Integer call(Integer t) {
         ts.assertNoErrors();
         assertEquals(RxRingBuffer.SIZE * 2, ts.getOnNextEvents().size());
     }
+    
+    @Test
+    public void testErrorRunsBeforeOnNext() {
+        TestScheduler test = Schedulers.test();
+        
+        PublishSubject<Integer> ps = PublishSubject.create();
+        
+        TestSubscriber<Integer> ts = TestSubscriber.create();
+        
+        ps.delay(1, TimeUnit.SECONDS, test).subscribe(ts);
+        
+        ps.onNext(1);
+        
+        test.advanceTimeBy(500, TimeUnit.MILLISECONDS);
+        
+        ps.onError(new TestException());
+        
+        test.advanceTimeBy(1, TimeUnit.SECONDS);
+        
+        ts.assertNoValues();
+        ts.assertError(TestException.class);
+        ts.assertNotCompleted();
+    }
 }