RxGuava { public static <T> Observable<T> fromScalar(final ListenableFuture<T> future) { return fromScalar(future, MoreExecutors.sameThreadExecutor()); }  static Observable<T> fromIterator(final ListenableFuture<Iterator<T>> futureIterator); static Observable<T> fromIterable(final ListenableFuture<? extends Iterable<T>> futureIterable); static Observable<T> fromIterator(final Supplier<ListenableFuture<Iterator<T>>> futureIteratorSupplier); static Observable<T> fromIterator(final ListenableFuture<Iterator<T>> futureIterator,
      final Executor executor); static Observable<T> fromIterable(final ListenableFuture<? extends Iterable<T>> futureIterable,
      final Executor executor); static Observable<T> fromIterator(final Supplier<ListenableFuture<Iterator<T>>> futureIteratorSupplier,
      final Executor executor); static Observable<T> fromIterable(final Supplier<ListenableFuture<Iterable<T>>> futureSupplier,
      final Executor executor); static Observable<T> fromScalar(final ListenableFuture<T> future); static Observable<T> fromScalar(final Supplier<ListenableFuture<T>> futureSupplier); static Observable<T> fromScalar(final ListenableFuture<T> future, final Executor executor); static Observable<T> fromScalar(final Supplier<ListenableFuture<T>> futureSupplier, final Executor executor);  }
@Test public void testFromScalar() throws InterruptedException { String value = "test"; SettableFuture<String> future = SettableFuture.create(); Executors.newSingleThreadScheduledExecutor().schedule(()->future.set(value), 100, TimeUnit.MILLISECONDS); CountDownLatch latch = new CountDownLatch(1); long start = System.currentTimeMillis(); AtomicReference<Throwable> error = new AtomicReference<>(); Observable<String> o = RxGuava.fromScalar(future); o.subscribe(item->latch.countDown(),e->error.set(e)); Assert.assertNull(error.get()); Assert.assertTrue("should not block", System.currentTimeMillis() - start < 90); latch.await(150, TimeUnit.MILLISECONDS); }
@Test public void testFromScalarWithExecutor() throws InterruptedException { String value = "test"; SettableFuture<String> future = SettableFuture.create(); Executors.newSingleThreadScheduledExecutor().schedule(()->future.set(value), 100, TimeUnit.MILLISECONDS); CountDownLatch latch = new CountDownLatch(1); AtomicReference<String> threadName = new AtomicReference<>(); Observable<String> o = RxGuava.fromScalar(future, executor); o.doOnNext(item->threadName.set(Thread.currentThread().getName())) .subscribe(item->latch.countDown()); latch.await(); Assert.assertNotNull(threadName.get()); Assert.assertEquals("test-fixed-0", threadName.get()); }
@Test public void testFromLazyScalar() throws Exception { PublishSubject<Integer> o1 = PublishSubject.create(); AtomicBoolean fetched = new AtomicBoolean(); CountDownLatch latch = new CountDownLatch(3); AtomicReference<Throwable> error = new AtomicReference<>(); Observable<String> o2 = RxGuava.fromScalar(()->{ fetched.set(true); return Futures.immediateFuture("test"); }, executor); Observable.concat(o1, o2).subscribe(item->latch.countDown(),e->error.set(e)); Thread.sleep(100); Assert.assertFalse("lazy future should not have been fetched yet", fetched.get()); o1.onCompleted(); Thread.sleep(50); Assert.assertTrue("lazy future should have been fetched by now", fetched.get()); Assert.assertNull(error.get()); }