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();  }
@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); }
@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); }
@Test public void testFailingComputations() throws Exception { AtomicBoolean flag = new AtomicBoolean(); ComputationTaskInvoker<String> slowInvoker = new ComputationTaskInvoker<>( Observable.create(subscriber -> { if (flag.getAndSet(true)) { subscriber.onNext("OK!"); subscriber.onCompleted(); } else { Observable.<String>timer(1, TimeUnit.SECONDS, testScheduler) .flatMap(tick -> Observable.<String>error(new RuntimeException("simulated computation error"))) .subscribe(subscriber); } }), testScheduler ); ExtTestSubscriber<String> failingSubscriber = new ExtTestSubscriber<>(); slowInvoker.recompute().subscribe(failingSubscriber); testScheduler.triggerActions(); ExtTestSubscriber<String> okSubscriber = new ExtTestSubscriber<>(); slowInvoker.recompute().subscribe(okSubscriber); testScheduler.advanceTimeBy(1, TimeUnit.SECONDS); failingSubscriber.assertOnError(); assertThat(okSubscriber.takeNext()).isEqualTo("OK!"); okSubscriber.assertOnCompleted(); }