@Test public void testNonContentedComputation() throws Exception { ExtTestSubscriber<String> testSubscriber = new ExtTestSubscriber<>(); invoker.recompute().subscribe(testSubscriber); testScheduler.advanceTimeBy(TASK_DELAY_MS, TimeUnit.MILLISECONDS); assertThat(testSubscriber.takeNext()).isEqualTo("Done round 0"); assertThat(testSubscriber.isUnsubscribed()).isTrue(); assertThat(invocationCounter.get()).isEqualTo(1); }
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();  }