Skip to content

Commit

Permalink
test
Browse files Browse the repository at this point in the history
  • Loading branch information
@saksham2105
saksham2105 committed Jul 16, 2024
1 parent 20ffca3 commit 7906e00
Show file tree
Hide file tree
Showing 5 changed files with 25 additions and 82 deletions.
6 changes: 1 addition & 5 deletions src/main/java/io/orkes/samples/models/Distribution.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,9 +8,5 @@
public class Distribution {
private String translation;
private String distributeTo;
private String sendToORB = "Y";
private String taskType = "HTTP";
private String natsWorkflowName;
private Integer natsWorkflowVersion;
private Map<String, Object> natsWorkflowInput;
private OrbpFlags orbpFlags;
}
7 changes: 1 addition & 6 deletions src/main/java/io/orkes/samples/models/Enrichment.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,14 +2,9 @@

import lombok.Data;

import java.util.Map;

@Data
public class Enrichment {
private String enrichmentType;
private String sendToORB = "N";
private String taskType = "HTTP";
private String natsWorkflowName;
private Integer natsWorkflowVersion;
private Map<String, Object> natsWorkflowInput;
private OrbpFlags orbpFlags;
}
10 changes: 10 additions & 0 deletions src/main/java/io/orkes/samples/models/OrbpFlags.java
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
package io.orkes.samples.models;

import lombok.Data;

@Data
public class OrbpFlags {
private String audit = "N";
private String parse = "N";
private String status = "N";
}
7 changes: 1 addition & 6 deletions src/main/java/io/orkes/samples/models/Translation.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,10 +9,5 @@
public class Translation {
private String name;
private List<String> enrichments;
private String sendToORB = "N";
private String taskType = "HTTP";
private String natsWorkflowName;
private Integer natsWorkflowVersion;
private Map<String, Object> natsWorkflowInput;

private OrbpFlags orbpFlags;
}
77 changes: 12 additions & 65 deletions src/main/java/io/orkes/samples/workers/DynamicSubworkflowWorker.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,19 +5,16 @@
import com.netflix.conductor.client.worker.Worker;
import com.netflix.conductor.common.metadata.tasks.Task;
import com.netflix.conductor.common.metadata.tasks.TaskResult;
import com.netflix.conductor.common.metadata.workflow.StartWorkflowRequest;
import com.netflix.conductor.common.metadata.workflow.WorkflowDef;
import com.netflix.conductor.sdk.workflow.def.ConductorWorkflow;
import com.netflix.conductor.sdk.workflow.def.tasks.*;
import com.netflix.conductor.sdk.workflow.executor.WorkflowExecutor;
import io.orkes.conductor.client.WorkflowClient;
import io.orkes.samples.models.MediationRules;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;


import java.util.HashMap;
import java.util.List;
import java.util.Map;

Expand All @@ -26,13 +23,12 @@
@AllArgsConstructor
public class DynamicSubworkflowWorker implements Worker {

private final WorkflowClient workflowClient;
private final ObjectMapper objectMapper = new ObjectMapper();
private final WorkflowExecutor executor;

@Override
public String getTaskDefName() {
return "quest_start_subworkflow";
return "create_dynamic_workflow_def";
}

/**
Expand All @@ -42,7 +38,7 @@ public String getTaskDefName() {
*/
@Override
public TaskResult execute(Task task) {
System.out.println("Starting quest_start_subworkflow task");
System.out.println("Starting create_dynamic_workflow_def task");
TaskResult result = new TaskResult(task);
try {
MediationRules mediationRules = objectMapper.convertValue(task.getInputData().get("mediation_rules"), MediationRules.class);
Expand All @@ -55,17 +51,8 @@ public TaskResult execute(Task task) {
}

public Map<String, Object> startExistingWorkflow(MediationRules mediationRules) throws JsonProcessingException {
StartWorkflowRequest request = new StartWorkflowRequest();
request.setName("dynamic_workflow");
Map<String, Object> inputData = new HashMap<>();
//inputData.put("enrichmentSubworkflowsDef", subworkflowDef());
Object dynamicSubworkflowDef = objectMapper.convertValue(createDynamicSubworkflow(mediationRules), Object.class);
inputData.put("dynamicSubworkflowDef", dynamicSubworkflowDef);
request.setInput(inputData);

String workflowId = workflowClient.startWorkflow(request);
log.info("Workflow id: {}", workflowId);
return Map.of("workflowId", workflowId);
return Map.of("workflow_def", dynamicSubworkflowDef);
}

private WorkflowDef createDynamicSubworkflow(MediationRules mediationRules) throws JsonProcessingException {
Expand All @@ -83,20 +70,8 @@ private WorkflowDef createDynamicSubworkflow(MediationRules mediationRules) thro
// --------------- Enrichment level started ------------------
com.netflix.conductor.sdk.workflow.def.tasks.Task[][] enrichmentForkTasks = new com.netflix.conductor.sdk.workflow.def.tasks.Task[mediationRules.getEnrichments().size()][1];
for (int i = 0; i < mediationRules.getEnrichments().size(); i++) {
ConductorWorkflow conductorWorkflow = new ConductorWorkflow(executor);
conductorWorkflow.setName(mediationRules.getEnrichments().get(i).getEnrichmentType() + "_workflow");

SubWorkflow natsSubworkflow = new SubWorkflow("nats_" + mediationRules.getEnrichments().get(i).getEnrichmentType() + "_subworkflow_ref", mediationRules.getEnrichments().get(i).getNatsWorkflowName(), mediationRules.getEnrichments().get(i).getNatsWorkflowVersion());
natsSubworkflow.input(mediationRules.getEnrichments().get(i).getNatsWorkflowInput());
Switch sendToORBEnrichmentSwitch = new Switch("send_to_" + mediationRules.getEnrichments().get(i).getEnrichmentType() + "_switch", "${workflow.input.sendToORB}").switchCase("Y", natsSubworkflow).defaultCase(List.of());
conductorWorkflow.add(sendToORBEnrichmentSwitch);

Http httptask = new Http(mediationRules.getEnrichments().get(i).getEnrichmentType() + "_enrichment_workflow_task");
httptask.url("https://orkes-api-tester.orkesconductor.com/api");
conductorWorkflow.add(httptask);

SubWorkflow forkSubworkflow = new SubWorkflow(mediationRules.getEnrichments().get(i).getEnrichmentType() + "_subworkflow_ref", conductorWorkflow);
forkSubworkflow.input("sendToORB", mediationRules.getEnrichments().get(i).getSendToORB());
SubWorkflow forkSubworkflow = new SubWorkflow(mediationRules.getEnrichments().get(i).getEnrichmentType() + "_subworkflow_ref", "OP_" + mediationRules.getEnrichments().get(i).getEnrichmentType(), 1);
forkSubworkflow.input("sendToORB", mediationRules.getEnrichments().get(i).getOrbpFlags());
enrichmentForkTasks[i][0] = forkSubworkflow;
}
ForkJoin forkEnrichment = new ForkJoin("fork_enrichment", enrichmentForkTasks);
Expand All @@ -107,24 +82,11 @@ private WorkflowDef createDynamicSubworkflow(MediationRules mediationRules) thro
// -------------- Translation Level started ----------------
com.netflix.conductor.sdk.workflow.def.tasks.Task[][] translationForkTasks = new com.netflix.conductor.sdk.workflow.def.tasks.Task[mediationRules.getTranslations().size()][1];
for (int i = 0; i < mediationRules.getTranslations().size(); i++) {
ConductorWorkflow conductorWorkflow = new ConductorWorkflow(executor);
SubWorkflow forkSubworkflow = new SubWorkflow(mediationRules.getTranslations().get(i).getName() + "_subworkflow_ref", conductorWorkflow);
forkSubworkflow.input("sendToORB", mediationRules.getTranslations().get(i).getSendToORB());
conductorWorkflow.setName(mediationRules.getTranslations().get(i).getName() + "_workflow");
SubWorkflow natsSubworkflow = new SubWorkflow("nats_" + mediationRules.getTranslations().get(i).getName() + "_subworkflow_ref", mediationRules.getTranslations().get(i).getNatsWorkflowName(), mediationRules.getTranslations().get(i).getNatsWorkflowVersion());
natsSubworkflow.input(mediationRules.getTranslations().get(i).getNatsWorkflowInput());
Switch sendToORBTranslationSwitch = new Switch("send_to_" + mediationRules.getTranslations().get(i).getName() + "_switch", "${workflow.input.sendToORB}").switchCase("Y", natsSubworkflow).defaultCase(List.of());
conductorWorkflow.add(sendToORBTranslationSwitch);

for (int j = 0; j < mediationRules.getTranslations().get(i).getEnrichments().size(); j++) {
Http httptask = new Http(mediationRules.getTranslations().get(i).getEnrichments().get(j)+ "_translations_workflow_task");
httptask.url("https://orkes-api-tester.orkesconductor.com/api");
String taskRef = mediationRules.getTranslations().get(i).getEnrichments().get(j) + "_subworkflow_ref";
String outputExpression = "${" + taskRef + ".output.response}"; //Can differ with different different tasks. Example with Simple/Inline tasks we might have to use result
forkSubworkflow.input(mediationRules.getTranslations().get(i).getEnrichments().get(j), outputExpression);
conductorWorkflow.add(httptask);
SubWorkflow forkSubworkflow = new SubWorkflow(mediationRules.getTranslations().get(i).getName() + "_subworkflow_ref", "OP_" + mediationRules.getTranslations().get(i).getName(), 1);
forkSubworkflow.input("sendToORB", mediationRules.getTranslations().get(i).getOrbpFlags());
for (String enrichmentInput : mediationRules.getTranslations().get(i).getEnrichments()) {
forkSubworkflow.input(enrichmentInput, "${" + enrichmentInput + "_subworkflow_ref" + ".output.result}");
}

translationForkTasks[i][0] = forkSubworkflow;
}
ForkJoin forkTranslation = new ForkJoin("fork_translation", translationForkTasks);
Expand All @@ -135,24 +97,9 @@ private WorkflowDef createDynamicSubworkflow(MediationRules mediationRules) thro
// -------------- Distribution level started --------------------
com.netflix.conductor.sdk.workflow.def.tasks.Task[][] distributionForkTasks = new com.netflix.conductor.sdk.workflow.def.tasks.Task[mediationRules.getDistributions().size()][1];
for (int i = 0; i < mediationRules.getDistributions().size(); i++) {
ConductorWorkflow conductorWorkflow = new ConductorWorkflow(executor);
conductorWorkflow.setName(mediationRules.getDistributions().get(i).getDistributeTo() + "_workflow");

SubWorkflow natsSubworkflow = new SubWorkflow("nats_" + mediationRules.getDistributions().get(i).getDistributeTo() + "_subworkflow_ref", mediationRules.getDistributions().get(i).getNatsWorkflowName(), mediationRules.getDistributions().get(i).getNatsWorkflowVersion());
natsSubworkflow.input(mediationRules.getDistributions().get(i).getNatsWorkflowInput());
Switch sendToORBDistributionSwitch = new Switch("send_to_" + mediationRules.getDistributions().get(i).getDistributeTo() + "_switch", "${workflow.input.sendToORB}").switchCase("Y", natsSubworkflow).defaultCase(List.of());
conductorWorkflow.add(sendToORBDistributionSwitch);

Http httptask = new Http(mediationRules.getDistributions().get(i).getDistributeTo() + "_distributions_workflow_task");
httptask.url("https://orkes-api-tester.orkesconductor.com/api");
conductorWorkflow.add(httptask);

SubWorkflow forkSubworkflow = new SubWorkflow(mediationRules.getDistributions().get(i).getDistributeTo() + "_subworkflow_ref", conductorWorkflow);
forkSubworkflow.input("sendToORB", mediationRules.getDistributions().get(i).getSendToORB());
String taskRef = mediationRules.getDistributions().get(i).getTranslation() + "_subworkflow_ref";
String outputExpression = "${" + taskRef + ".output.response}"; //Can differ with different different tasks. Example with Simple/Inline tasks we might have to use result
forkSubworkflow.input(mediationRules.getDistributions().get(i).getTranslation(), outputExpression);
forkSubworkflow.input("sink", "nats:nats-integ:subject");
SubWorkflow forkSubworkflow = new SubWorkflow(mediationRules.getDistributions().get(i).getDistributeTo() + "_subworkflow_ref", "OP_" + mediationRules.getDistributions().get(i).getDistributeTo(), 1);
forkSubworkflow.input("sendToORB", mediationRules.getDistributions().get(i).getOrbpFlags());
forkSubworkflow.input(mediationRules.getDistributions().get(i).getTranslation(), "${" + mediationRules.getDistributions().get(i).getTranslation() + "_subworkflow_ref" + ".output.result}");
distributionForkTasks[i][0] = forkSubworkflow;
}
ForkJoin forkDistribution = new ForkJoin("fork_distribution", distributionForkTasks);
Expand Down

0 comments on commit 7906e00

Please sign in to comment.