ObservableExt { public static <T> Observable<T> fromWithDelay(List<Observable<T>> chunks, long delay, TimeUnit timeUnit, Scheduler scheduler) { if (chunks.isEmpty()) { return Observable.empty(); } final Iterator<Observable<T>> chunkIterator = chunks.iterator(); Observable<T> result = chunkIterator.next(); while (chunkIterator.hasNext()) { result = result.concatWith(chunkIterator.next().delay(delay, timeUnit, scheduler)); } return result; }  static Observable<T> fromCallable(Supplier<Collection<T>> supplier); static Completable fromCallableSupplier(Callable<Completable> completableSupplier); static Observable<T> toObservable(Future<T> future, Scheduler scheduler); static Observable<T> fromWithDelay(List<Observable<T>> chunks, long delay, TimeUnit timeUnit, Scheduler scheduler); static Observable<T> fromCollection(Supplier<Collection<T>> listProvider); static Observable.Transformer<T, T> head(Supplier<Collection<T>> headSupplier); static Observable.Transformer<T, Pair<T, S>> combine(Supplier<S> stateFactory); static Observable.Transformer<T, R> mapWithState(S zero, BiFunction<T, S, Pair<R, S>> transformer); static Observable.Transformer<T, R> mapWithState(Supplier<S> zeroSupplier, BiFunction<T, S, Pair<R, S>> transformer); static Observable.Transformer<T, R> mapWithState(S zero,
                                                                      BiFunction<T, S, Pair<R, S>> transformer,
                                                                      Observable<Function<S, Pair<R, S>>> cleanupActions); static Observable.Transformer<T, T> subscriptionTimeout(Supplier<Long> timeout, TimeUnit unit, Scheduler scheduler); static Single<Optional<Throwable>> emitError(Observable<?> source); static Observable.Transformer<T, Batch<T, I>> batchWithRateLimit(RateLimitedBatcher<T, I> batcher,
                                                              String metricsRootName,
                                                              Registry registry); static Observable.Transformer<T, Batch<T, I>> batchWithRateLimit(RateLimitedBatcher<T, I> batcher,
                                                              String metricsRootName,
                                                              List<Tag> tags,
                                                              Registry registry); static Single<Optional<Throwable>> emitError(Completable source); static Observable<T> generatorFrom(Supplier<T> source); static Observable<T> generatorFrom(Function<Long, T> source, Scheduler scheduler); static Observable<List<T>> periodicGenerator(Observable<T> sourceObservable,
                                                            long initialDelay,
                                                            long interval,
                                                            TimeUnit timeUnit,
                                                            Scheduler scheduler); static Observable<Optional<Throwable>> schedule(String metricNameRoot,
                                                           Registry registry,
                                                           String completableName,
                                                           Completable completable,
                                                           long initialDelay,
                                                           long interval,
                                                           TimeUnit timeUnit,
                                                           Scheduler scheduler); static Observable<T> protectFromMissingExceptionHandlers(Observable<T> unprotectedStream, Logger logger); static Observable<T> onBackpressureDropAndNotify(Observable<T> unprotectedStream,
                                                                long maxBufferSize,
                                                                Consumer<Long> onDropAction,
                                                                long notificationInterval,
                                                                TimeUnit timeUnit); static void safeUnsubscribe(Subscription... subscriptions); static Subscriber<T> silentSubscriber(); static InstrumentedEventLoop createEventLoop(String metricNameRoot, Registry registry, Scheduler scheduler); static Observable.Transformer<T, T> reemiter(Function<T, T> transformer, long interval, TimeUnit timeUnit, Scheduler scheduler); static List<Observable<T>> propagate(Observable<T> source, int outputs);  }
@Test public void testFromWithDelayEmpty() { final List<Observable<Object>> chunks = Collections.emptyList(); Observable<Object> observable = ObservableExt.fromWithDelay(chunks, DELAY_MS, TimeUnit.MILLISECONDS, testScheduler); observable = observable.defaultIfEmpty(B); observable.subscribe(testSubscriber); assertThat(testSubscriber.takeNext()).isSameAs(B); assertThat(testSubscriber.takeNext()).isNull(); }
@Test public void testFromWithDelaySingle() { final List<Observable<Object>> chunks = Collections.singletonList(Observable.just(A)); Observable<Object> observable = ObservableExt.fromWithDelay(chunks, DELAY_MS, TimeUnit.MILLISECONDS, testScheduler); observable = observable.defaultIfEmpty(B); observable.subscribe(testSubscriber); assertThat(testSubscriber.takeNext()).isSameAs(A); assertThat(testSubscriber.takeNext()).isNull(); }
@Test public void testFromWithDelayMultiple() { final List<Observable<Object>> chunks = Arrays.asList( Observable.just(A).delay(1, TimeUnit.MILLISECONDS, testScheduler), Observable.just(B, C, D), Observable.just(E)); Observable<Object> observable = ObservableExt.fromWithDelay(chunks, DELAY_MS, TimeUnit.MILLISECONDS, testScheduler); observable = observable.defaultIfEmpty(B); observable.subscribe(testSubscriber); assertThat(testSubscriber.takeNext()).isNull(); assertThat(1).isLessThan(DELAY_MS); testScheduler.advanceTimeBy(1, TimeUnit.MILLISECONDS); assertThat(testSubscriber.takeNext()).isSameAs(A); assertThat(testSubscriber.takeNext()).isNull(); testScheduler.advanceTimeBy(DELAY_MS, TimeUnit.MILLISECONDS); assertThat(testSubscriber.takeNext()).isSameAs(B); assertThat(testSubscriber.takeNext()).isSameAs(C); assertThat(testSubscriber.takeNext()).isSameAs(D); assertThat(testSubscriber.takeNext()).isNull(); testScheduler.advanceTimeBy(DELAY_MS, TimeUnit.MILLISECONDS); assertThat(testSubscriber.takeNext()).isSameAs(E); assertThat(testSubscriber.takeNext()).isNull(); }
@Test public void testFromWithDelayHundreds() { final int NUM_CHUNKS_TO_TEST = 300; final List<Observable<Integer>> chunks = new ArrayList<>(NUM_CHUNKS_TO_TEST); for (int i = 0; i < NUM_CHUNKS_TO_TEST; ++i) { chunks.add(Observable.just(i).delay(1000, TimeUnit.MILLISECONDS, testScheduler)); } Observable<Integer> observable = ObservableExt.fromWithDelay(chunks, DELAY_MS, TimeUnit.MILLISECONDS, testScheduler); observable = observable.defaultIfEmpty(-1); observable.subscribe(testSubscriber); assertThat(testSubscriber.takeNext()).isNull(); for (int i = 0; i < NUM_CHUNKS_TO_TEST; ++i) { if (i > 0) testScheduler.advanceTimeBy(DELAY_MS, TimeUnit.MILLISECONDS); testScheduler.advanceTimeBy(1000, TimeUnit.MILLISECONDS); assertThat(testSubscriber.takeNext()).isEqualTo(i); assertThat(testSubscriber.takeNext()).isNull(); } }