WorkflowServiceImpl extends AbstractIndexProducer implements WorkflowService, JobProducer, ManagedService { @Override public WorkflowInstanceImpl getWorkflowById(long id) throws NotFoundException, UnauthorizedException { try { Job job = serviceRegistry.getJob(id); if (Status.DELETED.equals(job.getStatus())) { throw new NotFoundException("Workflow '" + id + "' has been deleted"); } if (JOB_TYPE.equals(job.getJobType()) && Operation.START_WORKFLOW.toString().equals(job.getOperation())) { WorkflowInstanceImpl workflow = WorkflowParser.parseWorkflowInstance(job.getPayload()); assertPermission(workflow, Permissions.Action.READ.toString(), job.getOrganization()); return workflow; } else { throw new NotFoundException("'" + id + "' is a job identifier, but it is not a workflow identifier"); } } catch (WorkflowParsingException e) { throw new IllegalStateException("The workflow job payload is malformed"); } catch (ServiceRegistryException e) { throw new IllegalStateException("Error loading workflow job from the service registry"); } } WorkflowServiceImpl(); @Activate void activate(ComponentContext componentContext); @Override @Deactivate void deactivate(); @Override void addWorkflowListener(WorkflowListener listener); @Override void removeWorkflowListener(WorkflowListener listener); @Override List<WorkflowDefinition> listAvailableWorkflowDefinitions(); boolean isRunnable(WorkflowDefinition workflowDefinition); Set<HandlerRegistration> getRegisteredHandlers(); @Override WorkflowInstanceImpl getWorkflowById(long id); @Override WorkflowInstance start(WorkflowDefinition workflowDefinition, MediaPackage mediaPackage); @Override WorkflowInstance start(WorkflowDefinition workflowDefinition, MediaPackage mediaPackage,
          Map<String, String> properties); @Override WorkflowInstance start(WorkflowDefinition workflowDefinition, MediaPackage sourceMediaPackage,
          Long parentWorkflowId, Map<String, String> originalProperties); @Override WorkflowDefinition getWorkflowDefinitionById(String id); @Override WorkflowInstance stop(long workflowInstanceId); @Override void remove(long workflowInstanceId); @Override void remove(long workflowInstanceId, boolean force); @Override WorkflowInstance suspend(long workflowInstanceId); @Override WorkflowInstance resume(long id); @Override WorkflowInstance resume(long workflowInstanceId, Map<String, String> properties); @Override void update(final WorkflowInstance workflowInstance); @Override long countWorkflowInstances(); @Override long countWorkflowInstances(WorkflowState state, String operation); @Override WorkflowStatistics getStatistics(); @Override WorkflowSet getWorkflowInstances(WorkflowQuery query); @Override WorkflowSet getWorkflowInstancesForAdministrativeRead(WorkflowQuery query); @Override boolean isReadyToAcceptJobs(String operation); @Override boolean isReadyToAccept(Job job); @Override synchronized void acceptJob(Job job); @Override long countJobs(Status status); ServiceRegistry getServiceRegistry(); @Reference(name = "security-service") void setSecurityService(SecurityService securityService); @Reference(name = "authorization") void setAuthorizationService(AuthorizationService authorizationService); @Reference(name = "user-directory") void setUserDirectoryService(UserDirectoryService userDirectoryService); @Reference(name = "orgDirectory") void setOrganizationDirectoryService(OrganizationDirectoryService organizationDirectory); @Reference(name = "series") void setSeriesService(SeriesService seriesService); @Reference(name = "assetManager") void setAssetManager(AssetManager assetManager); @Reference(name = "message-broker-receiver") void setMessageReceiver(MessageReceiver messageReceiver); @Reference(name = "message-broker-sender") void setMessageSender(MessageSender messageSender); @Override String getJobType(); @Override @SuppressWarnings("rawtypes") void updated(Dictionary properties); @Override synchronized void cleanupWorkflowInstances(int buffer, WorkflowState state); @Override Map<String, Map<String, String>> getWorkflowStateMappings(); @Override void repopulate(final String indexName); @Override MessageReceiver getMessageReceiver(); @Override Service getService(); @Override String getClassName(); @Override MessageSender getMessageSender(); @Override SecurityService getSecurityService(); @Override String getSystemUserName(); static final String STATS_COLLECT_CONFIG_KEY; static final Boolean DEFAULT_STATS_COLLECT_CONFIG; static final String ERROR_RESOLUTION_HANDLER_ID; }
@Test public void testFailingOperationWithErrorHandler() throws Exception { WorkflowInstance instance = startAndWait(failingDefinitionWithErrorHandler, mediapackage1, WorkflowState.FAILED); Assert.assertEquals(WorkflowState.FAILED, service.getWorkflowById(instance.getId()).getState()); Assert.assertEquals(OperationState.FAILED, service.getWorkflowById(instance.getId()).getOperations().get(1) .getState()); WorkflowInstance storedInstance = service.getWorkflowById(instance.getId()); Assert.assertEquals(4, storedInstance.getOperations().size()); Assert.assertEquals("op1", storedInstance.getOperations().get(0).getTemplate()); Assert.assertEquals("op3", storedInstance.getOperations().get(1).getTemplate()); Assert.assertEquals("op1", storedInstance.getOperations().get(2).getTemplate()); Assert.assertEquals("op2", storedInstance.getOperations().get(3).getTemplate()); }
@Test public void testFailingOperationWithoutErrorHandler() throws Exception { WorkflowInstance instance = startAndWait(failingDefinitionWithoutErrorHandler, mediapackage1, WorkflowState.FAILED); Assert.assertEquals(WorkflowState.FAILED, service.getWorkflowById(instance.getId()).getState()); }
@Test public void testRetryStrategyNone() throws Exception { WorkflowDefinitionImpl def = new WorkflowDefinitionImpl(); def.setId("workflow-definition-1"); def.setTitle("workflow-definition-1"); def.setDescription("workflow-definition-1"); WorkflowOperationDefinitionImpl opDef = new WorkflowOperationDefinitionImpl("failOneTime", "fails once", null, true); def.add(opDef); MediaPackage mp = MediaPackageBuilderFactory.newInstance().newMediaPackageBuilder().createNew(); WorkflowInstance workflow = startAndWait(def, mp, WorkflowState.FAILED); Assert.assertTrue(service.getWorkflowById(workflow.getId()).getOperations().get(0).getState() == OperationState.FAILED); Assert.assertTrue(service.getWorkflowById(workflow.getId()).getOperations().get(0).getMaxAttempts() == 1); Assert.assertTrue(service.getWorkflowById(workflow.getId()).getOperations().get(0).getFailedAttempts() == 1); }
@Test public void testRetryStrategyRetry() throws Exception { WorkflowDefinitionImpl def = new WorkflowDefinitionImpl(); def.setId("workflow-definition-1"); def.setTitle("workflow-definition-1"); def.setDescription("workflow-definition-1"); WorkflowOperationDefinitionImpl opDef = new WorkflowOperationDefinitionImpl("failOneTime", "fails once", null, true); opDef.setRetryStrategy(RetryStrategy.RETRY); def.add(opDef); MediaPackage mp = MediaPackageBuilderFactory.newInstance().newMediaPackageBuilder().createNew(); WorkflowInstance workflow = startAndWait(def, mp, WorkflowState.SUCCEEDED); Assert.assertTrue(service.getWorkflowById(workflow.getId()).getOperations().get(0).getState() == OperationState.SUCCEEDED); Assert.assertTrue(service.getWorkflowById(workflow.getId()).getOperations().get(0).getMaxAttempts() == 2); Assert.assertTrue(service.getWorkflowById(workflow.getId()).getOperations().get(0).getFailedAttempts() == 1); }
@Test public void testRetryStrategyHold() throws Exception { WorkflowDefinitionImpl def = new WorkflowDefinitionImpl(); def.setId("workflow-definition-1"); def.setTitle("workflow-definition-1"); def.setDescription("workflow-definition-1"); WorkflowOperationDefinitionImpl opDef = new WorkflowOperationDefinitionImpl("failOneTime", "fails once", null, true); opDef.setRetryStrategy(RetryStrategy.HOLD); opDef.setMaxAttempts(2); def.add(opDef); MediaPackage mp = MediaPackageBuilderFactory.newInstance().newMediaPackageBuilder().createNew(); WorkflowInstance workflow = startAndWait(def, mp, WorkflowState.PAUSED); WorkflowOperationInstance errorResolutionOperation = service.getWorkflowById(workflow.getId()).getOperations() .get(0); WorkflowOperationInstance failOneTimeOperation = service.getWorkflowById(workflow.getId()).getOperations().get(1); Assert.assertTrue(errorResolutionOperation.getTemplate().equals(WorkflowServiceImpl.ERROR_RESOLUTION_HANDLER_ID)); Assert.assertTrue(errorResolutionOperation.getState() == OperationState.PAUSED); Assert.assertTrue(errorResolutionOperation.getFailedAttempts() == 0); Assert.assertTrue(failOneTimeOperation.getState() == OperationState.RETRY); Assert.assertTrue(failOneTimeOperation.getMaxAttempts() == 2); Assert.assertTrue(failOneTimeOperation.getFailedAttempts() == 1); }
@Test public void testRetryStrategyHoldMaxAttemptsThree() throws Exception { WorkflowDefinitionImpl def = new WorkflowDefinitionImpl(); def.setId("workflow-definition-1"); def.setTitle("workflow-definition-1"); def.setDescription("workflow-definition-1"); WorkflowOperationDefinitionImpl opDef = new WorkflowOperationDefinitionImpl("failTwice", "fails twice", null, true); opDef.setRetryStrategy(RetryStrategy.HOLD); opDef.setMaxAttempts(3); def.add(opDef); MediaPackage mp = MediaPackageBuilderFactory.newInstance().newMediaPackageBuilder().createNew(); WorkflowInstance workflow = startAndWait(def, mp, WorkflowState.PAUSED); WorkflowOperationInstance errorResolutionOperation = service.getWorkflowById(workflow.getId()).getOperations() .get(0); WorkflowOperationInstance failTwiceOperation = service.getWorkflowById(workflow.getId()).getOperations().get(1); Assert.assertTrue(errorResolutionOperation.getTemplate().equals(WorkflowServiceImpl.ERROR_RESOLUTION_HANDLER_ID)); Assert.assertTrue(errorResolutionOperation.getState() == OperationState.PAUSED); Assert.assertTrue(errorResolutionOperation.getFailedAttempts() == 0); Assert.assertTrue("failTwice".equals(failTwiceOperation.getTemplate())); Assert.assertTrue(failTwiceOperation.getState() == OperationState.RETRY); Assert.assertTrue(failTwiceOperation.getMaxAttempts() == 3); Assert.assertTrue(failTwiceOperation.getFailedAttempts() == 1); retryAndWait(service.getWorkflowById(workflow.getId()), "RETRY", WorkflowState.PAUSED); errorResolutionOperation = service.getWorkflowById(workflow.getId()).getOperations().get(1); failTwiceOperation = service.getWorkflowById(workflow.getId()).getOperations().get(2); Assert.assertTrue(errorResolutionOperation.getTemplate().equals(WorkflowServiceImpl.ERROR_RESOLUTION_HANDLER_ID)); Assert.assertTrue(errorResolutionOperation.getState() == OperationState.PAUSED); Assert.assertTrue(errorResolutionOperation.getFailedAttempts() == 0); Assert.assertTrue("failTwice".equals(failTwiceOperation.getTemplate())); Assert.assertTrue(failTwiceOperation.getState() == OperationState.RETRY); Assert.assertTrue(failTwiceOperation.getMaxAttempts() == 3); Assert.assertTrue(failTwiceOperation.getFailedAttempts() == 2); retryAndWait(service.getWorkflowById(workflow.getId()), "RETRY", WorkflowState.SUCCEEDED); failTwiceOperation = service.getWorkflowById(workflow.getId()).getOperations().get(2); Assert.assertTrue("failTwice".equals(failTwiceOperation.getTemplate())); Assert.assertTrue(failTwiceOperation.getState() == OperationState.SUCCEEDED); Assert.assertTrue(failTwiceOperation.getMaxAttempts() == 3); Assert.assertTrue(failTwiceOperation.getFailedAttempts() == 2); }
@Test public void testRetryStrategyHoldFailureByUser() throws Exception { WorkflowDefinitionImpl def = new WorkflowDefinitionImpl(); def.setId("workflow-definition-1"); def.setTitle("workflow-definition-1"); def.setDescription("workflow-definition-1"); WorkflowOperationDefinitionImpl opDef = new WorkflowOperationDefinitionImpl("failTwice", "fails twice", null, true); opDef.setRetryStrategy(RetryStrategy.HOLD); opDef.setMaxAttempts(3); opDef.setFailWorkflowOnException(true); def.add(opDef); MediaPackage mp = MediaPackageBuilderFactory.newInstance().newMediaPackageBuilder().createNew(); WorkflowInstance workflow = startAndWait(def, mp, WorkflowState.PAUSED); WorkflowOperationInstance errorResolutionOperation = service.getWorkflowById(workflow.getId()).getOperations() .get(0); WorkflowOperationInstance failTwiceOperation = service.getWorkflowById(workflow.getId()).getOperations().get(1); Assert.assertTrue(errorResolutionOperation.getTemplate().equals(WorkflowServiceImpl.ERROR_RESOLUTION_HANDLER_ID)); Assert.assertTrue(errorResolutionOperation.getState() == OperationState.PAUSED); Assert.assertTrue(errorResolutionOperation.getFailedAttempts() == 0); Assert.assertTrue("failTwice".equals(failTwiceOperation.getTemplate())); Assert.assertTrue(failTwiceOperation.getState() == OperationState.RETRY); Assert.assertTrue(failTwiceOperation.getMaxAttempts() == 3); Assert.assertTrue(failTwiceOperation.getFailedAttempts() == 1); retryAndWait(service.getWorkflowById(workflow.getId()), "RETRY", WorkflowState.PAUSED); errorResolutionOperation = service.getWorkflowById(workflow.getId()).getOperations().get(1); failTwiceOperation = service.getWorkflowById(workflow.getId()).getOperations().get(2); Assert.assertTrue(errorResolutionOperation.getTemplate().equals(WorkflowServiceImpl.ERROR_RESOLUTION_HANDLER_ID)); Assert.assertTrue(errorResolutionOperation.getState() == OperationState.PAUSED); Assert.assertTrue(errorResolutionOperation.getFailedAttempts() == 0); Assert.assertTrue("failTwice".equals(failTwiceOperation.getTemplate())); Assert.assertTrue(failTwiceOperation.getState() == OperationState.RETRY); Assert.assertTrue(failTwiceOperation.getMaxAttempts() == 3); Assert.assertTrue(failTwiceOperation.getFailedAttempts() == 2); retryAndWait(service.getWorkflowById(workflow.getId()), "NONE", WorkflowState.FAILED); failTwiceOperation = service.getWorkflowById(workflow.getId()).getOperations().get(2); Assert.assertTrue("failTwice".equals(failTwiceOperation.getTemplate())); Assert.assertTrue(failTwiceOperation.getState() == OperationState.FAILED); Assert.assertTrue(failTwiceOperation.getMaxAttempts() == 3); Assert.assertTrue(failTwiceOperation.getFailedAttempts() == 2); }