### Question:
ResultSetToRowsTransformer implements Transformer<ResultSet, Row> { @Override public Observable<Row> call(Observable<ResultSet> resultSetObservable) { return resultSetObservable.flatMap(resultSet -> { return Observable.create(subscriber -> { subscriber.setProducer(new RowProducer(resultSet, subscriber, scheduler)); }); }); } ResultSetToRowsTransformer();  ResultSetToRowsTransformer(Scheduler scheduler); @Override Observable<Row> call(Observable<ResultSet> resultSetObservable);  }### Answer:
@Test public void testSinglePage() throws Exception { int rowCount = 763; MockResultSet resultSet = MockResultSet.createSinglePage(rowCount); Observable<Row> rows = transformer.call(Observable.just(resultSet)); TestSubscriber<Row> subscriber = new TestSubscriber<>(); rows.subscribe(subscriber); subscriber.awaitTerminalEvent(); subscriber.assertNoErrors(); subscriber.assertValueCount(rowCount); subscriber.assertReceivedOnNext(LongStream.range(0, rowCount).mapToObj(MockRow::new).collect(toList())); }
@Test public void testMultiPage() throws Exception { int rowCount = 76377; int pageSize = 5000; MockResultSet resultSet = MockResultSet.createMultiPage(rowCount, pageSize); Observable<Row> rows = transformer.call(Observable.just(resultSet)); TestSubscriber<Row> subscriber = new TestSubscriber<>(); rows.subscribe(subscriber); subscriber.awaitTerminalEvent(); subscriber.assertNoErrors(); subscriber.assertValueCount(rowCount); subscriber.assertReceivedOnNext(LongStream.range(0, rowCount).mapToObj(MockRow::new).collect(toList())); }
@Test public void testBackPressure() throws Exception { int rowCount = 76377; int pageSize = 5000; MockResultSet resultSet = MockResultSet.createMultiPage(rowCount, pageSize); Observable<Row> rows = transformer.call(Observable.just(resultSet)); AtomicReference<TestSubscriber<Row>> subscriberRef = new AtomicReference<>(); int initialRequest = 750; int intermediateRequest = 100; int totalRequest = initialRequest + intermediateRequest; TestSubscriber<Row> subscriber = new TestSubscriber<>(new Observer<Row>() { final AtomicLong received = new AtomicLong(); @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(Row o) { long r = received.incrementAndGet(); if (r % totalRequest == 50) { subscriberRef.get().requestMore(intermediateRequest); return; } if (r / totalRequest == 1) { EXECUTOR.schedule(() -> { subscriberRef.get().requestMore(initialRequest); }, 5, TimeUnit.MILLISECONDS); } } }, initialRequest); subscriberRef.set(subscriber); rows.subscribe(subscriber); subscriber.awaitTerminalEvent(); subscriber.assertNoErrors(); subscriber.assertValueCount(rowCount); subscriber.assertReceivedOnNext(LongStream.range(0, rowCount).mapToObj(MockRow::new).collect(toList())); }
@Test public void testEmptyResultSet() throws Exception { MockResultSet resultSet = MockResultSet.createEmpty(); Observable<Row> rows = transformer.call(Observable.just(resultSet)); TestSubscriber<Row> subscriber = new TestSubscriber<>(); rows.subscribe(subscriber); subscriber.awaitTerminalEvent(); subscriber.assertNoErrors(); subscriber.assertNoValues(); }