ResultSets { public static Observable<ResultSet> queryAllAsObservable(Session session, String query, Object... partitionKeys) { List<ResultSetFuture> futures = sendQueries(session, query, partitionKeys); Scheduler scheduler = Schedulers.io(); List<Observable<ResultSet>> observables = Lists.transform(futures, (ResultSetFuture future) -> Observable.from(future, scheduler)); return Observable.merge(observables); }  static Future<List<ResultSet>> queryAllAsList(Session session, String query, Object... partitionKeys); static List<ListenableFuture<ResultSet>> queryAll(Session session, String query, Object... partitionKeys); static Observable<ResultSet> queryAllAsObservable(Session session, String query, Object... partitionKeys);  }
@Test public void should_query_partitions_as_observable() throws Exception { Observable<ResultSet> results = ResultSets.queryAllAsObservable(session, "SELECT * FROM users WHERE id = ?", uuids); Observable<String> names = results.map((ResultSet rs) -> rs.one().getString("name")); CountDownLatch latch = new CountDownLatch(1); CollectingObserver<String> collector = new CollectingObserver<>(latch); names.subscribe(collector); boolean completed = latch.await(100, MILLISECONDS); Assertions.assertThat(completed).isTrue() .as("Expected observable to complete in less than 100 milliseconds"); Assertions.assertThat(collector.values).containsOnly("user1", "user2", "user3", "user4", "user5"); }