add middleware implementation (#87)

* add middleware implementation

* update code with new API

* rename middleware chain

* minor updates

* refactor code

* update method/variable names make them more clear

* update ctx to taskOrchestrationContext make it more clear

* Update build-validation.yml

* Update build.gradle

* Update build.gradle

* update to latest worker 2.7.0

* update OrchestratorBlockedEvent to OrchestratorBlockedException

Co-authored-by: Shreyas Gopalakrishna <[email protected]>

Author: kaibocai

.github/workflows/build-validation.yml

@@ -83,7 +83,7 @@ jobs:
         continue-on-error: true
       - name: Download azure functions java library # TODO: Remove this step once gradle plugin is updated
         run: |
-          wget -P samples-azure-functions/build/azure-functions/azure-functions-sample/lib/ "https://repo.maven.apache.org/maven2/com/microsoft/azure/functions/azure-functions-java-library/2.0.1/azure-functions-java-library-2.0.1.jar" --show-progress
+          wget -P samples-azure-functions/build/azure-functions/azure-functions-sample/lib/ "https://repo.maven.apache.org/maven2/com/microsoft/azure/functions/azure-functions-java-library/2.1.0/azure-functions-java-library-2.1.0.jar" --show-progress
       - name: Run azure functions test
         run: samples-azure-functions/e2e-test.ps1 -DockerfilePath samples-azure-functions/Dockerfile -HttpStartPath api/StartOrchestration
-        shell: pwsh
+        shell: pwsh

azurefunctions/build.gradle

@@ -17,7 +17,7 @@ repositories {
 
 dependencies {
     api project(':client')
-    implementation group: 'com.microsoft.azure.functions', name: 'azure-functions-java-library', version: '2.0.1'
+    implementation group: 'com.microsoft.azure.functions', name: 'azure-functions-java-library', version: '2.1.0'
     implementation "com.google.protobuf:protobuf-java:${protocVersion}"
 }
 
@@ -59,4 +59,4 @@ publishing {
 java {
     withSourcesJar()
     withJavadocJar()
-}
+}

azurefunctions/src/main/java/com/microsoft/durabletask/azurefunctions/internal/middleware/OrchestrationMiddleware.java

@@ -0,0 +1,50 @@
+/**
+ * Copyright (c) Microsoft Corporation. All rights reserved.
+ * Licensed under the MIT License. See License.txt in the project root for
+ * license information.
+ */
+
+package com.microsoft.durabletask.azurefunctions.internal.middleware;
+
+import com.microsoft.azure.functions.internal.spi.middleware.Middleware;
+import com.microsoft.azure.functions.internal.spi.middleware.MiddlewareChain;
+import com.microsoft.azure.functions.internal.spi.middleware.MiddlewareContext;
+import com.microsoft.durabletask.OrchestrationRunner;
+import com.microsoft.durabletask.OrchestratorBlockedException;
+
+/**
+ * Durable Function Orchestration Middleware
+ *
+ * <p>This class is internal and is hence not for public use. Its APIs are unstable and can change
+ * at any time.
+ */
+public class OrchestrationMiddleware implements Middleware {
+
+    private static final String ORCHESTRATION_TRIGGER = "DurableOrchestrationTrigger";
+
+    @Override
+    public void invoke(MiddlewareContext context, MiddlewareChain chain) throws Exception {
+        String parameterName = context.getParameterName(ORCHESTRATION_TRIGGER);
+        if (parameterName == null){
+            chain.doNext(context);
+            return;
+        }
+        String orchestratorRequestEncodedProtoBytes = (String) context.getParameterValue(parameterName);
+        String orchestratorOutputEncodedProtoBytes = OrchestrationRunner.loadAndRun(orchestratorRequestEncodedProtoBytes, taskOrchestrationContext -> {
+            try {
+                context.updateParameterValue(parameterName, taskOrchestrationContext);
+                chain.doNext(context);
+                return context.getReturnValue();
+            } catch (Exception e) {
+                // The OrchestratorBlockedEvent will be wrapped into InvocationTargetException by using reflection to
+                // invoke method. Thus get the cause to check if it's OrchestratorBlockedEvent.
+                Throwable cause = e.getCause();
+                if (cause instanceof OrchestratorBlockedException){
+                    throw (OrchestratorBlockedException) cause;
+                }
+                throw new RuntimeException("Unexpected failure in the task execution", e);
+            }
+        });
+        context.updateReturnValue(orchestratorOutputEncodedProtoBytes);
+    }
+}

azurefunctions/src/main/resources/META-INF/services/com.microsoft.azure.functions.internal.spi.middleware.Middleware

@@ -0,0 +1 @@
+com.microsoft.durabletask.azurefunctions.internal.middleware.OrchestrationMiddleware

samples-azure-functions/Dockerfile

@@ -1,7 +1,7 @@
 FROM mcr.microsoft.com/azure-functions/java:4-java11
 
 RUN rm /azure-functions-host/workers/java/azure-functions-java-worker.jar
-COPY samples-azure-functions/azure-functions-java-worker-2.2.3.jar /azure-functions-host/workers/java/azure-functions-java-worker.jar
+COPY samples-azure-functions/azure-functions-java-worker-2.7.0.jar /azure-functions-host/workers/java/azure-functions-java-worker.jar
 COPY samples-azure-functions/build/azure-functions/azure-functions-sample/ /home/site/wwwroot/
 ENV AzureWebJobsScriptRoot=/home/site/wwwroot \
     AzureFunctionsJobHost__Logging__Console__IsEnabled=true

samples-azure-functions/azure-functions-java-worker-2.2.3.jar renamed to samples-azure-functions/azure-functions-java-worker-2.7.0.jar

@@ -19,7 +19,7 @@ dependencies {
     implementation project(':client')
     implementation project(':azurefunctions')
 
-    implementation 'com.microsoft.azure.functions:azure-functions-java-library:2.0.1'
+    implementation 'com.microsoft.azure.functions:azure-functions-java-library:2.1.0'
     testImplementation 'org.junit.jupiter:junit-jupiter:5.6.2'
     testImplementation 'org.mockito:mockito-core:3.3.3'
     runtimeOnly "io.grpc:grpc-netty-shaded:1.38.0"

samples-azure-functions/build.gradle

@@ -4,8 +4,7 @@
 import com.microsoft.azure.functions.*;
 import java.util.*;
 
-import com.microsoft.durabletask.DurableTaskClient;
-import com.microsoft.durabletask.OrchestrationRunner;
+import com.microsoft.durabletask.*;
 import com.microsoft.durabletask.azurefunctions.DurableActivityTrigger;
 import com.microsoft.durabletask.azurefunctions.DurableClientContext;
 import com.microsoft.durabletask.azurefunctions.DurableClientInput;
@@ -37,15 +36,13 @@ public HttpResponseMessage startOrchestration(
      */
     @FunctionName("Cities")
     public String citiesOrchestrator(
-            @DurableOrchestrationTrigger(name = "orchestratorRequestProtoBytes") String orchestratorRequestProtoBytes) {
-        return OrchestrationRunner.loadAndRun(orchestratorRequestProtoBytes, ctx -> {
-            String result = "";
-            result += ctx.callActivity("Capitalize", "Tokyo", String.class).await() + ", ";
-            result += ctx.callActivity("Capitalize", "London", String.class).await() + ", ";
-            result += ctx.callActivity("Capitalize", "Seattle", String.class).await() + ", ";
-            result += ctx.callActivity("Capitalize", "Austin", String.class).await();
-            return result;
-        });
+            @DurableOrchestrationTrigger(name = "taskOrchestrationContext") TaskOrchestrationContext ctx) {
+        String result = "";
+        result += ctx.callActivity("Capitalize", "Tokyo", String.class).await() + ", ";
+        result += ctx.callActivity("Capitalize", "London", String.class).await() + ", ";
+        result += ctx.callActivity("Capitalize", "Seattle", String.class).await() + ", ";
+        result += ctx.callActivity("Capitalize", "Austin", String.class).await();
+        return result;
     }
 
     /**