@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 ) ); }
@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 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 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 ); }
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 ); }
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 );  }