StreamingGeneratedTransExecution implements Runnable { @VisibleForTesting protected void injectRows( List<RowMetaAndData> rowIterator, LogChannelInterface log, RowProducer rowProducer ) { for ( RowMetaAndData injectRows : rowIterator ) { while ( !rowProducer.putRowWait( injectRows.getRowMeta(), injectRows.getData(), 1, TimeUnit.SECONDS ) && genTrans.isRunning() ) { log.logRowlevel( DataServiceConstants.ROW_BUFFER_IS_FULL_TRYING_AGAIN ); } } } 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 testInjectRows(){ List<RowMetaAndData> rowIterator = new ArrayList<>(); LogChannelInterface log = mock( LogChannelInterface.class ); RowProducer rowProducer = mock( RowProducer.class ); RowMeta rowMeta = new RowMeta( ); Object data = "data"; Object[] dataArray = new Object[] { data }; RowMetaAndData rowMetaAndData = new RowMetaAndData( rowMeta, data ); rowIterator.add( rowMetaAndData ); doReturn( true ).when( rowProducer ).putRowWait( rowMeta, dataArray, 1, TimeUnit.SECONDS ); doCallRealMethod().when( genTransExecutor ).injectRows( rowIterator, log, rowProducer ); genTransExecutor.injectRows( rowIterator, log, rowProducer ); verify( rowProducer, times( 1 ) ).putRowWait( rowMeta, dataArray, 1, TimeUnit.SECONDS ); }
@Test public void testInjectNoRows(){ List<RowMetaAndData> rowIterator = new ArrayList<>(); LogChannelInterface log = mock( LogChannelInterface.class ); RowProducer rowProducer = mock( RowProducer.class ); RowMeta rowMeta = new RowMeta( ); Object[] dataArray = new Object[0]; doReturn( true ).when( rowProducer ).putRowWait( rowMeta, dataArray, 1, TimeUnit.SECONDS ); doCallRealMethod().when( genTransExecutor ).injectRows( rowIterator, log, rowProducer ); genTransExecutor.injectRows( rowIterator, log, rowProducer ); verify( rowProducer, times( 0 ) ).putRowWait( rowMeta, dataArray, 1, TimeUnit.SECONDS ); }