Merge pull request #1039 from dapr/workflow

Reworked javascript workflow examples to use a webserver

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"]

workflows/javascript/sdk/order-processor/workflowApp.ts → 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);
+});