2.x: Fix MaybeTimber by using scheduler and unit

src/main/java/io/reactivex/internal/operators/completable/CompletableTimer.java

@@ -14,10 +14,15 @@
 package io.reactivex.internal.operators.completable;
 
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
 import io.reactivex.*;
-import io.reactivex.internal.disposables.SequentialDisposable;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.internal.disposables.DisposableHelper;
 
+/**
+ * Signals an {@code onCompleted} event after the specified delay
+ */
 public final class CompletableTimer extends Completable {
 
     final long delay;
@@ -30,20 +35,39 @@ public CompletableTimer(long delay, TimeUnit unit, Scheduler scheduler) {
         this.scheduler = scheduler;
     }
 
-
-
     @Override
     protected void subscribeActual(final CompletableObserver s) {
-        SequentialDisposable sd = new SequentialDisposable();
-        s.onSubscribe(sd);
-        if (!sd.isDisposed()) {
-            sd.replace(scheduler.scheduleDirect(new Runnable() {
-                @Override
-                public void run() {
-                    s.onComplete();
-                }
-            }, delay, unit));
-        }
+        TimerDisposable parent = new TimerDisposable(s);
+        s.onSubscribe(parent);
+        parent.setFuture(scheduler.scheduleDirect(parent, delay, unit));
     }
 
+    static final class TimerDisposable extends AtomicReference<Disposable> implements Disposable, Runnable {
+        /** */
+        private static final long serialVersionUID = 3167244060586201109L;
+        final CompletableObserver actual;
+
+        TimerDisposable(final CompletableObserver actual) {
+            this.actual = actual;
+        }
+
+        @Override
+        public void run() {
+            actual.onComplete();
+        }
+
+        @Override
+        public void dispose() {
+            DisposableHelper.dispose(this);
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return DisposableHelper.isDisposed(get());
+        }
+
+        void setFuture(Disposable d) {
+            DisposableHelper.replace(this, d);
+        }
+    }
 }

src/main/java/io/reactivex/internal/operators/maybe/MaybeTimer.java

@@ -21,7 +21,7 @@
 import io.reactivex.internal.disposables.DisposableHelper;
 
 /**
- * Signals a 0L after the specified delay
+ * Signals a {@code 0L} after the specified delay
  */
 public final class MaybeTimer extends Maybe<Long> {
 
@@ -38,20 +38,18 @@ public MaybeTimer(long delay, TimeUnit unit, Scheduler scheduler) {
     }
 
     @Override
-    protected void subscribeActual(MaybeObserver<? super Long> observer) {
+    protected void subscribeActual(final MaybeObserver<? super Long> observer) {
         TimerDisposable parent = new TimerDisposable(observer);
         observer.onSubscribe(parent);
-
-        parent.setFuture(scheduler.scheduleDirect(parent));
+        parent.setFuture(scheduler.scheduleDirect(parent, delay, unit));
     }
 
     static final class TimerDisposable extends AtomicReference<Disposable> implements Disposable, Runnable {
-
         /** */
         private static final long serialVersionUID = 2875964065294031672L;
         final MaybeObserver<? super Long> actual;
 
-        public TimerDisposable(MaybeObserver<? super Long> actual) {
+        TimerDisposable(final MaybeObserver<? super Long> actual) {
             this.actual = actual;
         }
 

src/main/java/io/reactivex/internal/operators/single/SingleTimer.java

@@ -14,10 +14,15 @@
 package io.reactivex.internal.operators.single;
 
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
 import io.reactivex.*;
-import io.reactivex.internal.disposables.SequentialDisposable;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.internal.disposables.DisposableHelper;
 
+/**
+ * Signals a {@code 0L} after the specified delay
+ */
 public final class SingleTimer extends Single<Long> {
 
     final long delay;
@@ -32,16 +37,37 @@ public SingleTimer(long delay, TimeUnit unit, Scheduler scheduler) {
 
     @Override
     protected void subscribeActual(final SingleObserver<? super Long> s) {
-        SequentialDisposable sd = new SequentialDisposable();
+        TimerDisposable parent = new TimerDisposable(s);
+        s.onSubscribe(parent);
+        parent.setFuture(scheduler.scheduleDirect(parent, delay, unit));
+    }
 
-        s.onSubscribe(sd);
+    static final class TimerDisposable extends AtomicReference<Disposable> implements Disposable, Runnable {
+        /** */
+        private static final long serialVersionUID = 8465401857522493082L;
+        final SingleObserver<? super Long> actual;
 
-        sd.replace(scheduler.scheduleDirect(new Runnable() {
-            @Override
-            public void run() {
-                s.onSuccess(0L);
-            }
-        }, delay, unit));
-    }
+        TimerDisposable(final SingleObserver<? super Long> actual) {
+            this.actual = actual;
+        }
 
+        @Override
+        public void run() {
+            actual.onSuccess(0L);
+        }
+
+        @Override
+        public void dispose() {
+            DisposableHelper.dispose(this);
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return DisposableHelper.isDisposed(get());
+        }
+
+        void setFuture(Disposable d) {
+            DisposableHelper.replace(this, d);
+        }
+    }
 }

src/test/java/io/reactivex/completable/CompletableTimerTest.java

@@ -0,0 +1,50 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.completable;
+
+import org.junit.Test;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import io.reactivex.Completable;
+import io.reactivex.functions.Action;
+import io.reactivex.schedulers.TestScheduler;
+
+import static org.junit.Assert.assertEquals;
+
+public class CompletableTimerTest {
+    @Test
+    public void timer() {
+        final TestScheduler testScheduler = new TestScheduler();
+
+        final AtomicLong atomicLong = new AtomicLong();
+        Completable.timer(2, TimeUnit.SECONDS, testScheduler).subscribe(new Action() {
+            @Override
+            public void run() throws Exception {
+                atomicLong.incrementAndGet();
+            }
+        });
+
+        assertEquals(0, atomicLong.get());
+
+        testScheduler.advanceTimeBy(1, TimeUnit.SECONDS);
+
+        assertEquals(0, atomicLong.get());
+
+        testScheduler.advanceTimeBy(1, TimeUnit.SECONDS);
+
+        assertEquals(1, atomicLong.get());
+    }
+}

src/test/java/io/reactivex/maybe/MaybeTimerTest.java

@@ -0,0 +1,50 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.maybe;
+
+import org.junit.Test;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import io.reactivex.Maybe;
+import io.reactivex.functions.Consumer;
+import io.reactivex.schedulers.TestScheduler;
+
+import static org.junit.Assert.assertEquals;
+
+public class MaybeTimerTest {
+    @Test
+    public void timer() {
+        final TestScheduler testScheduler = new TestScheduler();
+
+        final AtomicLong atomicLong = new AtomicLong();
+        Maybe.timer(2, TimeUnit.SECONDS, testScheduler).subscribe(new Consumer<Long>() {
+            @Override
+            public void accept(final Long value) throws Exception {
+                atomicLong.incrementAndGet();
+            }
+        });
+
+        assertEquals(0, atomicLong.get());
+
+        testScheduler.advanceTimeBy(1, TimeUnit.SECONDS);
+
+        assertEquals(0, atomicLong.get());
+
+        testScheduler.advanceTimeBy(1, TimeUnit.SECONDS);
+
+        assertEquals(1, atomicLong.get());
+    }
+}

src/test/java/io/reactivex/single/SingleTimerTest.java

@@ -0,0 +1,50 @@
+/**
+ * Copyright 2016 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.single;
+
+import org.junit.Test;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import io.reactivex.Single;
+import io.reactivex.functions.Consumer;
+import io.reactivex.schedulers.TestScheduler;
+
+import static org.junit.Assert.assertEquals;
+
+public class SingleTimerTest {
+    @Test
+    public void timer() {
+        final TestScheduler testScheduler = new TestScheduler();
+
+        final AtomicLong atomicLong = new AtomicLong();
+        Single.timer(2, TimeUnit.SECONDS, testScheduler).subscribe(new Consumer<Long>() {
+            @Override
+            public void accept(final Long value) throws Exception {
+                atomicLong.incrementAndGet();
+            }
+        });
+
+        assertEquals(0, atomicLong.get());
+
+        testScheduler.advanceTimeBy(1, TimeUnit.SECONDS);
+
+        assertEquals(0, atomicLong.get());
+
+        testScheduler.advanceTimeBy(1, TimeUnit.SECONDS);
+
+        assertEquals(1, atomicLong.get());
+    }
+}