StreamingGeneratedTransExecution implements Runnable { @Override public void run() { getGeneratedDataObservable() .toFlowable( BackpressureStrategy.LATEST ) .onBackpressureBuffer( 1, () -> { }, BackpressureOverflowStrategy.DROP_OLDEST ) .doOnError( t -> logger.error( "Error receiving data from the service transformation observable", t ) ) .doOnNext( rowMetaAndDataList -> { if ( consumersCount.get() > 0 ) { serviceExecutor.touchServiceListener( this.streamingGeneratedTransCacheKey ); } } ) .subscribe( rowMetaAndDataList -> this.runGenTrans( rowMetaAndDataList ) ); if ( this.stream == null ) { this.stream = serviceExecutor.getBuffer( query, getWindowConsumer(), windowMode, windowSize, windowEvery, windowLimit ); } try { if ( stream == null ) { this.runGenTrans( Collections.emptyList() ); } } catch ( KettleStepException e ) { throw Throwables.propagate( e ); } } StreamingGeneratedTransExecution( final StreamingServiceTransExecutor serviceExecutor, final Trans genTrans,
                                           final Observer<List<RowMetaAndData>> rowConsumer, boolean pollingMode,
                                           final String injectorStepName, final String resultStepName, final String query,
                                           final IDataServiceClientService.StreamingMode windowMode,
                                           long windowSize, long windowEvery, long windowLimit,
                                           String streamingGeneratedTransCacheKey ); @Override void run(); void addNewRowConsumer( final Observer<List<RowMetaAndData>> consumer, boolean pollingMode ); void clearRowConsumers( );  }
@Test public void testRunCachedWindow() throws Exception { when( streamExecutionListener.getCachePreWindow() ).thenReturn( rowIterator ); when( genTrans.isRunning() ).thenReturn( true ); genTransExecutor = spy( new StreamingGeneratedTransExecution( serviceExecutor, genTrans, consumer, pollingMode, MOCK_INJECTOR_STEP_NAME, MOCK_RESULT_STEP_NAME, MOCK_QUERY, MOCK_WINDOW_MODE_ROW_BASED, 10000, 1, 10000, "" ) ); genTransExecutor.run(); verifyExecution( 0 ); }
@Test public void testRunEmptyStream() throws Exception { doReturn( null ).when( serviceExecutor ).getBuffer( MOCK_QUERY, windowConsumer, MOCK_WINDOW_MODE_ROW_BASED, MOCK_WINDOW_SIZE, MOCK_WINDOW_EVERY, MOCK_WINDOW_MAX_SIZE ); when( genTrans.isRunning() ).thenReturn( true ); genTransExecutor.run(); verifyExecution( 1 ); }