diff --git a/src/main/java/rx/internal/operators/OperatorThrottleFirst.java b/src/main/java/rx/internal/operators/OperatorThrottleFirst.java index fbd3c64d7d..f9093efa39 100644 --- a/src/main/java/rx/internal/operators/OperatorThrottleFirst.java +++ b/src/main/java/rx/internal/operators/OperatorThrottleFirst.java @@ -48,7 +48,7 @@ public void onStart() { @Override public void onNext(T v) { long now = scheduler.now(); - if (lastOnNext == -1 || now - lastOnNext >= timeInMilliseconds) { + if (lastOnNext == -1 || now < lastOnNext || now - lastOnNext >= timeInMilliseconds) { lastOnNext = now; subscriber.onNext(v); } diff --git a/src/test/java/rx/internal/operators/OperatorThrottleFirstTest.java b/src/test/java/rx/internal/operators/OperatorThrottleFirstTest.java index 4908efc534..f0fc6527a6 100644 --- a/src/test/java/rx/internal/operators/OperatorThrottleFirstTest.java +++ b/src/test/java/rx/internal/operators/OperatorThrottleFirstTest.java @@ -16,26 +16,18 @@ package rx.internal.operators; import static org.mockito.Matchers.any; -import static org.mockito.Mockito.inOrder; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.verifyNoMoreInteractions; +import static org.mockito.Mockito.*; import java.util.concurrent.TimeUnit; -import org.junit.Before; -import org.junit.Test; +import org.junit.*; import org.mockito.InOrder; -import rx.Observable; +import rx.*; import rx.Observable.OnSubscribe; -import rx.Observer; -import rx.Scheduler; -import rx.Subscriber; import rx.exceptions.TestException; import rx.functions.Action0; -import rx.observers.TestSubscriber; +import rx.observers.*; import rx.schedulers.TestScheduler; import rx.subjects.PublishSubject; @@ -214,4 +206,28 @@ public void throttleWithTestSchedulerTimeOfZero() { verify(observer).onCompleted(); verifyNoMoreInteractions(observer); } + + @Test + public void nowDrift() { + TestScheduler s = new TestScheduler(); + s.advanceTimeBy(2, TimeUnit.SECONDS); + + PublishSubject o = PublishSubject.create(); + + AssertableSubscriber as = o.throttleFirst(500, TimeUnit.MILLISECONDS, s) + .test(); + + o.onNext(1); + s.advanceTimeBy(100, TimeUnit.MILLISECONDS); + o.onNext(2); + s.advanceTimeBy(100, TimeUnit.MILLISECONDS); + o.onNext(3); + s.advanceTimeBy(-1000, TimeUnit.MILLISECONDS); + o.onNext(4); + s.advanceTimeBy(100, TimeUnit.MILLISECONDS); + o.onNext(5); + o.onCompleted(); + + as.assertResult(1, 4); + } }