Add interceptor to propagate headers to Nexus operations

README.md

@@ -149,6 +149,14 @@ See the README.md file in each main sample directory for cut/paste Gradle comman
 
 - [**AWS Encryption SDK**](/core/src/main/java/io/temporal/samples/keymanagementencryption/awsencryptionsdk): Demonstrates how to use the AWS Encryption SDK to encrypt and decrypt payloads with AWS KMS.
 
+#### Nexus Samples
+
+- [**Getting Started**](/core/src/main/java/io/temporal/samples/nexus): Demonstrates how to get started with Temporal and Nexus.
+
+- [**Cancellation**](/core/src/main/java/io/temporal/samples/nexuscancellation): Demonstrates how to cancel an async Nexus operation.
+
+- [**Context/Header Propagation**](/core/src/main/java/io/temporal/samples/nexuscontextpropagation): Demonstrates how to propagate context through Nexus operation headers.
+
 <!-- @@@SNIPEND -->
 
 ### Running SpringBoot Samples

core/src/main/java/io/temporal/samples/nexus/options/ClientOptions.java

@@ -34,7 +34,13 @@
 import org.apache.commons.cli.*;
 
 public class ClientOptions {
+
   public static WorkflowClient getWorkflowClient(String[] args) {
+    return getWorkflowClient(args, WorkflowClientOptions.newBuilder());
+  }
+
+  public static WorkflowClient getWorkflowClient(
+      String[] args, WorkflowClientOptions.Builder clientOptions) {
     Options options = new Options();
     Option targetHostOption = new Option("target-host", true, "Host:port for the Temporal service");
     targetHostOption.setRequired(false);
@@ -149,7 +155,6 @@ public static WorkflowClient getWorkflowClient(String[] args) {
 
     WorkflowServiceStubs service =
         WorkflowServiceStubs.newServiceStubs(serviceStubOptionsBuilder.build());
-    return WorkflowClient.newInstance(
-        service, WorkflowClientOptions.newBuilder().setNamespace(namespace).build());
+    return WorkflowClient.newInstance(service, clientOptions.setNamespace(namespace).build());
   }
 }

core/src/main/java/io/temporal/samples/nexuscontextpropagation/README.MD

@@ -0,0 +1,45 @@
+# Nexus Context Propagation
+
+This sample shows how to propagate MDC context values from Workflows to Nexus operations. 
+Nexus does not support `ContextPropagator` since the header format is not compatible. Users should look at `NexusMDCContextInterceptor` for propagating MDC context values.
+
+From more details on Nexus and how to setup to run this samples please see the [Nexus Sample](../nexus/README.MD).
+
+In separate terminal windows:
+
+### Nexus handler worker
+
+```
+./gradlew -q execute -PmainClass=io.temporal.samples.nexuscontextpropagation.handler.HandlerWorker \
+    --args="-target-host localhost:7233 -namespace my-target-namespace"
+```
+
+### Nexus caller worker
+
+```
+./gradlew -q execute -PmainClass=io.temporal.samples.nexuscontextpropagation.caller.CallerWorker \
+    --args="-target-host localhost:7233 -namespace my-caller-namespace"
+```
+
+### Start caller workflow
+
+```
+./gradlew -q execute -PmainClass=io.temporal.samples.nexuscontextpropagation.caller.CallerStarter \
+    --args="-target-host localhost:7233 -namespace my-caller-namespace"
+```
+
+### Output
+
+which should result in this on the caller side:
+```
+INFO  i.t.s.n.caller.CallerStarter - Started EchoCallerWorkflow workflowId: 7ac97cb9-b457-4052-af94-d82478c35c5e runId: 01954eb9-6963-7b52-9a1d-b74e64643846 
+INFO  i.t.s.n.caller.CallerStarter - Workflow result: Nexus Echo 👋 
+INFO  i.t.s.n.caller.CallerStarter - Started HelloCallerWorkflow workflowId: 9e0bc89c-5709-4742-b7c0-868464c2fccf runId: 01954eb9-6ae3-7d6d-b355-71545688309d 
+INFO  i.t.s.n.caller.CallerStarter - Workflow result: Hello Nexus 👋 
+```
+
+And this on the handler side:
+```
+INFO  i.t.s.n.handler.NexusServiceImpl - Echo called from a workflow with ID : 7ac97cb9-b457-4052-af94-d82478c35c5e 
+INFO  i.t.s.n.h.HelloHandlerWorkflowImpl - HelloHandlerWorkflow called from a workflow with ID : 9e0bc89c-5709-4742-b7c0-868464c2fccf 
+```

core/src/main/java/io/temporal/samples/nexuscontextpropagation/caller/CallerStarter.java

@@ -0,0 +1,64 @@
+/*
+ *  Copyright (c) 2020 Temporal Technologies, Inc. All Rights Reserved
+ *
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file is distributed on
+ *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *  express or implied. See the License for the specific language governing
+ *  permissions and limitations under the License.
+ */
+
+package io.temporal.samples.nexuscontextpropagation.caller;
+
+import io.temporal.api.common.v1.WorkflowExecution;
+import io.temporal.client.WorkflowClient;
+import io.temporal.client.WorkflowClientOptions;
+import io.temporal.client.WorkflowOptions;
+import io.temporal.samples.nexus.caller.EchoCallerWorkflow;
+import io.temporal.samples.nexus.caller.HelloCallerWorkflow;
+import io.temporal.samples.nexus.options.ClientOptions;
+import io.temporal.samples.nexus.service.NexusService;
+import io.temporal.samples.nexuscontextpropagation.propogation.MDCContextPropagator;
+import java.util.Collections;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CallerStarter {
+  private static final Logger logger = LoggerFactory.getLogger(CallerStarter.class);
+
+  public static void main(String[] args) {
+    WorkflowClient client =
+        ClientOptions.getWorkflowClient(
+            args,
+            WorkflowClientOptions.newBuilder()
+                .setContextPropagators(Collections.singletonList(new MDCContextPropagator())));
+
+    WorkflowOptions workflowOptions =
+        WorkflowOptions.newBuilder().setTaskQueue(CallerWorker.DEFAULT_TASK_QUEUE_NAME).build();
+    EchoCallerWorkflow echoWorkflow =
+        client.newWorkflowStub(EchoCallerWorkflow.class, workflowOptions);
+    WorkflowExecution execution = WorkflowClient.start(echoWorkflow::echo, "Nexus Echo 👋");
+    logger.info(
+        "Started EchoCallerWorkflow workflowId: {} runId: {}",
+        execution.getWorkflowId(),
+        execution.getRunId());
+    logger.info("Workflow result: {}", echoWorkflow.echo("Nexus Echo 👋"));
+    HelloCallerWorkflow helloWorkflow =
+        client.newWorkflowStub(HelloCallerWorkflow.class, workflowOptions);
+    execution = WorkflowClient.start(helloWorkflow::hello, "Nexus", NexusService.Language.EN);
+    logger.info(
+        "Started HelloCallerWorkflow workflowId: {} runId: {}",
+        execution.getWorkflowId(),
+        execution.getRunId());
+    logger.info("Workflow result: {}", helloWorkflow.hello("Nexus", NexusService.Language.ES));
+  }
+}

core/src/main/java/io/temporal/samples/nexuscontextpropagation/caller/CallerWorker.java

@@ -0,0 +1,58 @@
+/*
+ *  Copyright (c) 2020 Temporal Technologies, Inc. All Rights Reserved
+ *
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file is distributed on
+ *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *  express or implied. See the License for the specific language governing
+ *  permissions and limitations under the License.
+ */
+
+package io.temporal.samples.nexuscontextpropagation.caller;
+
+import io.temporal.client.WorkflowClient;
+import io.temporal.samples.nexus.options.ClientOptions;
+import io.temporal.samples.nexuscontextpropagation.propogation.NexusMDCContextInterceptor;
+import io.temporal.worker.Worker;
+import io.temporal.worker.WorkerFactory;
+import io.temporal.worker.WorkerFactoryOptions;
+import io.temporal.worker.WorkflowImplementationOptions;
+import io.temporal.workflow.NexusServiceOptions;
+import java.util.Collections;
+
+public class CallerWorker {
+  public static final String DEFAULT_TASK_QUEUE_NAME = "my-caller-workflow-task-queue";
+
+  public static void main(String[] args) {
+    WorkflowClient client = ClientOptions.getWorkflowClient(args);
+
+    WorkerFactory factory =
+        WorkerFactory.newInstance(
+            client,
+            WorkerFactoryOptions.newBuilder()
+                .setWorkerInterceptors(new NexusMDCContextInterceptor())
+                .build());
+
+    Worker worker = factory.newWorker(DEFAULT_TASK_QUEUE_NAME);
+    worker.registerWorkflowImplementationTypes(
+        WorkflowImplementationOptions.newBuilder()
+            .setNexusServiceOptions(
+                Collections.singletonMap(
+                    "NexusService",
+                    NexusServiceOptions.newBuilder().setEndpoint("my-nexus-endpoint-name").build()))
+            .build(),
+        EchoCallerWorkflowImpl.class,
+        HelloCallerWorkflowImpl.class);
+
+    factory.start();
+  }
+}

core/src/main/java/io/temporal/samples/nexuscontextpropagation/caller/EchoCallerWorkflowImpl.java

@@ -0,0 +1,46 @@
+/*
+ *  Copyright (c) 2020 Temporal Technologies, Inc. All Rights Reserved
+ *
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file is distributed on
+ *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *  express or implied. See the License for the specific language governing
+ *  permissions and limitations under the License.
+ */
+
+package io.temporal.samples.nexuscontextpropagation.caller;
+
+import io.temporal.samples.nexus.caller.EchoCallerWorkflow;
+import io.temporal.samples.nexus.service.NexusService;
+import io.temporal.workflow.NexusOperationOptions;
+import io.temporal.workflow.NexusServiceOptions;
+import io.temporal.workflow.Workflow;
+import java.time.Duration;
+import org.slf4j.MDC;
+
+public class EchoCallerWorkflowImpl implements EchoCallerWorkflow {
+  NexusService nexusService =
+      Workflow.newNexusServiceStub(
+          NexusService.class,
+          NexusServiceOptions.newBuilder()
+              .setOperationOptions(
+                  NexusOperationOptions.newBuilder()
+                      .setScheduleToCloseTimeout(Duration.ofSeconds(10))
+                      .build())
+              .build());
+
+  @Override
+  public String echo(String message) {
+    MDC.put("x-nexus-caller-workflow-id", Workflow.getInfo().getWorkflowId());
+    return nexusService.echo(new NexusService.EchoInput(message)).getMessage();
+  }
+}

core/src/main/java/io/temporal/samples/nexuscontextpropagation/caller/HelloCallerWorkflowImpl.java

@@ -0,0 +1,53 @@
+/*
+ *  Copyright (c) 2020 Temporal Technologies, Inc. All Rights Reserved
+ *
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file is distributed on
+ *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *  express or implied. See the License for the specific language governing
+ *  permissions and limitations under the License.
+ */
+
+package io.temporal.samples.nexuscontextpropagation.caller;
+
+import io.temporal.samples.nexus.caller.HelloCallerWorkflow;
+import io.temporal.samples.nexus.service.NexusService;
+import io.temporal.workflow.NexusOperationHandle;
+import io.temporal.workflow.NexusOperationOptions;
+import io.temporal.workflow.NexusServiceOptions;
+import io.temporal.workflow.Workflow;
+import java.time.Duration;
+import org.slf4j.MDC;
+
+public class HelloCallerWorkflowImpl implements HelloCallerWorkflow {
+  NexusService nexusService =
+      Workflow.newNexusServiceStub(
+          NexusService.class,
+          NexusServiceOptions.newBuilder()
+              .setOperationOptions(
+                  NexusOperationOptions.newBuilder()
+                      .setScheduleToCloseTimeout(Duration.ofSeconds(10))
+                      .build())
+              .build());
+
+  @Override
+  public String hello(String message, NexusService.Language language) {
+    MDC.put("x-nexus-caller-workflow-id", Workflow.getInfo().getWorkflowId());
+    NexusOperationHandle<NexusService.HelloOutput> handle =
+        Workflow.startNexusOperation(
+            nexusService::hello, new NexusService.HelloInput(message, language));
+    // Optionally wait for the operation to be started. NexusOperationExecution will contain the
+    // operation token in case this operation is asynchronous.
+    handle.getExecution().get();
+    return handle.getResult().get().getMessage();
+  }
+}

core/src/main/java/io/temporal/samples/nexuscontextpropagation/handler/HandlerWorker.java

@@ -0,0 +1,55 @@
+/*
+ *  Copyright (c) 2020 Temporal Technologies, Inc. All Rights Reserved
+ *
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file is distributed on
+ *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *  express or implied. See the License for the specific language governing
+ *  permissions and limitations under the License.
+ */
+
+package io.temporal.samples.nexuscontextpropagation.handler;
+
+import io.temporal.client.WorkflowClient;
+import io.temporal.client.WorkflowClientOptions;
+import io.temporal.samples.nexus.options.ClientOptions;
+import io.temporal.samples.nexuscontextpropagation.propogation.MDCContextPropagator;
+import io.temporal.samples.nexuscontextpropagation.propogation.NexusMDCContextInterceptor;
+import io.temporal.worker.Worker;
+import io.temporal.worker.WorkerFactory;
+import io.temporal.worker.WorkerFactoryOptions;
+import java.util.Collections;
+
+public class HandlerWorker {
+  public static final String DEFAULT_TASK_QUEUE_NAME = "my-handler-task-queue";
+
+  public static void main(String[] args) {
+    WorkflowClient client =
+        ClientOptions.getWorkflowClient(
+            args,
+            WorkflowClientOptions.newBuilder()
+                .setContextPropagators(Collections.singletonList(new MDCContextPropagator())));
+
+    WorkerFactory factory =
+        WorkerFactory.newInstance(
+            client,
+            WorkerFactoryOptions.newBuilder()
+                .setWorkerInterceptors(new NexusMDCContextInterceptor())
+                .build());
+
+    Worker worker = factory.newWorker(DEFAULT_TASK_QUEUE_NAME);
+    worker.registerWorkflowImplementationTypes(HelloHandlerWorkflowImpl.class);
+    worker.registerNexusServiceImplementation(new NexusServiceImpl());
+
+    factory.start();
+  }
+}