Merge pull request #1039 from dapr/workflow

Reworked javascript workflow examples to use a webserver

Author: paulyuk

workflows/javascript/sdk/README.md

@@ -2,9 +2,11 @@
 
 In this quickstart, you'll create a simple console application to demonstrate Dapr's workflow programming model and the workflow management API. The console app starts and manages the lifecycle of a workflow that stores and retrieves data in a state store.
 
-This quickstart includes one project:
+This quickstart includes 3 entry points demonstrating different ways to host a workflow:
 
-- JavaScript console app `order-processor` 
+1. JavaScript console app `order-processor` 
+2. Express app `order-process-with-express-server`
+3. Express app via Dapr Server `order-process-with-dapr-server`
 
 The quickstart contains 1 workflow to simulate purchasing items from a store, and 5 unique activities within the workflow. These 5 activities are as follows:
 
@@ -16,7 +18,7 @@ The quickstart contains 1 workflow to simulate purchasing items from a store, an
 
 ### Run the order processor workflow with multi-app-run
 
-1. Open a new terminal window and navigate to `order-processor` directory: 
+1. Open a new terminal window and install the dependencies: 
 
 <!-- STEP
 name: build order-process app
@@ -29,12 +31,13 @@ npm run build
 ```
 
 <!-- END_STEP -->
-2. Run the console app with Dapr: 
+2. Run the app 
 
+- Entry point 1 : JavaScript console app
 <!-- STEP
 name: Run order-processor service
 expected_stdout_lines:
-  - '== APP - workflowApp == == APP == Payment of 100 for 10 item1 processed successfully'
+  - '== APP - workflowApp == Payment of 100 for 10 item1 processed successfully'
   - 'there are now 90 item1 in stock'
   - 'processed successfully!'
 expected_stderr_lines:
@@ -43,13 +46,33 @@ background: true
 sleep: 15
 timeout_seconds: 120
 -->
-    
+
 ```bash
 dapr run -f .
 ```
-
 <!-- END_STEP -->
 
+- Entry point 2 : Express app
+
+```bash
+dapr run -f dapr-AppWithExpressServer.yaml
+```
+
+```bash
+curl --request POST \
+  --url http://localhost:3500/v1.0/invoke/workflowApp/method/start-workflow
+```
+
+- Entry point 3 : Express app via Dapr Server
+
+```bash
+dapr run -f dapr-AppWithDaprServer.yaml
+```
+```bash
+curl --request POST \
+  --url http://localhost:3500/v1.0/invoke/workflowApp/method/start-workflow
+```
+
 3. Expected output
 
 

workflows/javascript/sdk/dapr-AppWithDaprServer.yaml

@@ -0,0 +1,10 @@
+version: 1
+common:
+  resourcesPath: ../../components
+apps:
+  - appID: workflowApp
+    appDirPath: ./order-processor/
+    appPort: 3000
+    daprHTTPPort: 3500
+    daprGRPCPort: 50001
+    command: ["npm", "run", "start:order-process-with-dapr-server"]

workflows/javascript/sdk/dapr-AppWithExpressServer.yaml

@@ -0,0 +1,10 @@
+version: 1
+common:
+  resourcesPath: ../../components
+apps:
+  - appID: workflowApp
+    appDirPath: ./order-processor/
+    appPort: 3000
+    daprHTTPPort: 3500
+    daprGRPCPort: 50001
+    command: ["npm", "run", "start:order-process-with-express-server"]

workflows/javascript/sdk/dapr.yaml

@@ -4,4 +4,4 @@ common:
 apps:
   - appID: workflowApp
     appDirPath: ./order-processor/
-    command: ["npm", "run", "start:dapr:order-process"]
+    command: ["npm", "run", "start:order-process"]

…cript/sdk/order-processor/workflowApp.ts …ws/javascript/sdk/order-processor/app.tsworkflows/javascript/sdk/order-processor/workflowApp.ts renamed to workflows/javascript/sdk/order-processor/app.ts workflows/javascript/sdk/order-processor/workflowApp.ts renamed to workflows/javascript/sdk/order-processor/app.ts

@@ -1,13 +1,23 @@
-import { DaprWorkflowClient, WorkflowRuntime, DaprClient } from "@dapr/dapr";
+import { DaprWorkflowClient, WorkflowRuntime, DaprClient, CommunicationProtocolEnum } from "@dapr/dapr";
 import { InventoryItem, OrderPayload } from "./model";
 import { notifyActivity, orderProcessingWorkflow, processPaymentActivity, requestApprovalActivity, reserveInventoryActivity, updateInventoryActivity } from "./orderProcessingWorkflow";
 
+const workflowWorker = new WorkflowRuntime();
+
 async function start() {
   // Update the gRPC client and worker to use a local address and port
   const workflowClient = new DaprWorkflowClient();
-  const workflowWorker = new WorkflowRuntime();
 
-  const daprClient = new DaprClient();
+
+  const daprHost = process.env.DAPR_HOST ?? "127.0.0.1";
+  const daprPort = process.env.DAPR_GRPC_PORT ?? "50001";
+
+  const daprClient = new DaprClient({
+    daprHost,
+    daprPort,
+    communicationProtocol: CommunicationProtocolEnum.GRPC,
+  });
+
   const storeName = "statestore";
 
   const inventory = new InventoryItem("item1", 100, 100);
@@ -52,10 +62,13 @@ async function start() {
     throw error;
   }
 
-  await workflowWorker.stop();
   await workflowClient.stop();
 }
 
+process.on('SIGTERM', () => {
+  workflowWorker.stop();
+})
+
 start().catch((e) => {
   console.error(e);
   process.exit(1);

workflows/javascript/sdk/order-processor/appWithDaprServer.ts

@@ -0,0 +1,87 @@
+import { DaprWorkflowClient, WorkflowRuntime, DaprClient } from "@dapr/dapr";
+import { InventoryItem, OrderPayload } from "./model";
+import { notifyActivity, orderProcessingWorkflow, processPaymentActivity, requestApprovalActivity, reserveInventoryActivity, updateInventoryActivity } from "./orderProcessingWorkflow";
+import { DaprServer, CommunicationProtocolEnum } from "@dapr/dapr";
+import express from "express";
+
+const daprHost = process.env.DAPR_HOST ?? "localhost"; // Dapr Sidecar Host
+const daprPort = process.env.DAPR_HTTP_PORT || "3500"; // Dapr Sidecar Port of this Example Server
+
+const app = express();
+
+const daprServer = new DaprServer({
+  serverHost:  "127.0.0.1", // App Host
+  serverPort: process.env.APP_PORT || "3000", // App Port
+  serverHttp: app,
+  communicationProtocol: CommunicationProtocolEnum.HTTP, // Add this line
+  clientOptions: {
+    daprHost,
+    daprPort
+  }
+});
+
+const daprClient = new DaprClient();
+const workflowClient = new DaprWorkflowClient();
+const workflowWorker = new WorkflowRuntime();
+
+app.post("/start-workflow", async (req, res) => {
+
+  const storeName = "statestore";
+  const inventory = new InventoryItem("item1", 100, 100);
+  const key = inventory.itemName;
+
+  await daprClient.state.delete(storeName, key);
+  await daprClient.state.save(storeName, [
+    {
+      key: key,
+      value: inventory,
+    }
+  ]);
+
+  const order = new OrderPayload("item1", 100, 10);
+
+  // Schedule a new orchestration
+  try {
+    const id = await workflowClient.scheduleNewWorkflow(orderProcessingWorkflow, order);
+    console.log(`Orchestration scheduled with ID: ${id}`);
+
+    // Wait for orchestration completion
+    const state = await workflowClient.waitForWorkflowCompletion(id, undefined, 30);
+
+    var orchestrationResult = `Orchestration completed! Result: ${state?.serializedOutput}`;
+    console.log(orchestrationResult);
+  } catch (error) {
+    console.error("Error scheduling or waiting for orchestration:", error);
+    throw error;
+  }
+
+  res.send(orchestrationResult);
+});
+
+async function start() {
+  workflowWorker
+    .registerWorkflow(orderProcessingWorkflow)
+    .registerActivity(notifyActivity)
+    .registerActivity(reserveInventoryActivity)
+    .registerActivity(requestApprovalActivity)
+    .registerActivity(processPaymentActivity)
+    .registerActivity(updateInventoryActivity);
+
+  // Wrap the worker startup in a try-catch block to handle any errors during startup
+  try {
+    await workflowWorker.start();
+    console.log("Workflow runtime started successfully");
+  } catch (error) {
+    console.error("Error starting workflow runtime:", error);
+  }
+
+  // Initialize subscriptions before the server starts, the Dapr sidecar uses it.
+  // This will also initialize the app server itself (removing the need for `app.listen` to be called).
+  await daprServer.start();
+};
+
+start().catch((e) => {
+  workflowWorker.stop();
+  console.error(e);
+  process.exit(1);
+});

workflows/javascript/sdk/order-processor/appWithExpressServer.ts

@@ -0,0 +1,76 @@
+import { DaprWorkflowClient, WorkflowRuntime, DaprClient } from "@dapr/dapr";
+import { InventoryItem, OrderPayload } from "./model";
+import { notifyActivity, orderProcessingWorkflow, processPaymentActivity, requestApprovalActivity, reserveInventoryActivity, updateInventoryActivity } from "./orderProcessingWorkflow";
+import express from "express";
+
+const app = express();
+
+const daprClient = new DaprClient();
+const workflowClient = new DaprWorkflowClient();
+const workflowWorker = new WorkflowRuntime();
+
+app.post("/start-workflow", async (req, res) => {
+
+  const storeName = "statestore";
+  const inventory = new InventoryItem("item1", 100, 100);
+  const key = inventory.itemName;
+
+  await daprClient.state.save(storeName, [
+    {
+      key: key,
+      value: inventory,
+    }
+  ]);
+
+  const order = new OrderPayload("item1", 100, 10);
+
+  // Schedule a new orchestration
+  try {
+    const id = await workflowClient.scheduleNewWorkflow(orderProcessingWorkflow, order);
+    console.log(`Orchestration scheduled with ID: ${id}`);
+
+    // Wait for orchestration completion
+    const state = await workflowClient.waitForWorkflowCompletion(id, undefined, 30);
+
+    var orchestrationResult = `Orchestration completed! Result: ${state?.serializedOutput}`;
+    console.log(orchestrationResult);
+  } catch (error) {
+    console.error("Error scheduling or waiting for orchestration:", error);
+    throw error;
+  }
+
+  res.send(orchestrationResult);
+});
+
+async function start() {
+  workflowWorker
+    .registerWorkflow(orderProcessingWorkflow)
+    .registerActivity(notifyActivity)
+    .registerActivity(reserveInventoryActivity)
+    .registerActivity(requestApprovalActivity)
+    .registerActivity(processPaymentActivity)
+    .registerActivity(updateInventoryActivity);
+
+  // Wrap the worker startup in a try-catch block to handle any errors during startup
+  try {
+    await workflowWorker.start();
+    console.log("Workflow runtime started successfully");
+  } catch (error) {
+    console.error("Error starting workflow runtime:", error);
+  }
+};
+
+const server = app.listen(process.env.APP_PORT || 3000, () => {
+  console.log(`Example app listening on port APP_PORT or 3000`);
+})
+
+process.on('SIGTERM', () => {
+  workflowWorker.stop();
+  server.close();
+})
+
+start().catch((e) => {
+  workflowWorker.stop();
+  console.error(e);
+  process.exit(1);
+});