@Test(expected = ProcessingEngineException.class) @RunWithCustomExecutor public void logbookBeforeDistributorCallKO() throws Exception { VitamThreadUtils.getVitamSession().setTenantId(TENANT_ID); final ProcessWorkflow processWorkflow = processData.initProcessWorkflow(populate(WORKFLOW_FILE), workParams.getContainerName() ); doThrow(new LogbookClientServerException("")).when(logbookOperationsClient).update(any()); processEngine.start(processWorkflow.getSteps().get(0), workParams, PauseRecover.NO_RECOVER); }
@Override public void start(ProcessStep step, WorkerParameters workerParameters, PauseRecover pauseRecover) throws ProcessingEngineException { if (null == stateMachineCallback) { throw new ProcessingEngineException("IEventsProcessEngine is required"); } if (null == step) { throw new ProcessingEngineException("The parameter step cannot be null"); } if (null != workerParameters) { for (final WorkerParameterName key : workerParameters.getMapParameters().keySet()) { if (this.workerParameters.getParameterValue(key) == null) { this.workerParameters.putParameterValue(key, workerParameters.getMapParameters().get(key)); } } } final String operationId = this.workerParameters.getContainerName(); if (LOGGER.isDebugEnabled()) { LOGGER.debug("Start Workflow: " + step.getId() + " Step:" + step.getStepName()); } final int tenantId = ParameterHelper.getTenantParameter(); final LogbookTypeProcess logbookTypeProcess = this.workerParameters.getLogbookTypeProcess(); LogbookOperationParameters logbookParameter = logbookBeforeDistributorCall(step, this.workerParameters, tenantId, logbookTypeProcess); if (!PauseOrCancelAction.ACTION_RECOVER.equals(step.getPauseOrCancelAction()) && !PauseOrCancelAction.ACTION_REPLAY.equals(step.getPauseOrCancelAction())) { stateMachineCallback.onUpdate(StatusCode.STARTED); } this.workerParameters.setCurrentStep(step.getStepName()); this.workerParameters.putParameterValue(WorkerParameterName.workflowStatusKo, stateMachineCallback.getCurrentProcessWorkflowStatus().name()); Stopwatch stopwatch = Stopwatch.createStarted(); CompletableFuture .supplyAsync(() -> callDistributor(step, this.workerParameters, operationId, pauseRecover), VitamThreadPoolExecutor.getDefaultExecutor()) .thenApply(distributorResponse -> { try { switch (step.getPauseOrCancelAction()) { case ACTION_PAUSE: return distributorResponse; default: } logbookAfterDistributorCall(step, this.workerParameters, tenantId, logbookTypeProcess, logbookParameter, distributorResponse); return distributorResponse; } catch (Exception e) { throw new RuntimeException(e); } }) .thenApply(distributorResponse -> { try { switch (step.getPauseOrCancelAction()) { case ACTION_CANCEL: stateMachineCallback.onProcessEngineCancel(this.workerParameters); return distributorResponse; } stateMachineCallback.onProcessEngineCompleteStep(distributorResponse, this.workerParameters); } finally { PERFORMANCE_LOGGER.log(step.getStepName(), stopwatch.elapsed(TimeUnit.MILLISECONDS)); } return distributorResponse; }) .exceptionally((e) -> { LOGGER.error("Error while process workflow", e); stateMachineCallback.onError(e); PERFORMANCE_LOGGER.log(step.getStepName(), stopwatch.elapsed(TimeUnit.MILLISECONDS)); return null; }); }
ProcessEngineImpl implements ProcessEngine { @Override public void start(ProcessStep step, WorkerParameters workerParameters, PauseRecover pauseRecover) throws ProcessingEngineException { if (null == stateMachineCallback) { throw new ProcessingEngineException("IEventsProcessEngine is required"); } if (null == step) { throw new ProcessingEngineException("The parameter step cannot be null"); } if (null != workerParameters) { for (final WorkerParameterName key : workerParameters.getMapParameters().keySet()) { if (this.workerParameters.getParameterValue(key) == null) { this.workerParameters.putParameterValue(key, workerParameters.getMapParameters().get(key)); } } } final String operationId = this.workerParameters.getContainerName(); if (LOGGER.isDebugEnabled()) { LOGGER.debug("Start Workflow: " + step.getId() + " Step:" + step.getStepName()); } final int tenantId = ParameterHelper.getTenantParameter(); final LogbookTypeProcess logbookTypeProcess = this.workerParameters.getLogbookTypeProcess(); LogbookOperationParameters logbookParameter = logbookBeforeDistributorCall(step, this.workerParameters, tenantId, logbookTypeProcess); if (!PauseOrCancelAction.ACTION_RECOVER.equals(step.getPauseOrCancelAction()) && !PauseOrCancelAction.ACTION_REPLAY.equals(step.getPauseOrCancelAction())) { stateMachineCallback.onUpdate(StatusCode.STARTED); } this.workerParameters.setCurrentStep(step.getStepName()); this.workerParameters.putParameterValue(WorkerParameterName.workflowStatusKo, stateMachineCallback.getCurrentProcessWorkflowStatus().name()); Stopwatch stopwatch = Stopwatch.createStarted(); CompletableFuture .supplyAsync(() -> callDistributor(step, this.workerParameters, operationId, pauseRecover), VitamThreadPoolExecutor.getDefaultExecutor()) .thenApply(distributorResponse -> { try { switch (step.getPauseOrCancelAction()) { case ACTION_PAUSE: return distributorResponse; default: } logbookAfterDistributorCall(step, this.workerParameters, tenantId, logbookTypeProcess, logbookParameter, distributorResponse); return distributorResponse; } catch (Exception e) { throw new RuntimeException(e); } }) .thenApply(distributorResponse -> { try { switch (step.getPauseOrCancelAction()) { case ACTION_CANCEL: stateMachineCallback.onProcessEngineCancel(this.workerParameters); return distributorResponse; } stateMachineCallback.onProcessEngineCompleteStep(distributorResponse, this.workerParameters); } finally { PERFORMANCE_LOGGER.log(step.getStepName(), stopwatch.elapsed(TimeUnit.MILLISECONDS)); } return distributorResponse; }) .exceptionally((e) -> { LOGGER.error("Error while process workflow", e); stateMachineCallback.onError(e); PERFORMANCE_LOGGER.log(step.getStepName(), stopwatch.elapsed(TimeUnit.MILLISECONDS)); return null; }); } }
ProcessEngineImpl implements ProcessEngine { @Override public void start(ProcessStep step, WorkerParameters workerParameters, PauseRecover pauseRecover) throws ProcessingEngineException { if (null == stateMachineCallback) { throw new ProcessingEngineException("IEventsProcessEngine is required"); } if (null == step) { throw new ProcessingEngineException("The parameter step cannot be null"); } if (null != workerParameters) { for (final WorkerParameterName key : workerParameters.getMapParameters().keySet()) { if (this.workerParameters.getParameterValue(key) == null) { this.workerParameters.putParameterValue(key, workerParameters.getMapParameters().get(key)); } } } final String operationId = this.workerParameters.getContainerName(); if (LOGGER.isDebugEnabled()) { LOGGER.debug("Start Workflow: " + step.getId() + " Step:" + step.getStepName()); } final int tenantId = ParameterHelper.getTenantParameter(); final LogbookTypeProcess logbookTypeProcess = this.workerParameters.getLogbookTypeProcess(); LogbookOperationParameters logbookParameter = logbookBeforeDistributorCall(step, this.workerParameters, tenantId, logbookTypeProcess); if (!PauseOrCancelAction.ACTION_RECOVER.equals(step.getPauseOrCancelAction()) && !PauseOrCancelAction.ACTION_REPLAY.equals(step.getPauseOrCancelAction())) { stateMachineCallback.onUpdate(StatusCode.STARTED); } this.workerParameters.setCurrentStep(step.getStepName()); this.workerParameters.putParameterValue(WorkerParameterName.workflowStatusKo, stateMachineCallback.getCurrentProcessWorkflowStatus().name()); Stopwatch stopwatch = Stopwatch.createStarted(); CompletableFuture .supplyAsync(() -> callDistributor(step, this.workerParameters, operationId, pauseRecover), VitamThreadPoolExecutor.getDefaultExecutor()) .thenApply(distributorResponse -> { try { switch (step.getPauseOrCancelAction()) { case ACTION_PAUSE: return distributorResponse; default: } logbookAfterDistributorCall(step, this.workerParameters, tenantId, logbookTypeProcess, logbookParameter, distributorResponse); return distributorResponse; } catch (Exception e) { throw new RuntimeException(e); } }) .thenApply(distributorResponse -> { try { switch (step.getPauseOrCancelAction()) { case ACTION_CANCEL: stateMachineCallback.onProcessEngineCancel(this.workerParameters); return distributorResponse; } stateMachineCallback.onProcessEngineCompleteStep(distributorResponse, this.workerParameters); } finally { PERFORMANCE_LOGGER.log(step.getStepName(), stopwatch.elapsed(TimeUnit.MILLISECONDS)); } return distributorResponse; }) .exceptionally((e) -> { LOGGER.error("Error while process workflow", e); stateMachineCallback.onError(e); PERFORMANCE_LOGGER.log(step.getStepName(), stopwatch.elapsed(TimeUnit.MILLISECONDS)); return null; }); } ProcessEngineImpl(WorkerParameters workerParameters, ProcessDistributor processDistributor,
        LogbookOperationsClientFactory logbookOperationsClientFactory); }
ProcessEngineImpl implements ProcessEngine { @Override public void start(ProcessStep step, WorkerParameters workerParameters, PauseRecover pauseRecover) throws ProcessingEngineException { if (null == stateMachineCallback) { throw new ProcessingEngineException("IEventsProcessEngine is required"); } if (null == step) { throw new ProcessingEngineException("The parameter step cannot be null"); } if (null != workerParameters) { for (final WorkerParameterName key : workerParameters.getMapParameters().keySet()) { if (this.workerParameters.getParameterValue(key) == null) { this.workerParameters.putParameterValue(key, workerParameters.getMapParameters().get(key)); } } } final String operationId = this.workerParameters.getContainerName(); if (LOGGER.isDebugEnabled()) { LOGGER.debug("Start Workflow: " + step.getId() + " Step:" + step.getStepName()); } final int tenantId = ParameterHelper.getTenantParameter(); final LogbookTypeProcess logbookTypeProcess = this.workerParameters.getLogbookTypeProcess(); LogbookOperationParameters logbookParameter = logbookBeforeDistributorCall(step, this.workerParameters, tenantId, logbookTypeProcess); if (!PauseOrCancelAction.ACTION_RECOVER.equals(step.getPauseOrCancelAction()) && !PauseOrCancelAction.ACTION_REPLAY.equals(step.getPauseOrCancelAction())) { stateMachineCallback.onUpdate(StatusCode.STARTED); } this.workerParameters.setCurrentStep(step.getStepName()); this.workerParameters.putParameterValue(WorkerParameterName.workflowStatusKo, stateMachineCallback.getCurrentProcessWorkflowStatus().name()); Stopwatch stopwatch = Stopwatch.createStarted(); CompletableFuture .supplyAsync(() -> callDistributor(step, this.workerParameters, operationId, pauseRecover), VitamThreadPoolExecutor.getDefaultExecutor()) .thenApply(distributorResponse -> { try { switch (step.getPauseOrCancelAction()) { case ACTION_PAUSE: return distributorResponse; default: } logbookAfterDistributorCall(step, this.workerParameters, tenantId, logbookTypeProcess, logbookParameter, distributorResponse); return distributorResponse; } catch (Exception e) { throw new RuntimeException(e); } }) .thenApply(distributorResponse -> { try { switch (step.getPauseOrCancelAction()) { case ACTION_CANCEL: stateMachineCallback.onProcessEngineCancel(this.workerParameters); return distributorResponse; } stateMachineCallback.onProcessEngineCompleteStep(distributorResponse, this.workerParameters); } finally { PERFORMANCE_LOGGER.log(step.getStepName(), stopwatch.elapsed(TimeUnit.MILLISECONDS)); } return distributorResponse; }) .exceptionally((e) -> { LOGGER.error("Error while process workflow", e); stateMachineCallback.onError(e); PERFORMANCE_LOGGER.log(step.getStepName(), stopwatch.elapsed(TimeUnit.MILLISECONDS)); return null; }); } ProcessEngineImpl(WorkerParameters workerParameters, ProcessDistributor processDistributor,
        LogbookOperationsClientFactory logbookOperationsClientFactory); @Override void setStateMachineCallback(IEventsProcessEngine stateMachineCallback); @Override void start(ProcessStep step, WorkerParameters workerParameters, PauseRecover pauseRecover); }
ProcessEngineImpl implements ProcessEngine { @Override public void start(ProcessStep step, WorkerParameters workerParameters, PauseRecover pauseRecover) throws ProcessingEngineException { if (null == stateMachineCallback) { throw new ProcessingEngineException("IEventsProcessEngine is required"); } if (null == step) { throw new ProcessingEngineException("The parameter step cannot be null"); } if (null != workerParameters) { for (final WorkerParameterName key : workerParameters.getMapParameters().keySet()) { if (this.workerParameters.getParameterValue(key) == null) { this.workerParameters.putParameterValue(key, workerParameters.getMapParameters().get(key)); } } } final String operationId = this.workerParameters.getContainerName(); if (LOGGER.isDebugEnabled()) { LOGGER.debug("Start Workflow: " + step.getId() + " Step:" + step.getStepName()); } final int tenantId = ParameterHelper.getTenantParameter(); final LogbookTypeProcess logbookTypeProcess = this.workerParameters.getLogbookTypeProcess(); LogbookOperationParameters logbookParameter = logbookBeforeDistributorCall(step, this.workerParameters, tenantId, logbookTypeProcess); if (!PauseOrCancelAction.ACTION_RECOVER.equals(step.getPauseOrCancelAction()) && !PauseOrCancelAction.ACTION_REPLAY.equals(step.getPauseOrCancelAction())) { stateMachineCallback.onUpdate(StatusCode.STARTED); } this.workerParameters.setCurrentStep(step.getStepName()); this.workerParameters.putParameterValue(WorkerParameterName.workflowStatusKo, stateMachineCallback.getCurrentProcessWorkflowStatus().name()); Stopwatch stopwatch = Stopwatch.createStarted(); CompletableFuture .supplyAsync(() -> callDistributor(step, this.workerParameters, operationId, pauseRecover), VitamThreadPoolExecutor.getDefaultExecutor()) .thenApply(distributorResponse -> { try { switch (step.getPauseOrCancelAction()) { case ACTION_PAUSE: return distributorResponse; default: } logbookAfterDistributorCall(step, this.workerParameters, tenantId, logbookTypeProcess, logbookParameter, distributorResponse); return distributorResponse; } catch (Exception e) { throw new RuntimeException(e); } }) .thenApply(distributorResponse -> { try { switch (step.getPauseOrCancelAction()) { case ACTION_CANCEL: stateMachineCallback.onProcessEngineCancel(this.workerParameters); return distributorResponse; } stateMachineCallback.onProcessEngineCompleteStep(distributorResponse, this.workerParameters); } finally { PERFORMANCE_LOGGER.log(step.getStepName(), stopwatch.elapsed(TimeUnit.MILLISECONDS)); } return distributorResponse; }) .exceptionally((e) -> { LOGGER.error("Error while process workflow", e); stateMachineCallback.onError(e); PERFORMANCE_LOGGER.log(step.getStepName(), stopwatch.elapsed(TimeUnit.MILLISECONDS)); return null; }); } ProcessEngineImpl(WorkerParameters workerParameters, ProcessDistributor processDistributor,
        LogbookOperationsClientFactory logbookOperationsClientFactory); @Override void setStateMachineCallback(IEventsProcessEngine stateMachineCallback); @Override void start(ProcessStep step, WorkerParameters workerParameters, PauseRecover pauseRecover); static final String DETAILS; }