diff --git a/src/main/java/zw/qantra/tm/domain/services/GroupBatchService.java b/src/main/java/zw/qantra/tm/domain/services/GroupBatchService.java index 6a039b0..da0305e 100644 --- a/src/main/java/zw/qantra/tm/domain/services/GroupBatchService.java +++ b/src/main/java/zw/qantra/tm/domain/services/GroupBatchService.java @@ -16,11 +16,7 @@ import zw.qantra.tm.domain.dtos.GroupBatchCreateRequest; import zw.qantra.tm.domain.dtos.GroupBatchItemUpdateRequest; import zw.qantra.tm.domain.dtos.GroupBatchUpdateRequest; import zw.qantra.tm.domain.dtos.StateResponse; -import zw.qantra.tm.domain.enums.AuthType; -import zw.qantra.tm.domain.enums.GroupBatchItemConfirmStatus; -import zw.qantra.tm.domain.enums.GroupBatchItemIntegrationStatus; -import zw.qantra.tm.domain.enums.GroupBatchItemStatus; -import zw.qantra.tm.domain.enums.GroupBatchStatus; +import zw.qantra.tm.domain.enums.*; import zw.qantra.tm.domain.models.*; import zw.qantra.tm.domain.repositories.GroupBatchItemRepository; import zw.qantra.tm.domain.repositories.GroupBatchRepository; @@ -209,7 +205,7 @@ public class GroupBatchService { try { CompletableFuture future = workflowUtils .waitForState(batch.getWorkflowId(), new String[]{"requestSuccess", "done", "failed"}, - 15000, 50); + 45000, 50); StateResponse response = (StateResponse) future.get(); String id = response.getBody().toString(); // this returns Batch Transaction ID String cleaned = id.replaceAll("^\"|\"$", ""); @@ -234,6 +230,7 @@ public class GroupBatchService { throw new ApiException("Batch workflow has not been started. Trigger confirm first."); } + // STEP 1: TRIGGER POLL SYNCHRONOUSLY WorkflowInstance update = new WorkflowInstance.Builder() .setId(batch.getWorkflowId()) .setState("poll") @@ -252,13 +249,35 @@ public class GroupBatchService { workflowInstanceService.updateWorkflowInstance(update, action); try { + // wait for poll to finish CompletableFuture future = workflowUtils - .waitForState(batch.getWorkflowId(), new String[]{"done", "failed"}, - 15000, 50); + .waitForState(batch.getWorkflowId(), new String[]{"pollSuccess", "failed"}, + 45000, 50); StateResponse response = (StateResponse) future.get(); String id = response.getBody().toString(); // this returns Batch Transaction ID String cleaned = id.replaceAll("^\"|\"$", ""); - return transactionService.findById(UUID.fromString(cleaned)); + Transaction transaction = transactionService.findById(UUID.fromString(cleaned)); + + // if failed then return + if(!transaction.getPollingStatus().equals(Status.SUCCESS)) { + return transaction; + } + + // STEP 2: TRIGGER INTEGRATION ASYNCHRONOUSLY + WorkflowInstance integrationUpdate = new WorkflowInstance.Builder() + .setId(batch.getWorkflowId()) + .setState("integration") + .setNextActivation(DateTime.now()) + .setStateText("Resumed from network trigger") + .build(); + + WorkflowInstanceAction integrationAction = new WorkflowInstanceAction.Builder(integrationUpdate) + .setType(WorkflowInstanceAction.WorkflowActionType.externalChange) + .setExecutionEnd(DateTime.now()) + .build(); + + workflowInstanceService.updateWorkflowInstance(integrationUpdate, integrationAction); + return transaction; } catch (InterruptedException | ExecutionException e) { throw new RuntimeException(e); } diff --git a/src/main/java/zw/qantra/tm/domain/services/processors/workflows/BatchWorkflow.java b/src/main/java/zw/qantra/tm/domain/services/processors/workflows/BatchWorkflow.java index 36c437f..e009291 100644 --- a/src/main/java/zw/qantra/tm/domain/services/processors/workflows/BatchWorkflow.java +++ b/src/main/java/zw/qantra/tm/domain/services/processors/workflows/BatchWorkflow.java @@ -24,6 +24,7 @@ public class BatchWorkflow extends WorkflowDefinition { public static final State REQUEST = new State("request"); public static final State REQUEST_SUCCESS = new State("requestSuccess", WorkflowStateType.manual); public static final State POLL = new State("poll"); + public static final State POLL_SUCCESS = new State("pollSuccess", WorkflowStateType.manual); public static final State INTEGRATION = new State("integration"); public static final State FAILED = new State("failed", WorkflowStateType.manual); public static final State DONE = new State("done", WorkflowStateType.end); @@ -81,12 +82,16 @@ public class BatchWorkflow extends WorkflowDefinition { UUID batchId = UUID.fromString(execution.getVariable("batchId", String.class)); GroupBatch batch = groupBatchWorkflowService.processPoll(batchId); execution.setVariable("batchStatus", batch.getStatus().name()); - return NextAction.moveToState(INTEGRATION, "Batch poll complete"); + return NextAction.moveToState(POLL_SUCCESS, "Batch poll complete"); } catch (Exception e) { return NextAction.moveToState(FAILED, e.getMessage()); } } + public void pollSuccess(StateExecution execution) { + log.info("Batch poll succeeded, paused for integration trigger"); + } + public NextAction integration(StateExecution execution) { try { UUID batchId = UUID.fromString(execution.getVariable("batchId", String.class)); diff --git a/src/main/java/zw/qantra/tm/domain/services/processors/workflows/WorkflowUtils.java b/src/main/java/zw/qantra/tm/domain/services/processors/workflows/WorkflowUtils.java index 0c5cad0..c39f01c 100644 --- a/src/main/java/zw/qantra/tm/domain/services/processors/workflows/WorkflowUtils.java +++ b/src/main/java/zw/qantra/tm/domain/services/processors/workflows/WorkflowUtils.java @@ -26,12 +26,12 @@ public class WorkflowUtils { CompletableFuture future = new CompletableFuture<>(); while(!future.isDone()){ // check if the future is done - WorkflowInstance instance = workflowInstanceService.getWorkflowInstance( - workflowId, - EnumSet.of(ACTIONS, ACTION_STATE_VARIABLES, CURRENT_STATE_VARIABLES), - 1L); try { Thread.sleep(pollingInterval); + WorkflowInstance instance = workflowInstanceService.getWorkflowInstance( + workflowId, + EnumSet.of(ACTIONS, ACTION_STATE_VARIABLES, CURRENT_STATE_VARIABLES), + 1L); List stateList = Arrays.stream(states).toList(); List statusList = Arrays.asList(WorkflowInstance.WorkflowInstanceStatus.manual, @@ -40,13 +40,13 @@ public class WorkflowUtils { statusList.contains(instance.status)){ future.complete(getStateResponse(instance, "success")); } + timeout -= pollingInterval; + if(timeout <= 0){ + future.complete(getStateResponse(instance, "Request timed out")); + } } catch (InterruptedException e) { future.completeExceptionally(e); } - timeout -= pollingInterval; - if(timeout <= 0){ - future.complete(getStateResponse(instance, "Request timed out")); - } } return future;