@Test public void shouldCloseTokenDeletionWhenComplete() throws CoreTokenException { given(mockBuilder.executeRawResults()).willReturn(Collections.EMPTY_LIST); given(mockBuilder.getPagingCookie()).willReturn(QueryBuilder.getEmptyPagingCookie()); given(mockBuilder.withFilter(any(Filter.class))).willReturn(mockBuilder); given(mockBuilder.returnTheseAttributes(any(CoreTokenField.class))).willReturn(mockBuilder); QueryFilter queryFilter = new QueryFilter(new LDAPDataConversion()); given(mockQueryFactory.createFilter()).willReturn(queryFilter); reaper.run(); verify(mockTokenDeletion).close(); }
public void run() { StopWatch query = new StopWatch(); StopWatch waiting = new StopWatch(); List<CountDownLatch> latches = new ArrayList<CountDownLatch>(); calendar.setTimeInMillis(System.currentTimeMillis()); Filter expired = factory.createFilter().and().beforeDate(calendar).build(); QueryBuilder queryBuilder = factory.createInstance() .withFilter(expired) .returnTheseAttributes(CoreTokenField.TOKEN_ID); QueryPageIterator iterator = new QueryPageIterator(queryBuilder, config.getCleanupPageSize()); query.start(); long total = 0; try { while (iterator.hasNext()) { Collection<Entry> entries = iterator.next(); total += entries.size(); if (Thread.interrupted()) { return; } CountDownLatch latch = new CountDownLatch(entries.size()); DeleteComplete complete = new DeleteComplete(latch); latches.add(latch); try { tokenDeletion.deleteBatch(entries, complete); } catch (ErrorResultException e) { debug.error("Failed to get a connection, will retry later", e); return; } if (debug.messageEnabled()) { debug.message(MessageFormat.format( CoreTokenConstants.DEBUG_HEADER + "Reaper: Queried {0} Tokens", total)); } } query.stop(); waiting.start(); if (debug.messageEnabled()) { debug.message(MessageFormat.format( CoreTokenConstants.DEBUG_HEADER + "Reaper: Expired Token Query Time: {0}ms", query.getTime())); } for (CountDownLatch latch : latches) { try { latch.await(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } waiting.stop(); if (debug.messageEnabled()) { debug.message(MessageFormat.format( CoreTokenConstants.DEBUG_HEADER + "Reaper: Worker threads Time: {0}ms", waiting.getTime())); } } finally { IOUtils.closeIfNotNull(tokenDeletion); } }
CTSReaper implements Runnable { public void run() { StopWatch query = new StopWatch(); StopWatch waiting = new StopWatch(); List<CountDownLatch> latches = new ArrayList<CountDownLatch>(); calendar.setTimeInMillis(System.currentTimeMillis()); Filter expired = factory.createFilter().and().beforeDate(calendar).build(); QueryBuilder queryBuilder = factory.createInstance() .withFilter(expired) .returnTheseAttributes(CoreTokenField.TOKEN_ID); QueryPageIterator iterator = new QueryPageIterator(queryBuilder, config.getCleanupPageSize()); query.start(); long total = 0; try { while (iterator.hasNext()) { Collection<Entry> entries = iterator.next(); total += entries.size(); if (Thread.interrupted()) { return; } CountDownLatch latch = new CountDownLatch(entries.size()); DeleteComplete complete = new DeleteComplete(latch); latches.add(latch); try { tokenDeletion.deleteBatch(entries, complete); } catch (ErrorResultException e) { debug.error("Failed to get a connection, will retry later", e); return; } if (debug.messageEnabled()) { debug.message(MessageFormat.format( CoreTokenConstants.DEBUG_HEADER + "Reaper: Queried {0} Tokens", total)); } } query.stop(); waiting.start(); if (debug.messageEnabled()) { debug.message(MessageFormat.format( CoreTokenConstants.DEBUG_HEADER + "Reaper: Expired Token Query Time: {0}ms", query.getTime())); } for (CountDownLatch latch : latches) { try { latch.await(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } waiting.stop(); if (debug.messageEnabled()) { debug.message(MessageFormat.format( CoreTokenConstants.DEBUG_HEADER + "Reaper: Worker threads Time: {0}ms", waiting.getTime())); } } finally { IOUtils.closeIfNotNull(tokenDeletion); } } }
CTSReaper implements Runnable { public void run() { StopWatch query = new StopWatch(); StopWatch waiting = new StopWatch(); List<CountDownLatch> latches = new ArrayList<CountDownLatch>(); calendar.setTimeInMillis(System.currentTimeMillis()); Filter expired = factory.createFilter().and().beforeDate(calendar).build(); QueryBuilder queryBuilder = factory.createInstance() .withFilter(expired) .returnTheseAttributes(CoreTokenField.TOKEN_ID); QueryPageIterator iterator = new QueryPageIterator(queryBuilder, config.getCleanupPageSize()); query.start(); long total = 0; try { while (iterator.hasNext()) { Collection<Entry> entries = iterator.next(); total += entries.size(); if (Thread.interrupted()) { return; } CountDownLatch latch = new CountDownLatch(entries.size()); DeleteComplete complete = new DeleteComplete(latch); latches.add(latch); try { tokenDeletion.deleteBatch(entries, complete); } catch (ErrorResultException e) { debug.error("Failed to get a connection, will retry later", e); return; } if (debug.messageEnabled()) { debug.message(MessageFormat.format( CoreTokenConstants.DEBUG_HEADER + "Reaper: Queried {0} Tokens", total)); } } query.stop(); waiting.start(); if (debug.messageEnabled()) { debug.message(MessageFormat.format( CoreTokenConstants.DEBUG_HEADER + "Reaper: Expired Token Query Time: {0}ms", query.getTime())); } for (CountDownLatch latch : latches) { try { latch.await(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } waiting.stop(); if (debug.messageEnabled()) { debug.message(MessageFormat.format( CoreTokenConstants.DEBUG_HEADER + "Reaper: Worker threads Time: {0}ms", waiting.getTime())); } } finally { IOUtils.closeIfNotNull(tokenDeletion); } } @Inject  CTSReaper(QueryFactory factory, CoreTokenConfig config,
                     TokenDeletion tokenDeletion, @Named(CoreTokenConstants.CTS_REAPER_DEBUG) Debug debug); }
CTSReaper implements Runnable { public void run() { StopWatch query = new StopWatch(); StopWatch waiting = new StopWatch(); List<CountDownLatch> latches = new ArrayList<CountDownLatch>(); calendar.setTimeInMillis(System.currentTimeMillis()); Filter expired = factory.createFilter().and().beforeDate(calendar).build(); QueryBuilder queryBuilder = factory.createInstance() .withFilter(expired) .returnTheseAttributes(CoreTokenField.TOKEN_ID); QueryPageIterator iterator = new QueryPageIterator(queryBuilder, config.getCleanupPageSize()); query.start(); long total = 0; try { while (iterator.hasNext()) { Collection<Entry> entries = iterator.next(); total += entries.size(); if (Thread.interrupted()) { return; } CountDownLatch latch = new CountDownLatch(entries.size()); DeleteComplete complete = new DeleteComplete(latch); latches.add(latch); try { tokenDeletion.deleteBatch(entries, complete); } catch (ErrorResultException e) { debug.error("Failed to get a connection, will retry later", e); return; } if (debug.messageEnabled()) { debug.message(MessageFormat.format( CoreTokenConstants.DEBUG_HEADER + "Reaper: Queried {0} Tokens", total)); } } query.stop(); waiting.start(); if (debug.messageEnabled()) { debug.message(MessageFormat.format( CoreTokenConstants.DEBUG_HEADER + "Reaper: Expired Token Query Time: {0}ms", query.getTime())); } for (CountDownLatch latch : latches) { try { latch.await(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } waiting.stop(); if (debug.messageEnabled()) { debug.message(MessageFormat.format( CoreTokenConstants.DEBUG_HEADER + "Reaper: Worker threads Time: {0}ms", waiting.getTime())); } } finally { IOUtils.closeIfNotNull(tokenDeletion); } } @Inject  CTSReaper(QueryFactory factory, CoreTokenConfig config,
                     TokenDeletion tokenDeletion, @Named(CoreTokenConstants.CTS_REAPER_DEBUG) Debug debug); void run(); }
CTSReaper implements Runnable { public void run() { StopWatch query = new StopWatch(); StopWatch waiting = new StopWatch(); List<CountDownLatch> latches = new ArrayList<CountDownLatch>(); calendar.setTimeInMillis(System.currentTimeMillis()); Filter expired = factory.createFilter().and().beforeDate(calendar).build(); QueryBuilder queryBuilder = factory.createInstance() .withFilter(expired) .returnTheseAttributes(CoreTokenField.TOKEN_ID); QueryPageIterator iterator = new QueryPageIterator(queryBuilder, config.getCleanupPageSize()); query.start(); long total = 0; try { while (iterator.hasNext()) { Collection<Entry> entries = iterator.next(); total += entries.size(); if (Thread.interrupted()) { return; } CountDownLatch latch = new CountDownLatch(entries.size()); DeleteComplete complete = new DeleteComplete(latch); latches.add(latch); try { tokenDeletion.deleteBatch(entries, complete); } catch (ErrorResultException e) { debug.error("Failed to get a connection, will retry later", e); return; } if (debug.messageEnabled()) { debug.message(MessageFormat.format( CoreTokenConstants.DEBUG_HEADER + "Reaper: Queried {0} Tokens", total)); } } query.stop(); waiting.start(); if (debug.messageEnabled()) { debug.message(MessageFormat.format( CoreTokenConstants.DEBUG_HEADER + "Reaper: Expired Token Query Time: {0}ms", query.getTime())); } for (CountDownLatch latch : latches) { try { latch.await(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } waiting.stop(); if (debug.messageEnabled()) { debug.message(MessageFormat.format( CoreTokenConstants.DEBUG_HEADER + "Reaper: Worker threads Time: {0}ms", waiting.getTime())); } } finally { IOUtils.closeIfNotNull(tokenDeletion); } } @Inject  CTSReaper(QueryFactory factory, CoreTokenConfig config,
                     TokenDeletion tokenDeletion, @Named(CoreTokenConstants.CTS_REAPER_DEBUG) Debug debug); void run();  }