StreamingGeneratedTransExecution implements Runnable { @VisibleForTesting protected RowAdapter getRowWrittenListener( List<RowMetaAndData> tempList ) { return new RowAdapter() { @Override public void rowWrittenEvent( RowMetaInterface rowMetaInterface, Object[] objects ) throws KettleStepException { tempList.add( new RowMetaAndData( rowMetaInterface, objects ) ); } }; } 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 testGetRowListener() throws Exception { List<RowMetaAndData> tempList = new ArrayList<>( ); RowAdapter rowAdapter = genTransExecutor.getRowWrittenListener( tempList ); RowMeta rowMeta = new RowMeta( ); Object data = "data"; Object[] dataArray = new Object[] { data }; rowAdapter.rowWrittenEvent( rowMeta, dataArray ); assertEquals( dataArray.length, tempList.size() ); assertNotNull( tempList.get(0).getData() ); assertEquals( dataArray.length, tempList.get(0).getData().length ); assertSame( data, tempList.get(0).getData()[0] ); }