CachedService implements Serializable { public static CachedService complete( List<RowMetaAndData> rowMetaAndData ) { return new CachedService( rowMetaAndData, Optional.<Integer>absent(), Optional.<Integer>absent() ); } private  CachedService( List<RowMetaAndData> rowMetaAndData, Optional<Integer> ranking,  Optional<Integer> ranking2 ); static CachedService complete( List<RowMetaAndData> rowMetaAndData ); static CachedService partial( List<RowMetaAndData> rowMetaAndData, DataServiceExecutor executor ); List<RowMetaAndData> getRowMetaAndData(); Optional<Integer> getRanking(); @Override String toString(); boolean isComplete(); boolean answersQuery( DataServiceExecutor executor );  }
@Test public void testComplete() throws Exception { CachedService unbounded, rowLimit, ordered; unbounded = CachedService.complete( testData ); when( sqlTransGenerator.getRowLimit() ).thenReturn( 10, 0 ); rowLimit = CachedService.complete( testData ); ordered = CachedService.complete( testData ); for ( CachedService loader : ImmutableList.of( unbounded, rowLimit, ordered ) ) { assertThat( loader.getRowMetaAndData(), equalTo( testData ) ); assertThat( loader.isComplete(), is( true ) ); } }
@Test public void testReplayFullCache() throws Exception { DataServiceExecutor executor = dataServiceExecutor( BASE_QUERY ); CachedService cachedService = CachedService.complete( testData ); RowProducer rowProducer = genTrans.addRowProducer( INJECTOR_STEP, 0 ); Executor mockExecutor = mock( Executor.class ); final CachedServiceLoader cachedServiceLoader = new CachedServiceLoader( cachedService, mockExecutor ); ListenableFuture<Integer> replay = cachedServiceLoader.replay( executor ); ArgumentCaptor<Runnable> replayRunnable = ArgumentCaptor.forClass( Runnable.class ); verify( mockExecutor ).execute( replayRunnable.capture() ); stepMetaDataCombi.step = inputStep; stepMetaDataCombi.meta = inputStepMetaInterface; stepMetaDataCombi.data = inputStepDataInterface; List<StepMetaDataCombi> stepMetaDataCombis = new ArrayList<>(); stepMetaDataCombis.add( stepMetaDataCombi ); when( serviceTrans.getSteps() ).thenReturn( stepMetaDataCombis ); executor.executeListeners( DataServiceExecutor.ExecutionPoint.READY ); executor.executeListeners( DataServiceExecutor.ExecutionPoint.START ); verify( serviceTrans ).stopAll(); verify( inputStep ).setOutputDone(); verify( inputStep ).dispose( inputStepMetaInterface, inputStepDataInterface ); verify( inputStep ).markStop(); verify( serviceTrans, never() ).startThreads(); verify( genTrans ).startThreads(); when( rowProducer.putRowWait( any( RowMetaInterface.class ), any( Object[].class ), anyInt(), any( TimeUnit.class ) ) ).then( new Answer<Boolean>() { int calls = 0; @Override public Boolean answer( InvocationOnMock invocation ) throws Throwable { return ++calls != 10; } } ); when( genTrans.isRunning() ).thenReturn( true ); replayRunnable.getValue().run(); assertThat( replay.get( 1, TimeUnit.SECONDS ), equalTo( testData.size() ) ); InOrder rowsProduced = inOrder( rowProducer ); for ( int i = 0; i < testData.size(); i++ ) { RowMetaAndData metaAndData = testData.get( i ); Object[] data = metaAndData.getData(); rowsProduced.verify( rowProducer, times( i == 9 ? 2 : 1 ) ) .putRowWait( eq( metaAndData.getRowMeta() ), and( eq( data ), AdditionalMatchers.not( same( data ) ) ), anyInt(), any( TimeUnit.class ) ); } rowsProduced.verify( rowProducer ).finished(); rowsProduced.verifyNoMoreInteractions(); }
@Test public void testReplayPartialCache() throws Exception { DataServiceExecutor executor = dataServiceExecutor( BASE_QUERY + " LIMIT 20" ); CachedService cachedService = CachedService.complete( testData ); RowProducer rowProducer = genTrans.addRowProducer( INJECTOR_STEP, 0 ); Executor mockExecutor = mock( Executor.class ); final CachedServiceLoader cachedServiceLoader = new CachedServiceLoader( cachedService, mockExecutor ); ListenableFuture<Integer> replay = cachedServiceLoader.replay( executor ); ArgumentCaptor<Runnable> replayRunnable = ArgumentCaptor.forClass( Runnable.class ); verify( mockExecutor ).execute( replayRunnable.capture() ); stepMetaDataCombi.step = inputStep; stepMetaDataCombi.meta = inputStepMetaInterface; stepMetaDataCombi.data = inputStepDataInterface; List<StepMetaDataCombi> stepMetaDataCombis = new ArrayList<>(); stepMetaDataCombis.add( stepMetaDataCombi ); when( serviceTrans.getSteps() ).thenReturn( stepMetaDataCombis ); executor.executeListeners( DataServiceExecutor.ExecutionPoint.READY ); executor.executeListeners( DataServiceExecutor.ExecutionPoint.START ); verify( serviceTrans ).stopAll(); verify( inputStep ).setOutputDone(); verify( inputStep ).dispose( inputStepMetaInterface, inputStepDataInterface ); verify( inputStep ).markStop(); verify( serviceTrans, never() ).startThreads(); verify( genTrans ).startThreads(); final AtomicInteger rowsProduced = new AtomicInteger( 0 ); when( rowProducer.putRowWait( any( RowMetaInterface.class ), any( Object[].class ), anyInt(), any( TimeUnit.class ) ) ).then( new Answer<Boolean>() { @Override public Boolean answer( InvocationOnMock invocation ) throws Throwable { rowsProduced.getAndIncrement(); return true; } } ); when( genTrans.isRunning() ).then( new Answer<Boolean>() { @Override public Boolean answer( InvocationOnMock invocation ) throws Throwable { return rowsProduced.get() < 20; } } ); replayRunnable.getValue().run(); verify( rowProducer ).finished(); assertThat( replay.get( 1, TimeUnit.SECONDS ), equalTo( 20 ) ); assertThat( rowsProduced.get(), equalTo( 20 ) ); for ( RowMetaAndData metaAndData : Iterables.limit( testData, 20 ) ) { Object[] data = metaAndData.getData(); verify( rowProducer ) .putRowWait( eq( metaAndData.getRowMeta() ), and( eq( data ), AdditionalMatchers.not( same( data ) ) ), anyInt(), any( TimeUnit.class ) ); } }