### Question:
FixedTimeStreamWindow implements StreamWindow<I, Result> { @Override public Iterable<Result> buffer( Observable<I> observable ) { Observable<List<I>> buffer = millis > 0 ? batchSize > 0 ? observable.buffer( millis, MILLISECONDS, batchSize ) : observable.buffer( millis, MILLISECONDS ) : observable.buffer( batchSize ); return buffer .observeOn( Schedulers.io() ) .filter( list -> !list.isEmpty() ) .map( this::sendBufferToSubtrans ) .takeWhile( result -> result.getNrErrors() == 0 ) .blockingIterable(); } FixedTimeStreamWindow( SubtransExecutor subtransExecutor, RowMetaInterface rowMeta, long millis,
                                int batchSize ); @Override Iterable<Result> buffer( Observable<I> observable );  }### Answer:
@Test public void emptyResultShouldNotThrowException() throws KettleException { when( subtransExecutor.execute( any() ) ).thenReturn( Optional.empty() ); RowMetaInterface rowMeta = new RowMeta(); rowMeta.addValueMeta( new ValueMetaString( "field" ) ); FixedTimeStreamWindow<List> window = new FixedTimeStreamWindow<>( subtransExecutor, rowMeta, 0, 2 ); window.buffer( Observable.fromIterable( singletonList( asList( "v1", "v2" ) ) ) ).forEach( result -> { } ); }
@Test public void resultsComeBackToParent() throws KettleException { RowMetaInterface rowMeta = new RowMeta(); rowMeta.addValueMeta( new ValueMetaString( "field" ) ); Result mockResult = new Result(); mockResult.setRows( Arrays.asList( new RowMetaAndData( rowMeta, "queen" ), new RowMetaAndData( rowMeta, "king" ) ) ); when( subtransExecutor.execute( any() ) ).thenReturn( Optional.of( mockResult ) ); FixedTimeStreamWindow<List> window = new FixedTimeStreamWindow<>( subtransExecutor, rowMeta, 0, 2 ); window.buffer( Observable.fromIterable( singletonList( asList( "v1", "v2" ) ) ) ) .forEach( result -> assertEquals( mockResult, result ) ); }