### Question:
BatchStatementTransformer implements Transformer<Statement, BatchStatement> { @Override public Observable<BatchStatement> call(Observable<Statement> statements) { return statements .window(batchSize) .flatMap(window -> window.collect(batchStatementFactory, BatchStatement::add)); } BatchStatementTransformer();  BatchStatementTransformer(Func0<BatchStatement> batchStatementFactory, int batchSize); @Override Observable<BatchStatement> call(Observable<Statement> statements); static final int DEFAULT_BATCH_SIZE; static final Func0<BatchStatement> DEFAULT_BATCH_STATEMENT_FACTORY; }### Answer:
@Test public void testCall() throws Exception { int expected = 6; List<BatchStatement> result = Observable.range(0, (expected - 1) * batchSize + 1) .map(i -> mock(Statement.class)) .compose(batchStatementTransformer) .toList() .toBlocking() .single(); assertEquals(expected, result.size()); for (int i = 0; i < result.size(); i++) { assertEquals(i < (result.size() - 1) ? batchSize : 1, result.get(i).size()); } }