CassStoreHelper { public Observable<ResultSet> execute(Statement statement) { return buildResultSetObservable(() -> session.executeAsync(statement)); } CassStoreHelper(Session session, Scheduler scheduler); Observable<ResultSet> execute(Statement statement);  }
@Test public void paginatedResultSetCanBeFetchedAsync() { int numberOfRecords = 500; int numberOfPages = 10; Session session = cassandraCQLUnit.getSession(); PreparedStatement insertStmt = session.prepare(insertInto("app_scale_jobs").values( Arrays.asList("job_id", "ref_id"), Arrays.asList(bindMarker(), bindMarker()) )); for (int i = 0; i < numberOfRecords; i++) { ResultSet resultSet = session.execute(insertStmt.bind("job-" + i, UUID.randomUUID())); assertThat(resultSet.wasApplied()).isTrue(); } PreparedStatement loadStmt = session.prepare(select("job_id", "ref_id").from("app_scale_jobs")); Observable<ResultSet> results = new CassStoreHelper(session, Schedulers.io()).execute(loadStmt.bind() .setFetchSize(numberOfRecords / numberOfPages)); results.doOnNext(rows -> rows.forEach(row -> assertThat(row.getString(0)).startsWith("job-")) ).toCompletable().await(1, TimeUnit.MINUTES); }
@Test(expected = IllegalStateException.class ) public void paginatedResultInCassandraThreadsThrowsException() { int numberOfRecords = 500; int numberOfPages = 10; Session session = cassandraCQLUnit.getSession(); PreparedStatement insertStmt = session.prepare(insertInto("app_scale_jobs").values( Arrays.asList("job_id", "ref_id"), Arrays.asList(bindMarker(), bindMarker()) )); for (int i = 0; i < numberOfRecords; i++) { ResultSet resultSet = session.execute(insertStmt.bind("job-" + i, UUID.randomUUID())); assertThat(resultSet.wasApplied()).isTrue(); } PreparedStatement loadStmt = session.prepare(select("job_id", "ref_id").from("app_scale_jobs")); Observable<ResultSet> results = new CassStoreHelper(session, Schedulers.immediate()).execute(loadStmt.bind() .setFetchSize(numberOfRecords / numberOfPages)); results.doOnNext(rows -> rows.forEach(row -> assertThat(row.getString(0)).startsWith("job-")) ).toCompletable().await(1, TimeUnit.MINUTES); }