@Test public void testContentedComputation() throws Exception { ExtTestSubscriber<String> testSubscriber1 = new ExtTestSubscriber<>(); ExtTestSubscriber<String> testSubscriber2 = new ExtTestSubscriber<>(); ExtTestSubscriber<String> testSubscriber3 = new ExtTestSubscriber<>(); invoker.recompute().subscribe(testSubscriber1); testScheduler.advanceTimeBy(1, TimeUnit.MILLISECONDS); invoker.recompute().subscribe(testSubscriber2); invoker.recompute().subscribe(testSubscriber3); testScheduler.advanceTimeBy(TASK_DELAY_MS - 1, TimeUnit.MILLISECONDS); assertThat(testSubscriber1.takeNext()).isEqualTo("Done round 0"); assertThat(testSubscriber1.isUnsubscribed()).isTrue(); assertThat(testSubscriber2.takeNext()).isNull(); assertThat(testSubscriber3.takeNext()).isNull(); assertThat(invocationCounter.get()).isEqualTo(1); testScheduler.advanceTimeBy(TASK_DELAY_MS, TimeUnit.MILLISECONDS); assertThat(testSubscriber2.takeNext()).isEqualTo("Done round 1"); assertThat(testSubscriber3.takeNext()).isEqualTo("Done round 1"); assertThat(invocationCounter.get()).isEqualTo(2); }
public Observable<O> recompute() { return Observable.create(subscriber -> { waitingObservers.add(subscriber); worker.schedule(this::drain); }); }
ComputationTaskInvoker { public Observable<O> recompute() { return Observable.create(subscriber -> { waitingObservers.add(subscriber); worker.schedule(this::drain); }); } }
ComputationTaskInvoker { public Observable<O> recompute() { return Observable.create(subscriber -> { waitingObservers.add(subscriber); worker.schedule(this::drain); }); } ComputationTaskInvoker(Observable<O> computation, Scheduler scheduler); }
ComputationTaskInvoker { public Observable<O> recompute() { return Observable.create(subscriber -> { waitingObservers.add(subscriber); worker.schedule(this::drain); }); } ComputationTaskInvoker(Observable<O> computation, Scheduler scheduler); Observable<O> recompute(); }
ComputationTaskInvoker { public Observable<O> recompute() { return Observable.create(subscriber -> { waitingObservers.add(subscriber); worker.schedule(this::drain); }); } ComputationTaskInvoker(Observable<O> computation, Scheduler scheduler); Observable<O> recompute();  }