Add NexusInfo

Author: Quinn-With-Two-Ns

temporal-sdk/src/main/java/io/temporal/common/interceptors/NexusOperationOutboundCallsInterceptor.java

@@ -3,6 +3,7 @@
 import com.uber.m3.tally.Scope;
 import io.temporal.client.WorkflowClient;
 import io.temporal.common.Experimental;
+import io.temporal.nexus.NexusInfo;
 
 /**
  * Can be used to intercept calls from a Nexus operation into the Temporal APIs.
@@ -20,6 +21,9 @@
  */
 @Experimental
 public interface NexusOperationOutboundCallsInterceptor {
+  /** Intercepts call to get the Nexus info in a Nexus operation. */
+  NexusInfo getInfo();
+
   /** Intercepts call to get the metric scope in a Nexus operation. */
   Scope getMetricsScope();
 

temporal-sdk/src/main/java/io/temporal/common/interceptors/NexusOperationOutboundCallsInterceptorBase.java

@@ -3,6 +3,7 @@
 import com.uber.m3.tally.Scope;
 import io.temporal.client.WorkflowClient;
 import io.temporal.common.Experimental;
+import io.temporal.nexus.NexusInfo;
 
 /** Convenience base class for {@link NexusOperationOutboundCallsInterceptor} implementations. */
 @Experimental
@@ -14,6 +15,11 @@ public NexusOperationOutboundCallsInterceptorBase(NexusOperationOutboundCallsInt
     this.next = next;
   }
 
+  @Override
+  public NexusInfo getInfo() {
+    return next.getInfo();
+  }
+
   @Override
   public Scope getMetricsScope() {
     return next.getMetricsScope();

temporal-sdk/src/main/java/io/temporal/internal/nexus/InternalNexusOperationContext.java

@@ -4,6 +4,7 @@
 import io.temporal.api.common.v1.Link;
 import io.temporal.client.WorkflowClient;
 import io.temporal.common.interceptors.NexusOperationOutboundCallsInterceptor;
+import io.temporal.nexus.NexusInfo;
 import io.temporal.nexus.NexusOperationContext;
 
 public class InternalNexusOperationContext {
@@ -58,6 +59,11 @@ public Link getStartWorkflowResponseLink() {
   }
 
   private class NexusOperationContextImpl implements NexusOperationContext {
+    @Override
+    public NexusInfo getInfo() {
+      return outboundCalls.getInfo();
+    }
+
     @Override
     public Scope getMetricsScope() {
       return outboundCalls.getMetricsScope();

temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusInfoImpl.java

@@ -0,0 +1,23 @@
+package io.temporal.internal.nexus;
+
+import io.temporal.nexus.NexusInfo;
+
+class NexusInfoImpl implements NexusInfo {
+  private final String namespace;
+  private final String taskQueue;
+
+  NexusInfoImpl(String namespace, String taskQueue) {
+    this.namespace = namespace;
+    this.taskQueue = taskQueue;
+  }
+
+  @Override
+  public String getNamespace() {
+    return namespace;
+  }
+
+  @Override
+  public String getTaskQueue() {
+    return taskQueue;
+  }
+}

temporal-sdk/src/main/java/io/temporal/internal/nexus/RootNexusOperationOutboundCallsInterceptor.java

@@ -3,15 +3,24 @@
 import com.uber.m3.tally.Scope;
 import io.temporal.client.WorkflowClient;
 import io.temporal.common.interceptors.NexusOperationOutboundCallsInterceptor;
+import io.temporal.nexus.NexusInfo;
 
 public class RootNexusOperationOutboundCallsInterceptor
     implements NexusOperationOutboundCallsInterceptor {
   private final Scope scope;
   private final WorkflowClient client;
+  private final NexusInfo nexusInfo;
 
-  RootNexusOperationOutboundCallsInterceptor(Scope scope, WorkflowClient client) {
+  RootNexusOperationOutboundCallsInterceptor(
+      Scope scope, WorkflowClient client, NexusInfo nexusInfo) {
     this.scope = scope;
     this.client = client;
+    this.nexusInfo = nexusInfo;
+  }
+
+  @Override
+  public NexusInfo getInfo() {
+    return nexusInfo;
   }
 
   @Override

temporal-sdk/src/main/java/io/temporal/internal/nexus/TemporalInterceptorMiddleware.java

@@ -27,7 +27,10 @@ public OperationHandler<Object, Object> intercept(
     InternalNexusOperationContext temporalNexusContext = CurrentNexusOperationContext.get();
     inboundCallsInterceptor.init(
         new RootNexusOperationOutboundCallsInterceptor(
-            temporalNexusContext.getMetricsScope(), temporalNexusContext.getWorkflowClient()));
+            temporalNexusContext.getMetricsScope(),
+            temporalNexusContext.getWorkflowClient(),
+            new NexusInfoImpl(
+                temporalNexusContext.getNamespace(), temporalNexusContext.getTaskQueue())));
     return new OperationInterceptorConverter(inboundCallsInterceptor);
   }
 

temporal-sdk/src/main/java/io/temporal/nexus/NexusInfo.java

@@ -0,0 +1,17 @@
+package io.temporal.nexus;
+
+/**
+ * Temporal information about the Nexus Operation. Use {@link NexusOperationContext#getInfo()} from
+ * a Nexus Operation implementation to access.
+ */
+public interface NexusInfo {
+  /**
+   * @return Namespace of the worker that is executing the Nexus Operation
+   */
+  String getNamespace();
+
+  /**
+   * @return Nexus Task Queue of the worker that is executing the Nexus Operation
+   */
+  String getTaskQueue();
+}

temporal-sdk/src/main/java/io/temporal/nexus/NexusOperationContext.java

@@ -10,6 +10,9 @@
  */
 public interface NexusOperationContext {
 
+  /** Get Temporal information about the Nexus Operation. */
+  NexusInfo getInfo();
+
   /**
    * Get scope for reporting business metrics in a nexus handler. This scope is tagged with the
    * service and operation.

temporal-sdk/src/test/java/io/temporal/workflow/nexus/NexusOperationInfoTest.java

@@ -0,0 +1,54 @@
+package io.temporal.workflow.nexus;
+
+import io.nexusrpc.handler.OperationHandler;
+import io.nexusrpc.handler.OperationImpl;
+import io.nexusrpc.handler.ServiceImpl;
+import io.temporal.nexus.Nexus;
+import io.temporal.nexus.NexusInfo;
+import io.temporal.testing.internal.SDKTestWorkflowRule;
+import io.temporal.workflow.*;
+import io.temporal.workflow.shared.TestNexusServices;
+import io.temporal.workflow.shared.TestWorkflows;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+
+public class NexusOperationInfoTest {
+  @Rule
+  public SDKTestWorkflowRule testWorkflowRule =
+      SDKTestWorkflowRule.newBuilder()
+          .setWorkflowTypes(TestNexus.class)
+          .setNexusServiceImplementation(new TestNexusServiceImpl())
+          .build();
+
+  @Test
+  public void testOperationHeaders() {
+    TestWorkflows.TestWorkflow1 workflowStub =
+        testWorkflowRule.newWorkflowStubTimeoutOptions(TestWorkflows.TestWorkflow1.class);
+    Assert.assertEquals(
+        "UnitTest:" + testWorkflowRule.getTaskQueue(),
+        workflowStub.execute(testWorkflowRule.getTaskQueue()));
+  }
+
+  public static class TestNexus implements TestWorkflows.TestWorkflow1 {
+    @Override
+    public String execute(String input) {
+      // Try to call with the typed stub
+      TestNexusServices.TestNexusService1 serviceStub =
+          Workflow.newNexusServiceStub(TestNexusServices.TestNexusService1.class);
+      return serviceStub.operation(input);
+    }
+  }
+
+  @ServiceImpl(service = TestNexusServices.TestNexusService1.class)
+  public static class TestNexusServiceImpl {
+    @OperationImpl
+    public OperationHandler<String, String> operation() {
+      return OperationHandler.sync(
+          (context, details, input) -> {
+            NexusInfo info = Nexus.getOperationContext().getInfo();
+            return info.getNamespace() + ":" + info.getTaskQueue();
+          });
+    }
+  }
+}