Remove Nexus sync client handler (#2403)

Author: Quinn-With-Two-Ns

temporal-opentracing/src/test/java/io/temporal/opentracing/NexusOperationTest.java

@@ -34,6 +34,7 @@
 import io.temporal.client.WorkflowClient;
 import io.temporal.client.WorkflowClientOptions;
 import io.temporal.client.WorkflowOptions;
+import io.temporal.nexus.Nexus;
 import io.temporal.nexus.WorkflowClientOperationHandlers;
 import io.temporal.testing.internal.SDKTestWorkflowRule;
 import io.temporal.worker.WorkerFactoryOptions;
@@ -83,10 +84,14 @@ public class TestNexusServiceImpl {
     @OperationImpl
     public OperationHandler<String, String> operation() {
       return WorkflowClientOperationHandlers.fromWorkflowMethod(
-          (context, details, client, input) ->
-              client.newWorkflowStub(
-                      TestOtherWorkflow.class,
-                      WorkflowOptions.newBuilder().setWorkflowId(details.getRequestId()).build())
+          (context, details, input) ->
+              Nexus.getOperationContext()
+                      .getWorkflowClient()
+                      .newWorkflowStub(
+                          TestOtherWorkflow.class,
+                          WorkflowOptions.newBuilder()
+                              .setWorkflowId(details.getRequestId())
+                              .build())
                   ::workflow);
     }
   }

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

@@ -28,7 +28,8 @@
  * Intercepts inbound calls to a Nexus operation on the worker side.
  *
  * <p>An instance should be created in {@link
- * WorkerInterceptor#interceptNexusOperation(NexusOperationInboundCallsInterceptor)}.
+ * WorkerInterceptor#interceptNexusOperation(OperationContext,
+ * NexusOperationInboundCallsInterceptor)}.
  *
  * <p>Prefer extending {@link NexusOperationInboundCallsInterceptorBase} and overriding only the
  * methods you need instead of implementing this interface directly. {@link

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

@@ -21,6 +21,7 @@
 package io.temporal.common.interceptors;
 
 import com.uber.m3.tally.Scope;
+import io.temporal.client.WorkflowClient;
 import io.temporal.common.Experimental;
 
 /**
@@ -41,4 +42,7 @@
 public interface NexusOperationOutboundCallsInterceptor {
   /** Intercepts call to get the metric scope in a Nexus operation. */
   Scope getMetricsScope();
+
+  /** Intercepts call to get the workflow client in a Nexus operation. */
+  WorkflowClient getWorkflowClient();
 }

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

@@ -21,6 +21,7 @@
 package io.temporal.common.interceptors;
 
 import com.uber.m3.tally.Scope;
+import io.temporal.client.WorkflowClient;
 import io.temporal.common.Experimental;
 
 /** Convenience base class for {@link NexusOperationOutboundCallsInterceptor} implementations. */
@@ -37,4 +38,9 @@ public NexusOperationOutboundCallsInterceptorBase(NexusOperationOutboundCallsInt
   public Scope getMetricsScope() {
     return next.getMetricsScope();
   }
+
+  @Override
+  public WorkflowClient getWorkflowClient() {
+    return next.getWorkflowClient();
+  }
 }

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

@@ -25,10 +25,10 @@
  * used directly.
  */
 public final class CurrentNexusOperationContext {
-  private static final ThreadLocal<NexusOperationContextImpl> CURRENT = new ThreadLocal<>();
+  private static final ThreadLocal<InternalNexusOperationContext> CURRENT = new ThreadLocal<>();
 
-  public static NexusOperationContextImpl get() {
-    NexusOperationContextImpl result = CURRENT.get();
+  public static InternalNexusOperationContext get() {
+    InternalNexusOperationContext result = CURRENT.get();
     if (result == null) {
       throw new IllegalStateException(
           "NexusOperationContext can be used only inside of nexus operation handler "
@@ -37,7 +37,7 @@ public static NexusOperationContextImpl get() {
     return CURRENT.get();
   }
 
-  public static void set(NexusOperationContextImpl context) {
+  public static void set(InternalNexusOperationContext context) {
     if (context == null) {
       throw new IllegalArgumentException("null context");
     }

…nal/nexus/NexusOperationContextImpl.java …nexus/InternalNexusOperationContext.javatemporal-sdk/src/main/java/io/temporal/internal/nexus/NexusOperationContextImpl.java renamed to temporal-sdk/src/main/java/io/temporal/internal/nexus/InternalNexusOperationContext.java temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusOperationContextImpl.java renamed to temporal-sdk/src/main/java/io/temporal/internal/nexus/InternalNexusOperationContext.java

@@ -25,26 +25,23 @@
 import io.temporal.common.interceptors.NexusOperationOutboundCallsInterceptor;
 import io.temporal.nexus.NexusOperationContext;
 
-public class NexusOperationContextImpl implements NexusOperationContext {
+public class InternalNexusOperationContext {
   private final String namespace;
   private final String taskQueue;
+  private final Scope metricScope;
   private final WorkflowClient client;
   NexusOperationOutboundCallsInterceptor outboundCalls;
 
-  public NexusOperationContextImpl(
-      String namespace,
-      String taskQueue,
-      WorkflowClient client,
-      NexusOperationOutboundCallsInterceptor outboundCalls) {
+  public InternalNexusOperationContext(
+      String namespace, String taskQueue, Scope metricScope, WorkflowClient client) {
     this.namespace = namespace;
     this.taskQueue = taskQueue;
+    this.metricScope = metricScope;
     this.client = client;
-    this.outboundCalls = outboundCalls;
   }
 
-  @Override
   public Scope getMetricsScope() {
-    return outboundCalls.getMetricsScope();
+    return metricScope;
   }
 
   public WorkflowClient getWorkflowClient() {
@@ -62,4 +59,23 @@ public String getNamespace() {
   public void setOutboundInterceptor(NexusOperationOutboundCallsInterceptor outboundCalls) {
     this.outboundCalls = outboundCalls;
   }
+
+  public NexusOperationContext getUserFacingContext() {
+    if (outboundCalls == null) {
+      throw new IllegalStateException("Outbound interceptor is not set");
+    }
+    return new NexusOperationContextImpl();
+  }
+
+  private class NexusOperationContextImpl implements NexusOperationContext {
+    @Override
+    public Scope getMetricsScope() {
+      return outboundCalls.getMetricsScope();
+    }
+
+    @Override
+    public WorkflowClient getWorkflowClient() {
+      return outboundCalls.getWorkflowClient();
+    }
+  }
 }

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

@@ -26,6 +26,6 @@ public final class NexusInternal {
   private NexusInternal() {}
 
   public static NexusOperationContext getOperationContext() {
-    return CurrentNexusOperationContext.get();
+    return CurrentNexusOperationContext.get().getUserFacingContext();
   }
 }

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

@@ -132,11 +132,7 @@ public Result handle(NexusTask task, Scope metricsScope) throws TimeoutException
       }
 
       CurrentNexusOperationContext.set(
-          new NexusOperationContextImpl(
-              namespace,
-              taskQueue,
-              client,
-              new RootNexusOperationOutboundCallsInterceptor(metricsScope)));
+          new InternalNexusOperationContext(namespace, taskQueue, metricsScope, client));
 
       switch (request.getVariantCase()) {
         case START_OPERATION:

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

@@ -21,18 +21,26 @@
 package io.temporal.internal.nexus;
 
 import com.uber.m3.tally.Scope;
+import io.temporal.client.WorkflowClient;
 import io.temporal.common.interceptors.NexusOperationOutboundCallsInterceptor;
 
 public class RootNexusOperationOutboundCallsInterceptor
     implements NexusOperationOutboundCallsInterceptor {
   private final Scope scope;
+  private final WorkflowClient client;
 
-  RootNexusOperationOutboundCallsInterceptor(Scope scope) {
+  RootNexusOperationOutboundCallsInterceptor(Scope scope, WorkflowClient client) {
     this.scope = scope;
+    this.client = client;
   }
 
   @Override
   public Scope getMetricsScope() {
     return scope;
   }
+
+  @Override
+  public WorkflowClient getWorkflowClient() {
+    return client;
+  }
 }

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

@@ -44,9 +44,10 @@ public OperationHandler<Object, Object> intercept(
           interceptor.interceptNexusOperation(context, inboundCallsInterceptor);
     }
 
+    InternalNexusOperationContext temporalNexusContext = CurrentNexusOperationContext.get();
     inboundCallsInterceptor.init(
         new RootNexusOperationOutboundCallsInterceptor(
-            CurrentNexusOperationContext.get().getMetricsScope()));
+            temporalNexusContext.getMetricsScope(), temporalNexusContext.getWorkflowClient()));
     return new OperationInterceptorConverter(inboundCallsInterceptor);
   }