vusakhoza/velocity-pay-springboot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

changin batch integration to asynch

Browse Source
This commit is contained in:
Prince
2026-06-17 21:38:55 +02:00
parent 6c3ef60665
commit dcb917e899
3 changed files with 42 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
src/main/java/zw/qantra/tm/domain/services/GroupBatchService.java
View File

@@ -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<Object> 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<Object> 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);
}

7
src/main/java/zw/qantra/tm/domain/services/processors/workflows/BatchWorkflow.java
View File

@@ -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));

10
src/main/java/zw/qantra/tm/domain/services/processors/workflows/WorkflowUtils.java
View File

@@ -26,12 +26,12 @@ public class WorkflowUtils {
CompletableFuture<Object> future = new CompletableFuture<>();
while(!future.isDone()){ // check if the future is done
try {
Thread.sleep(pollingInterval);
WorkflowInstance instance = workflowInstanceService.getWorkflowInstance(
workflowId,
EnumSet.of(ACTIONS, ACTION_STATE_VARIABLES, CURRENT_STATE_VARIABLES),
1L);
try {
Thread.sleep(pollingInterval);
List<String> stateList = Arrays.stream(states).toList();
List<WorkflowInstance.WorkflowInstanceStatus> statusList =
Arrays.asList(WorkflowInstance.WorkflowInstanceStatus.manual,
@@ -40,13 +40,13 @@ public class WorkflowUtils {
statusList.contains(instance.status)){
future.complete(getStateResponse(instance, "success"));
}
} catch (InterruptedException e) {
future.completeExceptionally(e);
}
timeout -= pollingInterval;
if(timeout <= 0){
future.complete(getStateResponse(instance, "Request timed out"));
}
} catch (InterruptedException e) {
future.completeExceptionally(e);
}
}
return future;