### Question:
CTSReaper implements Runnable { public void run() { debug("Reaper starting"); StopWatch query = new StopWatch(); StopWatch waiting = new StopWatch(); List<CountDownLatch> latches = new ArrayList<CountDownLatch>(); ReaperQuery reaperQuery = queryFactory.getQuery(); try { long total = 0; query.start(); for (Collection<String> ids = reaperQuery.nextPage(); ids != null; ids = reaperQuery.nextPage()) { if (Thread.interrupted()) { Thread.currentThread().interrupt(); debug("Interrupted, returning"); return; } total += ids.size(); debug("Queried {0} tokens", Long.toString(total)); latches.add(tokenDeletion.deleteBatch(ids)); } query.stop(); waiting.start(); debug("Expired Token Query Time: {0}ms", Long.toString(query.getTime())); for (CountDownLatch latch : latches) { try { latch.await(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } waiting.stop(); monitoringStore.addReaperRun(query.getStartTime(), query.getTime() + waiting.getTime(), total); debug("Worker threads Time: {0}ms", Long.toString(waiting.getTime())); } catch (CoreTokenException e) { debug.error("CTS Reaper failed", e); } debug("Reaper complete"); } @Inject  CTSReaper(final ReaperQueryFactory queryFactory, final TokenDeletion tokenDeletion,
                     final CTSReaperMonitoringStore monitoringStore,
                     @Named(CoreTokenConstants.CTS_REAPER_DEBUG) final Debug debug); void run();  }### Answer:
@Test public void shouldUseReaperQueryForNextPage() throws CoreTokenException { given(mockQuery.nextPage()).willReturn(null); reaper.run(); verify(mockQuery).nextPage(); }
@Test public void shouldSignalTokensToTokenDeletion() throws CoreTokenException { Collection<String> tokens = Arrays.asList("badger", "weasel", "ferret"); given(mockQuery.nextPage()).willReturn(tokens).willReturn(null); given(mockTokenDeletion.deleteBatch(anyCollection())).willReturn(new CountDownLatch(0)); reaper.run(); verify(mockTokenDeletion).deleteBatch(eq(tokens)); }
@Test public void shouldWaitForAllCountDownLatchesBeforeContinuing() throws CoreTokenException, InterruptedException { CountDownLatch one = mock(CountDownLatch.class); CountDownLatch two = mock(CountDownLatch.class); CountDownLatch three = mock(CountDownLatch.class); Collection<String> tokens = Arrays.asList("badger", "weasel", "ferret"); given(mockQuery.nextPage()).willReturn(tokens).willReturn(tokens).willReturn(tokens).willReturn(null); given(mockTokenDeletion.deleteBatch(anyCollection())).willReturn(one).willReturn(two).willReturn(three); reaper.run(); verify(one).await(); verify(two).await(); verify(three).await(); }
@Test public void shouldRespondToInterruptSignal() throws CoreTokenException { Collection<String> tokens = Arrays.asList("badger", "weasel", "ferret"); given(mockQuery.nextPage()).willReturn(tokens).willReturn(null); Thread.currentThread().interrupt(); reaper.run(); verify(mockTokenDeletion, times(0)).deleteBatch(eq(tokens)); }