ServiceObserver extends AbstractFuture<CachedService> implements Runnable { @Override public void run() { StepInterface serviceStep = executor.getServiceTrans().findRunThread( executor.getService().getStepname() ); serviceStep.addRowListener( new RowAdapter() { @Override public synchronized void rowWrittenEvent( RowMetaInterface rowMeta, Object[] row ) { Object[] clonedRow; try { clonedRow = rowMeta.cloneRow( row ); } catch ( KettleValueException e ) { setException( e ); return; } rowMetaAndData.add( new RowMetaAndData( rowMeta, clonedRow ) ); latch.countDown(); } } ); serviceStep.addStepListener( new StepAdapter() { @Override public void stepFinished( Trans trans, StepMeta stepMeta, StepInterface step ) { isRunning = false; if ( executor.getGenTrans().getErrors() > 0 ) { setException( new KettleException( "Dynamic transformation finished with errors, could not cache results" ) ); } else if ( step.isStopped() ) { set( CachedService.partial( rowMetaAndData, executor ) ); } else { set( CachedService.complete( rowMetaAndData ) ); } } } ); } ServiceObserver( DataServiceExecutor executor ); Iterator<RowMetaAndData> rows(); ListenableFuture<CachedService> install(); @Override void run();  }
@Test public void verifyCachedRowIsCloned() throws Exception { when( stepInterface.isStopped() ).thenReturn( false ); observer.run(); verify( stepInterface ).addRowListener( rowAdapterCaptor.capture() ); verify( stepInterface ).addStepListener( stepAdapterCaptor.capture() ); RowAdapter rowAdapter = rowAdapterCaptor.getValue(); StepAdapter stepAdapter = stepAdapterCaptor.getValue(); Object[] clonedRow = new Object[0]; when( rowMeta.cloneRow( row ) ).thenReturn( clonedRow ); rowAdapter.rowWrittenEvent( rowMeta, row ); verify( rowMeta ).cloneRow( row ); stepAdapter.stepFinished( null, null, stepInterface ); CachedService cachedService = observer.get(); assertThat( cachedService.getRowMetaAndData().get( 0 ).getData(), is( clonedRow ) ); }
@Test public void cloneErrorIsPropogated() throws KettleValueException, ExecutionException, InterruptedException, KettleStepException { when( rowMeta.cloneRow( row ) ).thenThrow( exception ); observer.run(); verify( stepInterface ).addRowListener( rowAdapterCaptor.capture() ); RowAdapter rowAdapter = rowAdapterCaptor.getValue(); rowAdapter.rowWrittenEvent( rowMeta, row ); try { observer.get(); fail( "Expected exception" ); } catch ( Exception e ) { assertThat( e.getCause(), is( exception ) ); } }