@Test public void testCleanup() throws Exception { testWithMaxFinishedSize(10, 2); queue.gracefulShutdown(false); queue = new TaskQueueImpl( "testQueue", 3, Suppliers.ofInstance(3), Suppliers.ofInstance(10) ); testWithMaxFinishedSize(10, 10); queue.gracefulShutdown(false); queue = new TaskQueueImpl( "testQueue", 3, Suppliers.ofInstance(3), Suppliers.ofInstance(10) ); queue.cleanup(); assertTrue(0 == factory.resetNumEventsFired()); }
@Override public synchronized void cleanup() { logInfoWithQueueName("purging completed tasks"); final List<Future<Void>> completedTasks = Lists.newArrayListWithExpectedSize(this.completedTasksQueue.size()); this.completedTasksQueue.drainTo(completedTasks); for (Future<Void> future: completedTasks) { AppFactoryTask task = FutureStoringThreadPoolExecutor.getOriginal(future); if (null != task) { removeTask(task); } else { logErrorWithQueueName("Could not find taskId for future {}", future); } } }
TaskQueueImpl implements
      TaskQueue,
      ApplicationListener,
      ApplicationEventPublisherAware,
      DisposableBean { @Override public synchronized void cleanup() { logInfoWithQueueName("purging completed tasks"); final List<Future<Void>> completedTasks = Lists.newArrayListWithExpectedSize(this.completedTasksQueue.size()); this.completedTasksQueue.drainTo(completedTasks); for (Future<Void> future: completedTasks) { AppFactoryTask task = FutureStoringThreadPoolExecutor.getOriginal(future); if (null != task) { removeTask(task); } else { logErrorWithQueueName("Could not find taskId for future {}", future); } } } }
TaskQueueImpl implements
      TaskQueue,
      ApplicationListener,
      ApplicationEventPublisherAware,
      DisposableBean { @Override public synchronized void cleanup() { logInfoWithQueueName("purging completed tasks"); final List<Future<Void>> completedTasks = Lists.newArrayListWithExpectedSize(this.completedTasksQueue.size()); this.completedTasksQueue.drainTo(completedTasks); for (Future<Void> future: completedTasks) { AppFactoryTask task = FutureStoringThreadPoolExecutor.getOriginal(future); if (null != task) { removeTask(task); } else { logErrorWithQueueName("Could not find taskId for future {}", future); } } } TaskQueueImpl(String name,
                 int expectedTasksPerId,
                 Supplier<Integer> maxPoolSizeSupplier,
                 Supplier<Integer> maxFinishedSizeSupplier); }
TaskQueueImpl implements
      TaskQueue,
      ApplicationListener,
      ApplicationEventPublisherAware,
      DisposableBean { @Override public synchronized void cleanup() { logInfoWithQueueName("purging completed tasks"); final List<Future<Void>> completedTasks = Lists.newArrayListWithExpectedSize(this.completedTasksQueue.size()); this.completedTasksQueue.drainTo(completedTasks); for (Future<Void> future: completedTasks) { AppFactoryTask task = FutureStoringThreadPoolExecutor.getOriginal(future); if (null != task) { removeTask(task); } else { logErrorWithQueueName("Could not find taskId for future {}", future); } } } TaskQueueImpl(String name,
                 int expectedTasksPerId,
                 Supplier<Integer> maxPoolSizeSupplier,
                 Supplier<Integer> maxFinishedSizeSupplier); @Override synchronized void addTask(@Nonnull AppFactoryTask task); @Nonnull @Override synchronized Iterable<TaskState> getAllTasks(); @Nonnull @Override Iterable<TaskState> getTasks(Predicate<? super TaskState> predicate); @Override int countActiveTasksByDatastoreId(Long datastoreId); @Override @Nullable synchronized TaskState findTaskById(long id); @Nonnull @Override synchronized Iterable<? extends AbstractCaptureState> findActiveTasksForApp(
         final long appId); @Nonnull @Override synchronized Iterable<? extends TaskState> findActiveTasksForFeed(
         final long feedId); @Nullable @Override synchronized TaskState abortTask(long taskId); @Override synchronized boolean moveToHead(long taskIdToMove); @Override synchronized boolean moveToTail(long taskIdToMove); @Override synchronized boolean moveBefore(long taskIdToMove, long taskIdToFollow); @Override synchronized void cleanup(); @Override synchronized boolean cleanup(long taskId); @Nullable @Override synchronized TaskState unstallConversion(long taskId); @Override @Nonnull @VisibleForTesting List<Runnable> gracefulShutdown(boolean abortUnfinishedTasks); @Override synchronized void abortAllTasks(); @Override void destroy(); @Override void onApplicationEvent(ApplicationEvent applicationEvent); @Override void setApplicationEventPublisher(@Nullable ApplicationEventPublisher applicationEventPublisher); }
TaskQueueImpl implements
      TaskQueue,
      ApplicationListener,
      ApplicationEventPublisherAware,
      DisposableBean { @Override public synchronized void cleanup() { logInfoWithQueueName("purging completed tasks"); final List<Future<Void>> completedTasks = Lists.newArrayListWithExpectedSize(this.completedTasksQueue.size()); this.completedTasksQueue.drainTo(completedTasks); for (Future<Void> future: completedTasks) { AppFactoryTask task = FutureStoringThreadPoolExecutor.getOriginal(future); if (null != task) { removeTask(task); } else { logErrorWithQueueName("Could not find taskId for future {}", future); } } } TaskQueueImpl(String name,
                 int expectedTasksPerId,
                 Supplier<Integer> maxPoolSizeSupplier,
                 Supplier<Integer> maxFinishedSizeSupplier); @Override synchronized void addTask(@Nonnull AppFactoryTask task); @Nonnull @Override synchronized Iterable<TaskState> getAllTasks(); @Nonnull @Override Iterable<TaskState> getTasks(Predicate<? super TaskState> predicate); @Override int countActiveTasksByDatastoreId(Long datastoreId); @Override @Nullable synchronized TaskState findTaskById(long id); @Nonnull @Override synchronized Iterable<? extends AbstractCaptureState> findActiveTasksForApp(
         final long appId); @Nonnull @Override synchronized Iterable<? extends TaskState> findActiveTasksForFeed(
         final long feedId); @Nullable @Override synchronized TaskState abortTask(long taskId); @Override synchronized boolean moveToHead(long taskIdToMove); @Override synchronized boolean moveToTail(long taskIdToMove); @Override synchronized boolean moveBefore(long taskIdToMove, long taskIdToFollow); @Override synchronized void cleanup(); @Override synchronized boolean cleanup(long taskId); @Nullable @Override synchronized TaskState unstallConversion(long taskId); @Override @Nonnull @VisibleForTesting List<Runnable> gracefulShutdown(boolean abortUnfinishedTasks); @Override synchronized void abortAllTasks(); @Override void destroy(); @Override void onApplicationEvent(ApplicationEvent applicationEvent); @Override void setApplicationEventPublisher(@Nullable ApplicationEventPublisher applicationEventPublisher);  }