Tinkering

Author: Evanthx

core/src/main/java/io/temporal/samples/nexus_messaging/callerpattern/README.md

@@ -1,6 +1,6 @@
 ## Entity pattern
 
-The handler worker starts a single `GreetingWorkflow` on boot with a fixed workflow ID.
+The handler worker starts a single `GreetingWorkflow` with a fixed workflow ID.
 `NexusGreetingServiceImpl` holds that ID and routes every Nexus operation to it. The caller's
 inputs contain only business data — no workflow IDs.
 

core/src/main/java/io/temporal/samples/nexus_messaging/callerpattern/caller/CallerStarter.java

@@ -24,7 +24,7 @@ public static void main(String[] args) {
                 .setTaskQueue(CallerWorker.TASK_QUEUE)
                 .build());
 
-    List<String> log = workflow.run();
+    List<String> log = workflow.run("user-1");
     log.forEach(System.out::println);
   }
 }

core/src/main/java/io/temporal/samples/nexus_messaging/callerpattern/caller/CallerWorkflow.java

@@ -7,5 +7,5 @@
 @WorkflowInterface
 public interface CallerWorkflow {
   @WorkflowMethod
-  List<String> run();
+  List<String> run(String userId);
 }

core/src/main/java/io/temporal/samples/nexus_messaging/callerpattern/caller/CallerWorkflowImpl.java

@@ -28,22 +28,23 @@ public class CallerWorkflowImpl implements CallerWorkflow {
               .build());
 
   @Override
-  public List<String> run() {
+  public List<String> run(String userId) {
     List<String> log = new ArrayList<>();
 
     // 👉 Call a Nexus operation backed by a query against the entity workflow.
     NexusGreetingService.GetLanguagesOutput languagesOutput =
-        greetingService.getLanguages(new NexusGreetingService.GetLanguagesInput(false));
+        greetingService.getLanguages(new NexusGreetingService.GetLanguagesInput(false, userId));
     log.add("Supported languages: " + languagesOutput.getLanguages());
     logger.info("Supported languages: {}", languagesOutput.getLanguages());
 
     // 👉 Call a Nexus operation backed by an update against the entity workflow.
     Language previousLanguage =
-        greetingService.setLanguage(new NexusGreetingService.SetLanguageInput(Language.ARABIC));
+        greetingService.setLanguage(
+            new NexusGreetingService.SetLanguageInput(Language.ARABIC, userId));
 
     // 👉 Call a Nexus operation backed by a query to confirm the language change.
     Language currentLanguage =
-        greetingService.getLanguage(new NexusGreetingService.GetLanguageInput());
+        greetingService.getLanguage(new NexusGreetingService.GetLanguageInput(userId));
     if (currentLanguage != Language.ARABIC) {
       throw ApplicationFailure.newFailure(
           "Expected language ARABIC, got " + currentLanguage, "AssertionError");
@@ -53,7 +54,7 @@ public List<String> run() {
     logger.info("Language changed from {} to {}", previousLanguage, Language.ARABIC);
 
     // 👉 Call a Nexus operation backed by a signal against the entity workflow.
-    greetingService.approve(new NexusGreetingService.ApproveInput("caller"));
+    greetingService.approve(new NexusGreetingService.ApproveInput("caller", userId));
     log.add("Workflow approved");
     logger.info("Workflow approved");
 

core/src/main/java/io/temporal/samples/nexus_messaging/callerpattern/handler/GreetingWorkflow.java

@@ -1,7 +1,5 @@
 package io.temporal.samples.nexus_messaging.callerpattern.handler;
 
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
 import io.temporal.samples.nexus_messaging.callerpattern.service.Language;
 import io.temporal.samples.nexus_messaging.callerpattern.service.NexusGreetingService;
 import io.temporal.workflow.QueryMethod;
@@ -19,20 +17,6 @@
 @WorkflowInterface
 public interface GreetingWorkflow {
 
-  class ApproveInput {
-    private final String name;
-
-    @JsonCreator(mode = JsonCreator.Mode.PROPERTIES)
-    public ApproveInput(@JsonProperty("name") String name) {
-      this.name = name;
-    }
-
-    @JsonProperty("name")
-    public String getName() {
-      return name;
-    }
-  }
-
   @WorkflowMethod
   String run();
 
@@ -47,7 +31,7 @@ NexusGreetingService.GetLanguagesOutput getLanguages(
 
   // Approves the workflow, allowing it to complete.
   @SignalMethod
-  void approve(ApproveInput input);
+  void approve(NexusGreetingService.ApproveInput input);
 
   // Changes the active language synchronously (only supports languages already in the greetings
   // map).

core/src/main/java/io/temporal/samples/nexus_messaging/callerpattern/handler/GreetingWorkflowImpl.java

@@ -58,22 +58,28 @@ public Language getLanguage() {
   }
 
   @Override
-  public void approve(ApproveInput input) {
-    logger.info("Approval signal received");
+  public void approve(NexusGreetingService.ApproveInput input) {
+    logger.info(
+        "Approval signal received for workflow {}",
+        NexusGreetingServiceImpl.getWorkflowId(input.getUserId()));
     approvedForRelease = true;
   }
 
   @Override
   public Language setLanguage(NexusGreetingService.SetLanguageInput input) {
-    logger.info("setLanguage update received");
+    logger.info(
+        "setLanguage update received for workflow {}",
+        NexusGreetingServiceImpl.getWorkflowId(input.getUserId()));
     Language previous = language;
     language = input.getLanguage();
     return previous;
   }
 
   @Override
   public void validateSetLanguage(NexusGreetingService.SetLanguageInput input) {
-    logger.info("validateSetLanguage called");
+    logger.info(
+        "validateSetLanguage called for workflow {}",
+        NexusGreetingServiceImpl.getWorkflowId(input.getUserId()));
     if (!greetings.containsKey(input.getLanguage())) {
       throw new IllegalArgumentException(input.getLanguage().name() + " is not supported");
     }

core/src/main/java/io/temporal/samples/nexus_messaging/callerpattern/handler/HandlerWorker.java

@@ -15,7 +15,7 @@ public class HandlerWorker {
 
   public static final String NAMESPACE = "nexus-messaging-handler-namespace";
   public static final String TASK_QUEUE = "nexus-messaging-handler-task-queue";
-  static final String WORKFLOW_ID = "nexus-messaging-greeting-workflow";
+  static final String USER_ID = "user-1";
 
   public static void main(String[] args) throws InterruptedException {
     WorkflowServiceStubs service = WorkflowServiceStubs.newLocalServiceStubs();
@@ -24,25 +24,28 @@ public static void main(String[] args) throws InterruptedException {
             service, WorkflowClientOptions.newBuilder().setNamespace(NAMESPACE).build());
 
     // Start the long-running entity workflow that backs the Nexus service, if not already running.
+    // The workflow ID is derived from the user ID using the same prefix as
+    // NexusGreetingServiceImpl.
+    String workflowId = NexusGreetingServiceImpl.getWorkflowId(USER_ID);
     GreetingWorkflow greetingWorkflow =
         client.newWorkflowStub(
             GreetingWorkflow.class,
             WorkflowOptions.newBuilder()
-                .setWorkflowId(WORKFLOW_ID)
+                .setWorkflowId(workflowId)
                 .setTaskQueue(TASK_QUEUE)
                 .build());
     try {
       WorkflowClient.start(greetingWorkflow::run);
-      logger.info("Started greeting workflow: {}", WORKFLOW_ID);
+      logger.info("Started greeting workflow: {}", workflowId);
     } catch (WorkflowExecutionAlreadyStarted e) {
-      logger.info("Greeting workflow already running: {}", WORKFLOW_ID);
+      logger.info("Greeting workflow already running: {}", workflowId);
     }
 
     WorkerFactory factory = WorkerFactory.newInstance(client);
     Worker worker = factory.newWorker(TASK_QUEUE);
     worker.registerWorkflowImplementationTypes(GreetingWorkflowImpl.class);
     worker.registerActivitiesImplementations(new GreetingActivityImpl());
-    worker.registerNexusServiceImplementation(new NexusGreetingServiceImpl(WORKFLOW_ID));
+    worker.registerNexusServiceImplementation(new NexusGreetingServiceImpl());
 
     factory.start();
     logger.info("Handler worker started, ctrl+c to exit");

core/src/main/java/io/temporal/samples/nexus_messaging/callerpattern/handler/NexusGreetingServiceImpl.java

@@ -10,25 +10,25 @@
 import org.slf4j.LoggerFactory;
 
 /**
- * Nexus operation handler implementation. Each operation is backed by the long-running
- * GreetingWorkflow entity. The operations are synchronous (sync_operation) because queries and
+ * Nexus operation handler implementation. Each operation receives a userId, which is mapped to a
+ * workflow ID using {@link #WORKFLOW_ID_PREFIX}. The operations are synchronous because queries and
  * updates against a running workflow complete quickly.
  */
 @ServiceImpl(service = NexusGreetingService.class)
 public class NexusGreetingServiceImpl {
 
   private static final Logger logger = LoggerFactory.getLogger(NexusGreetingServiceImpl.class);
 
-  private final String workflowId;
+  static final String WORKFLOW_ID_PREFIX = "GreetingWorkflow_for_";
 
-  public NexusGreetingServiceImpl(String workflowId) {
-    this.workflowId = workflowId;
+  public static String getWorkflowId(String userId) {
+    return WORKFLOW_ID_PREFIX + userId;
   }
 
-  private GreetingWorkflow getWorkflowStub() {
+  private GreetingWorkflow getWorkflowStub(String userId) {
     return Nexus.getOperationContext()
         .getWorkflowClient()
-        .newWorkflowStub(GreetingWorkflow.class, workflowId);
+        .newWorkflowStub(GreetingWorkflow.class, getWorkflowId(userId));
   }
 
   @OperationImpl
@@ -37,17 +37,17 @@ private GreetingWorkflow getWorkflowStub() {
       getLanguages() {
     return OperationHandler.sync(
         (ctx, details, input) -> {
-          logger.info("Query for GetLanguages was received");
-          return getWorkflowStub().getLanguages(input);
+          logger.info("Query for GetLanguages was received for user {}", input.getUserId());
+          return getWorkflowStub(input.getUserId()).getLanguages(input);
         });
   }
 
   @OperationImpl
   public OperationHandler<NexusGreetingService.GetLanguageInput, Language> getLanguage() {
     return OperationHandler.sync(
         (ctx, details, input) -> {
-          logger.info("Query for GetLanguage was received");
-          return getWorkflowStub().getLanguage();
+          logger.info("Query for GetLanguage was received for user {}", input.getUserId());
+          return getWorkflowStub(input.getUserId()).getLanguage();
         });
   }
 
@@ -57,8 +57,8 @@ public OperationHandler<NexusGreetingService.GetLanguageInput, Language> getLang
   public OperationHandler<NexusGreetingService.SetLanguageInput, Language> setLanguage() {
     return OperationHandler.sync(
         (ctx, details, input) -> {
-          logger.info("Update for SetLanguage was received");
-          return getWorkflowStub().setLanguageUsingActivity(input);
+          logger.info("Update for SetLanguage was received for user {}", input.getUserId());
+          return getWorkflowStub(input.getUserId()).setLanguageUsingActivity(input);
         });
   }
 
@@ -67,8 +67,8 @@ public OperationHandler<NexusGreetingService.SetLanguageInput, Language> setLang
       approve() {
     return OperationHandler.sync(
         (ctx, details, input) -> {
-          logger.info("Signal for Approve was received");
-          getWorkflowStub().approve(new GreetingWorkflow.ApproveInput(input.getName()));
+          logger.info("Signal for Approve was received for user {}", input.getUserId());
+          getWorkflowStub(input.getUserId()).approve(input);
           return new NexusGreetingService.ApproveOutput();
         });
   }

core/src/main/java/io/temporal/samples/nexus_messaging/callerpattern/service/NexusGreetingService.java

@@ -16,16 +16,25 @@ public interface NexusGreetingService {
 
   class GetLanguagesInput {
     private final boolean includeUnsupported;
+    private final String userId;
 
     @JsonCreator(mode = JsonCreator.Mode.PROPERTIES)
-    public GetLanguagesInput(@JsonProperty("includeUnsupported") boolean includeUnsupported) {
+    public GetLanguagesInput(
+        @JsonProperty("includeUnsupported") boolean includeUnsupported,
+        @JsonProperty("userId") String userId) {
       this.includeUnsupported = includeUnsupported;
+      this.userId = userId;
     }
 
     @JsonProperty("includeUnsupported")
     public boolean isIncludeUnsupported() {
       return includeUnsupported;
     }
+
+    @JsonProperty("userId")
+    public String getUserId() {
+      return userId;
+    }
   }
 
   class GetLanguagesOutput {
@@ -42,24 +51,39 @@ public List<Language> getLanguages() {
     }
   }
 
-  @JsonAutoDetect(fieldVisibility = JsonAutoDetect.Visibility.ANY)
   class GetLanguageInput {
-    @JsonCreator
-    public GetLanguageInput() {}
+    private final String userId;
+
+    @JsonCreator(mode = JsonCreator.Mode.PROPERTIES)
+    public GetLanguageInput(@JsonProperty("userId") String userId) {
+      this.userId = userId;
+    }
+
+    @JsonProperty("userId")
+    public String getUserId() {
+      return userId;
+    }
   }
 
   class ApproveInput {
     private final String name;
+    private final String userId;
 
     @JsonCreator(mode = JsonCreator.Mode.PROPERTIES)
-    public ApproveInput(@JsonProperty("name") String name) {
+    public ApproveInput(@JsonProperty("name") String name, @JsonProperty("userId") String userId) {
       this.name = name;
+      this.userId = userId;
     }
 
     @JsonProperty("name")
     public String getName() {
       return name;
     }
+
+    @JsonProperty("userId")
+    public String getUserId() {
+      return userId;
+    }
   }
 
   @JsonAutoDetect(fieldVisibility = JsonAutoDetect.Visibility.ANY)
@@ -70,16 +94,24 @@ public ApproveOutput() {}
 
   class SetLanguageInput {
     private final Language language;
+    private final String userId;
 
     @JsonCreator(mode = JsonCreator.Mode.PROPERTIES)
-    public SetLanguageInput(@JsonProperty("language") Language language) {
+    public SetLanguageInput(
+        @JsonProperty("language") Language language, @JsonProperty("userId") String userId) {
       this.language = language;
+      this.userId = userId;
     }
 
     @JsonProperty("language")
     public Language getLanguage() {
       return language;
     }
+
+    @JsonProperty("userId")
+    public String getUserId() {
+      return userId;
+    }
   }
 
   // Returns the languages supported by the greeting workflow.