@Test public void testRun() { final int OP_COUNT = 10; addOperations(OP_COUNT, true, operationConductor.getPendingOperations()); doAnswer(new Answer() { @Override public Object answer(InvocationOnMock invocationOnMock) throws Throwable { OperationWrapper operationWrapper = (OperationWrapper)invocationOnMock.getArguments()[0]; OnOperationCompleteListener onOperationCompleteListener = (OnOperationCompleteListener)invocationOnMock.getArguments()[2]; onOperationCompleteListener.onComplete(operationWrapper); return null; } }).when(mockOperationRunnerFactory).createOperationRunner(any(OperationWrapper.class), any(ExecutorContext.class), any(OnOperationCompleteListener.class)); operationConductor.run(); Assert.assertEquals(operationConductor.getCompletedOperations().size(), OP_COUNT); }
public void run() { final int ORIGINAL_OP_COUNT = pendingOperations.size(); try { logger.debug("Loading Prior Progress (if available)"); loadPriorProgress(); logger.debug("Getting executor service"); if(executorService == null) { executorService = Executors.newFixedThreadPool( Integer.parseInt(executorContext.getLaunchDataWrapper().getPropertyRetriever().getField(THREAD_POOL_SIZE_SETTING, Integer.toString(DEFAULT_THREAD_POOL_SIZE))), new OperationRunnerThreadFactory("OpRunner-")); } logger.debug("Adding progress"); executorContext.getAgendaProgressReporter().addProgress(ProcessingState.EXECUTING, ExecutorMessages.OPERATIONS_RUNNING.getMessage()); logger.debug("Before drain/launch operations."); while (pendingOperations.size() > 0 || runningOperations.size() > 0) { drainPostProcessOperations(); launchReadyPendingOperations(); waitOnPostProcessOperations(); } } catch(Throwable t) { diagnosticEvents.add(new DiagnosticEvent(ExecutorMessages.OPERATIONS_ERROR.getMessage(), t)); logger.error(ExecutorMessages.OPERATIONS_ERROR.getMessage(), t); } finally { logger.info("Original OpCount: {} Completed OpCount: {}", ORIGINAL_OP_COUNT, completedOperations.size()); operationRunnerFactory.shutdown(); List<Runnable> remainingTasks = executorService.shutdownNow(); logger.info("ExecutorService shutdownNow called. {} Runnables were waiting.", remainingTasks.size()); } }
OperationConductor implements OnOperationCompleteListener { public void run() { final int ORIGINAL_OP_COUNT = pendingOperations.size(); try { logger.debug("Loading Prior Progress (if available)"); loadPriorProgress(); logger.debug("Getting executor service"); if(executorService == null) { executorService = Executors.newFixedThreadPool( Integer.parseInt(executorContext.getLaunchDataWrapper().getPropertyRetriever().getField(THREAD_POOL_SIZE_SETTING, Integer.toString(DEFAULT_THREAD_POOL_SIZE))), new OperationRunnerThreadFactory("OpRunner-")); } logger.debug("Adding progress"); executorContext.getAgendaProgressReporter().addProgress(ProcessingState.EXECUTING, ExecutorMessages.OPERATIONS_RUNNING.getMessage()); logger.debug("Before drain/launch operations."); while (pendingOperations.size() > 0 || runningOperations.size() > 0) { drainPostProcessOperations(); launchReadyPendingOperations(); waitOnPostProcessOperations(); } } catch(Throwable t) { diagnosticEvents.add(new DiagnosticEvent(ExecutorMessages.OPERATIONS_ERROR.getMessage(), t)); logger.error(ExecutorMessages.OPERATIONS_ERROR.getMessage(), t); } finally { logger.info("Original OpCount: {} Completed OpCount: {}", ORIGINAL_OP_COUNT, completedOperations.size()); operationRunnerFactory.shutdown(); List<Runnable> remainingTasks = executorService.shutdownNow(); logger.info("ExecutorService shutdownNow called. {} Runnables were waiting.", remainingTasks.size()); } } }
OperationConductor implements OnOperationCompleteListener { public void run() { final int ORIGINAL_OP_COUNT = pendingOperations.size(); try { logger.debug("Loading Prior Progress (if available)"); loadPriorProgress(); logger.debug("Getting executor service"); if(executorService == null) { executorService = Executors.newFixedThreadPool( Integer.parseInt(executorContext.getLaunchDataWrapper().getPropertyRetriever().getField(THREAD_POOL_SIZE_SETTING, Integer.toString(DEFAULT_THREAD_POOL_SIZE))), new OperationRunnerThreadFactory("OpRunner-")); } logger.debug("Adding progress"); executorContext.getAgendaProgressReporter().addProgress(ProcessingState.EXECUTING, ExecutorMessages.OPERATIONS_RUNNING.getMessage()); logger.debug("Before drain/launch operations."); while (pendingOperations.size() > 0 || runningOperations.size() > 0) { drainPostProcessOperations(); launchReadyPendingOperations(); waitOnPostProcessOperations(); } } catch(Throwable t) { diagnosticEvents.add(new DiagnosticEvent(ExecutorMessages.OPERATIONS_ERROR.getMessage(), t)); logger.error(ExecutorMessages.OPERATIONS_ERROR.getMessage(), t); } finally { logger.info("Original OpCount: {} Completed OpCount: {}", ORIGINAL_OP_COUNT, completedOperations.size()); operationRunnerFactory.shutdown(); List<Runnable> remainingTasks = executorService.shutdownNow(); logger.info("ExecutorService shutdownNow called. {} Runnables were waiting.", remainingTasks.size()); } } OperationConductor(Collection<Operation> operations, ExecutorContext executorContext,
        List<OperationConductorModifier> completedOperationModifiers);  OperationConductor(Collection<Operation> operations, ExecutorContext executorContext); }
OperationConductor implements OnOperationCompleteListener { public void run() { final int ORIGINAL_OP_COUNT = pendingOperations.size(); try { logger.debug("Loading Prior Progress (if available)"); loadPriorProgress(); logger.debug("Getting executor service"); if(executorService == null) { executorService = Executors.newFixedThreadPool( Integer.parseInt(executorContext.getLaunchDataWrapper().getPropertyRetriever().getField(THREAD_POOL_SIZE_SETTING, Integer.toString(DEFAULT_THREAD_POOL_SIZE))), new OperationRunnerThreadFactory("OpRunner-")); } logger.debug("Adding progress"); executorContext.getAgendaProgressReporter().addProgress(ProcessingState.EXECUTING, ExecutorMessages.OPERATIONS_RUNNING.getMessage()); logger.debug("Before drain/launch operations."); while (pendingOperations.size() > 0 || runningOperations.size() > 0) { drainPostProcessOperations(); launchReadyPendingOperations(); waitOnPostProcessOperations(); } } catch(Throwable t) { diagnosticEvents.add(new DiagnosticEvent(ExecutorMessages.OPERATIONS_ERROR.getMessage(), t)); logger.error(ExecutorMessages.OPERATIONS_ERROR.getMessage(), t); } finally { logger.info("Original OpCount: {} Completed OpCount: {}", ORIGINAL_OP_COUNT, completedOperations.size()); operationRunnerFactory.shutdown(); List<Runnable> remainingTasks = executorService.shutdownNow(); logger.info("ExecutorService shutdownNow called. {} Runnables were waiting.", remainingTasks.size()); } } OperationConductor(Collection<Operation> operations, ExecutorContext executorContext,
        List<OperationConductorModifier> completedOperationModifiers);  OperationConductor(Collection<Operation> operations, ExecutorContext executorContext); void run(); @Override void onComplete(OperationWrapper operationWrapper); List<OperationWrapper> getRunningOperations(); List<OperationWrapper> getPendingOperations(); List<OperationWrapper> getFailedOperations(); JsonContextUpdater getJsonContextUpdater(); List<DiagnosticEvent> retrieveAllDiagnosticEvents(); boolean hasExecutionFailed(); String getFailedOperationsDelimited(String delimiter); void setCompletedOperationModifiers(List<OperationConductorModifier> completedOperationModifiers); }
OperationConductor implements OnOperationCompleteListener { public void run() { final int ORIGINAL_OP_COUNT = pendingOperations.size(); try { logger.debug("Loading Prior Progress (if available)"); loadPriorProgress(); logger.debug("Getting executor service"); if(executorService == null) { executorService = Executors.newFixedThreadPool( Integer.parseInt(executorContext.getLaunchDataWrapper().getPropertyRetriever().getField(THREAD_POOL_SIZE_SETTING, Integer.toString(DEFAULT_THREAD_POOL_SIZE))), new OperationRunnerThreadFactory("OpRunner-")); } logger.debug("Adding progress"); executorContext.getAgendaProgressReporter().addProgress(ProcessingState.EXECUTING, ExecutorMessages.OPERATIONS_RUNNING.getMessage()); logger.debug("Before drain/launch operations."); while (pendingOperations.size() > 0 || runningOperations.size() > 0) { drainPostProcessOperations(); launchReadyPendingOperations(); waitOnPostProcessOperations(); } } catch(Throwable t) { diagnosticEvents.add(new DiagnosticEvent(ExecutorMessages.OPERATIONS_ERROR.getMessage(), t)); logger.error(ExecutorMessages.OPERATIONS_ERROR.getMessage(), t); } finally { logger.info("Original OpCount: {} Completed OpCount: {}", ORIGINAL_OP_COUNT, completedOperations.size()); operationRunnerFactory.shutdown(); List<Runnable> remainingTasks = executorService.shutdownNow(); logger.info("ExecutorService shutdownNow called. {} Runnables were waiting.", remainingTasks.size()); } } OperationConductor(Collection<Operation> operations, ExecutorContext executorContext,
        List<OperationConductorModifier> completedOperationModifiers);  OperationConductor(Collection<Operation> operations, ExecutorContext executorContext); void run(); @Override void onComplete(OperationWrapper operationWrapper); List<OperationWrapper> getRunningOperations(); List<OperationWrapper> getPendingOperations(); List<OperationWrapper> getFailedOperations(); JsonContextUpdater getJsonContextUpdater(); List<DiagnosticEvent> retrieveAllDiagnosticEvents(); boolean hasExecutionFailed(); String getFailedOperationsDelimited(String delimiter); void setCompletedOperationModifiers(List<OperationConductorModifier> completedOperationModifiers);  }