class 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();
  }

  public  FixedTimeStreamWindow( SubtransExecutor subtransExecutor, RowMetaInterface rowMeta, long millis,
                                int batchSize );

}