Skip to content

Commit f78bd95

Browse files
soshialakarnokd
authored andcommitted
Add delaySubscription() methods to Completable #5081 (#6242)
* Add delaySubscription() methods to Completable #5081 * fix parameter test and documentation for delaySubscription() * add tests to delayCompletable() * remove mocked observer from delaySubscription() tests for Completable
1 parent a1758c4 commit f78bd95

File tree

3 files changed

+119
-2
lines changed

3 files changed

+119
-2
lines changed

src/main/java/io/reactivex/Completable.java

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1390,6 +1390,51 @@ public final Completable delay(final long delay, final TimeUnit unit, final Sche
13901390
return RxJavaPlugins.onAssembly(new CompletableDelay(this, delay, unit, scheduler, delayError));
13911391
}
13921392

1393+
/**
1394+
* Returns a Completable that delays the subscription to the source CompletableSource by a given amount of time.
1395+
* <p>
1396+
* <dl>
1397+
* <dt><b>Scheduler:</b></dt>
1398+
* <dd>This version of {@code delaySubscription} operates by default on the {@code computation} {@link Scheduler}.</dd>
1399+
* </dl>
1400+
*
1401+
* @param delay the time to delay the subscription
1402+
* @param unit the time unit of {@code delay}
1403+
* @return a Completable that delays the subscription to the source CompletableSource by the given amount
1404+
* @since 2.2.3 - experimental
1405+
* @see <a href="http://reactivex.io/documentation/operators/delay.html">ReactiveX operators documentation: Delay</a>
1406+
*/
1407+
@CheckReturnValue
1408+
@Experimental
1409+
@SchedulerSupport(SchedulerSupport.COMPUTATION)
1410+
public final Completable delaySubscription(long delay, TimeUnit unit) {
1411+
return delaySubscription(delay, unit, Schedulers.computation());
1412+
}
1413+
1414+
/**
1415+
* Returns a Completable that delays the subscription to the source CompletableSource by a given amount of time,
1416+
* both waiting and subscribing on a given Scheduler.
1417+
* <p>
1418+
* <dl>
1419+
* <dt><b>Scheduler:</b></dt>
1420+
* <dd>You specify which {@link Scheduler} this operator will use.</dd>
1421+
* </dl>
1422+
*
1423+
* @param delay the time to delay the subscription
1424+
* @param unit the time unit of {@code delay}
1425+
* @param scheduler the Scheduler on which the waiting and subscription will happen
1426+
* @return a Completable that delays the subscription to the source CompletableSource by a given
1427+
* amount, waiting and subscribing on the given Scheduler
1428+
* @since 2.2.3 - experimental
1429+
* @see <a href="http://reactivex.io/documentation/operators/delay.html">ReactiveX operators documentation: Delay</a>
1430+
*/
1431+
@CheckReturnValue
1432+
@Experimental
1433+
@SchedulerSupport(SchedulerSupport.CUSTOM)
1434+
public final Completable delaySubscription(long delay, TimeUnit unit, Scheduler scheduler) {
1435+
return Completable.timer(delay, unit, scheduler).andThen(this);
1436+
}
1437+
13931438
/**
13941439
* Returns a Completable which calls the given onComplete callback if this Completable completes.
13951440
* <p>

src/test/java/io/reactivex/internal/operators/completable/CompletableDelayTest.java

Lines changed: 70 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,11 +20,14 @@
2020

2121
import org.junit.Test;
2222

23-
import io.reactivex.*;
23+
import io.reactivex.CompletableSource;
24+
import io.reactivex.TestHelper;
25+
import io.reactivex.Completable;
2426
import io.reactivex.exceptions.TestException;
2527
import io.reactivex.functions.*;
2628
import io.reactivex.observers.TestObserver;
27-
import io.reactivex.schedulers.*;
29+
import io.reactivex.schedulers.Schedulers;
30+
import io.reactivex.schedulers.TestScheduler;
2831

2932
public class CompletableDelayTest {
3033

@@ -120,4 +123,69 @@ public void errorDelayed() {
120123

121124
to.assertFailure(TestException.class);
122125
}
126+
127+
@Test
128+
public void errorDelayedSubscription() {
129+
TestScheduler scheduler = new TestScheduler();
130+
131+
TestObserver<Void> to = Completable.error(new TestException())
132+
.delaySubscription(100, TimeUnit.MILLISECONDS, scheduler)
133+
.test();
134+
135+
to.assertEmpty();
136+
137+
scheduler.advanceTimeBy(90, TimeUnit.MILLISECONDS);
138+
139+
to.assertEmpty();
140+
141+
scheduler.advanceTimeBy(15, TimeUnit.MILLISECONDS);
142+
143+
to.assertFailure(TestException.class);
144+
}
145+
146+
@Test
147+
public void errorDelayedSubscriptionDisposeBeforeTime() {
148+
TestScheduler scheduler = new TestScheduler();
149+
150+
Completable result = Completable.complete()
151+
.delaySubscription(100, TimeUnit.MILLISECONDS, scheduler);
152+
TestObserver<Void> to = result.test();
153+
154+
to.assertEmpty();
155+
156+
scheduler.advanceTimeBy(90, TimeUnit.MILLISECONDS);
157+
to.dispose();
158+
159+
scheduler.advanceTimeBy(15, TimeUnit.MILLISECONDS);
160+
161+
to.assertEmpty();
162+
}
163+
164+
@Test
165+
public void testDelaySubscriptionDisposeBeforeTime() {
166+
TestScheduler scheduler = new TestScheduler();
167+
168+
Completable result = Completable.complete()
169+
.delaySubscription(100, TimeUnit.MILLISECONDS, scheduler);
170+
TestObserver<Void> to = result.test();
171+
172+
to.assertEmpty();
173+
scheduler.advanceTimeBy(90, TimeUnit.MILLISECONDS);
174+
to.dispose();
175+
scheduler.advanceTimeBy(15, TimeUnit.MILLISECONDS);
176+
to.assertEmpty();
177+
}
178+
179+
@Test
180+
public void testDelaySubscription() {
181+
TestScheduler scheduler = new TestScheduler();
182+
Completable result = Completable.complete()
183+
.delaySubscription(100, TimeUnit.MILLISECONDS, scheduler);
184+
TestObserver<Void> to = result.test();
185+
186+
scheduler.advanceTimeBy(90, TimeUnit.MILLISECONDS);
187+
to.assertEmpty();
188+
scheduler.advanceTimeBy(15, TimeUnit.MILLISECONDS);
189+
to.assertResult();
190+
}
123191
}

src/test/java/io/reactivex/validators/ParamValidationCheckerTest.java

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -261,6 +261,10 @@ public void checkParallelFlowable() {
261261
addOverride(new ParamOverride(Completable.class, 0, ParamMode.ANY, "delay", Long.TYPE, TimeUnit.class, Scheduler.class));
262262
addOverride(new ParamOverride(Completable.class, 0, ParamMode.ANY, "delay", Long.TYPE, TimeUnit.class, Scheduler.class, Boolean.TYPE));
263263

264+
// negative time is considered as zero time
265+
addOverride(new ParamOverride(Completable.class, 0, ParamMode.ANY, "delaySubscription", Long.TYPE, TimeUnit.class));
266+
addOverride(new ParamOverride(Completable.class, 0, ParamMode.ANY, "delaySubscription", Long.TYPE, TimeUnit.class, Scheduler.class));
267+
264268
// zero repeat is allowed
265269
addOverride(new ParamOverride(Completable.class, 0, ParamMode.NON_NEGATIVE, "repeat", Long.TYPE));
266270

0 commit comments

Comments
 (0)