// 不要这样做！
Instant currentTime = Instant.now();
UUID newIdentifier = UUID.randomUUID();
String randomString = getRandomString();

// 不要这样做！
const currentTime = new Date();
const newIdentifier = uuidv4();
const randomString = getRandomString();

// 不要这样做！
const currentTime = time.Now()

// 这样做！！
Instant currentTime = context.getCurrentInstant();
Guid newIdentifier = context.newGuid();
String randomString = context.callActivity(GetRandomString.class.getName(), String.class).await();

// 这样做！！
const currentTime = context.getCurrentUtcDateTime();
const randomString = yield context.callActivity(getRandomString);

const currentTime = ctx.CurrentUTCDateTime()

// 不要这样做！
String configuration = System.getenv("MY_CONFIGURATION");

HttpRequest request = HttpRequest.newBuilder().uri(new URI("https://postman-echo.com/post")).GET().build();
HttpResponse<String> response = HttpClient.newBuilder().build().send(request, HttpResponse.BodyHandlers.ofString());

// 不要这样做！
// 访问环境变量（Node.js）
const configuration = process.env.MY_CONFIGURATION;

fetch('https://postman-echo.com/get')
  .then(response => response.text())
  .then(data => {
    console.log(data);
  })
  .catch(error => {
    console.error('Error:', error);
  });

// 不要这样做！
resp, err := http.Get("http://example.com/api/data")

// 这样做！！
String configuration = ctx.getInput(InputType.class).getConfiguration(); // 假想的工作流输入参数
String data = ctx.callActivity(MakeHttpCall.class, "https://example.com/api/data", String.class).await();

// 这样做！！
const configuration = workflowInput.getConfiguration(); // 假想的工作流输入参数
const data = yield ctx.callActivity(makeHttpCall, "https://example.com/api/data");

// 这样做！！
err := ctx.CallActivity(MakeHttpCallActivity, workflow.ActivityInput("https://example.com/api/data")).Await(&output)

// 不要这样做！
new Thread(() -> {
    ctx.callActivity(DoSomethingActivity.class.getName()).await();
}).start();
ctx.createTimer(Duration.ofSeconds(5)).await();

// 不要这样做！
go func() {
  err := ctx.CallActivity(DoSomething).Await(nil)
}()
err := ctx.CreateTimer(time.Second).Await(nil)

// 这样做！！
ctx.callActivity(DoSomethingActivity.class.getName()).await();
ctx.createTimer(Duration.ofSeconds(5)).await();

// 这样做！
task := ctx.CallActivity(DoSomething)
task.Await(nil)

import { DaprWorkflowClient, WorkflowActivityContext, WorkflowContext, WorkflowRuntime, TWorkflow } from "@dapr/dapr";

async function start() {
  // 更新 gRPC 客户端和工作者以使用本地地址和端口
  const daprHost = "localhost";
  const daprPort = "50001";
  const workflowClient = new DaprWorkflowClient({
    daprHost,
    daprPort,
  });
  const workflowRuntime = new WorkflowRuntime({
    daprHost,
    daprPort,
  });

  const hello = async (_: WorkflowActivityContext, name: string) => {
    return `Hello ${name}!`;
  };

  const sequence: TWorkflow = async function* (ctx: WorkflowContext): any {
    const cities: string[] = [];

    const result1 = yield ctx.callActivity(hello, "Tokyo");
    cities.push(result1);
    const result2 = yield ctx.callActivity(hello, "Seattle");
    cities.push(result2);
    const result3 = yield ctx.callActivity(hello, "London");
    cities.push(result3);

    return cities;
  };

  workflowRuntime.registerWorkflow(sequence).registerActivity(hello);

  // 将工作者启动包装在 try-catch 块中以处理启动期间的任何错误
  try {
    await workflowRuntime.start();
    console.log("工作流运行时启动成功");
  } catch (error) {
    console.error("启动工作流运行时时出错:", error);
  }

  // 调度新的编排
  try {
    const id = await workflowClient.scheduleNewWorkflow(sequence);
    console.log(`编排已调度，ID: ${id}`);

    // 等待编排完成
    const state = await workflowClient.waitForWorkflowCompletion(id, undefined, 30);

    console.log(`编排完成！结果: ${state?.serializedOutput}`);
  } catch (error) {
    console.error("调度或等待编排时出错:", error);
  }

  await workflowRuntime.stop();
  await workflowClient.stop();

  // 停止 dapr sidecar
  process.exit(0);
}

start().catch((e) => {
  console.error(e);
  process.exit(1);
});

// 支持长时间中断的指数退避重试策略
var retryOptions = new WorkflowTaskOptions
{
    RetryPolicy = new WorkflowRetryPolicy(
        firstRetryInterval: TimeSpan.FromMinutes(1),
        backoffCoefficient: 2.0,
        maxRetryInterval: TimeSpan.FromHours(1),
        maxNumberOfAttempts: 10),
};

try
{
    var result1 = await context.CallActivityAsync<string>("Step1", wfInput, retryOptions);
    var result2 = await context.CallActivityAsync<byte[]>("Step2", result1, retryOptions);
    var result3 = await context.CallActivityAsync<long[]>("Step3", result2, retryOptions);
    return string.Join(", ", result4);
}
catch (TaskFailedException) // 任务失败会作为 TaskFailedException 显示
{
    // 重试过期 - 应用自定义补偿逻辑
    await context.CallActivityAsync<long[]>("MyCompensation", options: retryOptions);
    throw;
}

public class ChainWorkflow extends Workflow {
    @Override
    public WorkflowStub create() {
        return ctx -> {
            StringBuilder sb = new StringBuilder();
            String wfInput = ctx.getInput(String.class);
            String result1 = ctx.callActivity("Step1", wfInput, String.class).await();
            String result2 = ctx.callActivity("Step2", result1, String.class).await();
            String result3 = ctx.callActivity("Step3", result2, String.class).await();
            String result = sb.append(result1).append(',').append(result2).append(',').append(result3).toString();
            ctx.complete(result);
        };
    }
}

    class Step1 implements WorkflowActivity {

        @Override
        public Object run(WorkflowActivityContext ctx) {
            Logger logger = LoggerFactory.getLogger(Step1.class);
            logger.info("Starting Activity: " + ctx.getName());
            // Do some work
            return null;
        }
    }

    class Step2 implements WorkflowActivity {

        @Override
        public Object run(WorkflowActivityContext ctx) {
            Logger logger = LoggerFactory.getLogger(Step2.class);
            logger.info("Starting Activity: " + ctx.getName());
            // Do some work
            return null;
        }
    }

    class Step3 implements WorkflowActivity {

        @Override
        public Object run(WorkflowActivityContext ctx) {
            Logger logger = LoggerFactory.getLogger(Step3.class);
            logger.info("Starting Activity: " + ctx.getName());
            // Do some work
            return null;
        }
    }

func TaskChainWorkflow(ctx *workflow.WorkflowContext) (any, error) {
	var input int
	if err := ctx.GetInput(&input); err != nil {
		return "", err
	}
	var result1 int
	if err := ctx.CallActivity(Step1, workflow.ActivityInput(input)).Await(&result1); err != nil {
		return nil, err
	}
	var result2 int
	if err := ctx.CallActivity(Step2, workflow.ActivityInput(input)).Await(&result2); err != nil {
		return nil, err
	}
	var result3 int
	if err := ctx.CallActivity(Step3, workflow.ActivityInput(input)).Await(&result3); err != nil {
		return nil, err
	}
	return []int{result1, result2, result3}, nil
}
func Step1(ctx workflow.ActivityContext) (any, error) {
	var input int
	if err := ctx.GetInput(&input); err != nil {
		return "", err
	}
	fmt.Printf("步骤 1: 接收到输入: %s", input)
	return input + 1, nil
}
func Step2(ctx workflow.ActivityContext) (any, error) {
	var input int
	if err := ctx.GetInput(&input); err != nil {
		return "", err
	}
	fmt.Printf("步骤 2: 接收到输入: %s", input)
	return input * 2, nil
}
func Step3(ctx workflow.ActivityContext) (any, error) {
	var input int
	if err := ctx.GetInput(&input); err != nil {
		return "", err
	}
	fmt.Printf("步骤 3: 接收到输入: %s", input)
	return int(math.Pow(float64(input), 2)), nil
}

import {
  Task,
  DaprWorkflowClient,
  WorkflowActivityContext,
  WorkflowContext,
  WorkflowRuntime,
  TWorkflow,
} from "@dapr/dapr";

// 将整个代码包装在一个立即调用的异步函数中
async function start() {
  // 更新 gRPC 客户端和工作者以使用本地地址和端口
  const daprHost = "localhost";
  const daprPort = "50001";
  const workflowClient = new DaprWorkflowClient({
    daprHost,
    daprPort,
  });
  const workflowRuntime = new WorkflowRuntime({
    daprHost,
    daprPort,
  });

  function getRandomInt(min: number, max: number): number {
    return Math.floor(Math.random() * (max - min + 1)) + min;
  }

  async function getWorkItemsActivity(_: WorkflowActivityContext): Promise<string[]> {
    const count: number = getRandomInt(2, 10);
    console.log(`生成 ${count} 个工作项...`);

    const workItems: string[] = Array.from({ length: count }, (_, i) => `工作项 ${i}`);
    return workItems;
  }

  function sleep(ms: number): Promise<void> {
    return new Promise((resolve) => setTimeout(resolve, ms));
  }

  async function processWorkItemActivity(context: WorkflowActivityContext, item: string): Promise<number> {
    console.log(`处理工作项: ${item}`);

    // 模拟一些需要可变时间的工作
    const sleepTime = Math.random() * 5000;
    await sleep(sleepTime);

    // 返回给定工作项的结果，在这种情况下也是一个随机数
    // 有关工作流中随机数的更多信息，请查看
    // https://learn.microsoft.com/azure/azure-functions/durable/durable-functions-code-constraints?tabs=csharp#random-numbers
    return Math.floor(Math.random() * 11);
  }

  const workflow: TWorkflow = async function* (ctx: WorkflowContext): any {
    const tasks: Task<any>[] = [];
    const workItems = yield ctx.callActivity(getWorkItemsActivity);
    for (const workItem of workItems) {
      tasks.push(ctx.callActivity(processWorkItemActivity, workItem));
    }
    const results: number[] = yield ctx.whenAll(tasks);
    const sum: number = results.reduce((accumulator, currentValue) => accumulator + currentValue, 0);
    return sum;
  };

  workflowRuntime.registerWorkflow(workflow);
  workflowRuntime.registerActivity(getWorkItemsActivity);
  workflowRuntime.registerActivity(processWorkItemActivity);

  // 将工作者启动包装在 try-catch 块中以处理启动期间的任何错误
  try {
    await workflowRuntime.start();
    console.log("工作者启动成功");
  } catch (error) {
    console.error("启动工作者时出错:", error);
  }

  // 调度新的编排
  try {
    const id = await workflowClient.scheduleNewWorkflow(workflow);
    console.log(`编排已调度，ID: ${id}`);

    // 等待编排完成
    const state = await workflowClient.waitForWorkflowCompletion(id, undefined, 30);

    console.log(`编排完成！结果: ${state?.serializedOutput}`);
  } catch (error) {
    console.error("调度或等待编排时出错:", error);
  }

  // 停止工作者和客户端
  await workflowRuntime.stop();
  await workflowClient.stop();

  // 停止 dapr sidecar
  process.exit(0);
}

start().catch((e) => {
  console.error(e);
  process.exit(1);
});

// 获取要并行处理的 N 个工作项的列表。
object[] workBatch = await context.CallActivityAsync<object[]>("GetWorkBatch", null);

// 调度并行任务，但不等待它们完成。
var parallelTasks = new List<Task<int>>(workBatch.Length);
for (int i = 0; i < workBatch.Length; i++)
{
    Task<int> task = context.CallActivityAsync<int>("ProcessWorkItem", workBatch[i]);
    parallelTasks.Add(task);
}

// 一切都已调度。在此处等待，直到所有并行任务完成。
await Task.WhenAll(parallelTasks);

// 聚合所有 N 个输出并发布结果。
int sum = parallelTasks.Sum(t => t.Result);
await context.CallActivityAsync("PostResults", sum);

public class FaninoutWorkflow extends Workflow {
    @Override
    public WorkflowStub create() {
        return ctx -> {
            // 获取要并行处理的 N 个工作项的列表。
            Object[] workBatch = ctx.callActivity("GetWorkBatch", Object[].class).await();
            // 调度并行任务，但不等待它们完成。
            List<Task<Integer>> tasks = Arrays.stream(workBatch)
                    .map(workItem -> ctx.callActivity("ProcessWorkItem", workItem, int.class))
                    .collect(Collectors.toList());
            // 一切都已调度。在此处等待，直到所有并行任务完成。
            List<Integer> results = ctx.allOf(tasks).await();
            // 聚合所有 N 个输出并发布结果。
            int sum = results.stream().mapToInt(Integer::intValue).sum();
            ctx.complete(sum);
        };
    }
}

func BatchProcessingWorkflow(ctx *workflow.WorkflowContext) (any, error) {
	var input int
	if err := ctx.GetInput(&input); err != nil {
		return 0, err
	}
	var workBatch []int
	if err := ctx.CallActivity(GetWorkBatch, workflow.ActivityInput(input)).Await(&workBatch); err != nil {
		return 0, err
	}
	parallelTasks := workflow.NewTaskSlice(len(workBatch))
	for i, workItem := range workBatch {
		parallelTasks[i] = ctx.CallActivity(ProcessWorkItem, workflow.ActivityInput(workItem))
	}
	var outputs int
	for _, task := range parallelTasks {
		var output int
		err := task.Await(&output)
		if err == nil {
			outputs += output
		} else {
			return 0, err
		}
	}
	if err := ctx.CallActivity(ProcessResults, workflow.ActivityInput(outputs)).Await(nil); err != nil {
		return 0, err
	}
	return 0, nil
}
func GetWorkBatch(ctx workflow.ActivityContext) (any, error) {
	var batchSize int
	if err := ctx.GetInput(&batchSize); err != nil {
		return 0, err
	}
	batch := make([]int, batchSize)
	for i := 0; i < batchSize; i++ {
		batch[i] = i
	}
	return batch, nil
}
func ProcessWorkItem(ctx workflow.ActivityContext) (any, error) {
	var workItem int
	if err := ctx.GetInput(&workItem); err != nil {
		return 0, err
	}
	fmt.Printf("处理工作项: %d\n", workItem)
	time.Sleep(time.Second * 5)
	result := workItem * 2
	fmt.Printf("工作项 %d 已处理. 结果: %d\n", workItem, result)
	return result, nil
}
func ProcessResults(ctx workflow.ActivityContext) (any, error) {
	var finalResult int
	if err := ctx.GetInput(&finalResult); err != nil {
		return 0, err
	}
	fmt.Printf("最终结果: %d\n", finalResult)
	return finalResult, nil
}

const statusMonitorWorkflow: TWorkflow = async function* (ctx: WorkflowContext): any {
    let duration;
    const status = yield ctx.callActivity(checkStatusActivity);
    if (status === "healthy") {
      // 在健康状态下检查频率较低
      // 设置持续时间为 1 小时
      duration = 60 * 60;
    } else {
      yield ctx.callActivity(alertActivity, "job unhealthy");
      // 在不健康状态下检查频率较高
      // 设置持续时间为 5 分钟
      duration = 5 * 60;
    }

    // 将工作流置于休眠状态，直到确定的时间
    ctx.createTimer(duration);

    // 使用更新的状态从头开始重新启动
    ctx.continueAsNew();
  };

public override async Task<object> RunAsync(WorkflowContext context, MyEntityState myEntityState)
{
    TimeSpan nextSleepInterval;

    var status = await context.CallActivityAsync<string>("GetStatus");
    if (status == "healthy")
    {
        myEntityState.IsHealthy = true;

        // 在健康状态下检查频率较低
        nextSleepInterval = TimeSpan.FromMinutes(60);
    }
    else
    {
        if (myEntityState.IsHealthy)
        {
            myEntityState.IsHealthy = false;
            await context.CallActivityAsync("SendAlert", myEntityState);
        }

        // 在不健康状态下检查频率较高
        nextSleepInterval = TimeSpan.FromMinutes(5);
    }

    // 将工作流置于休眠状态，直到确定的时间
    await context.CreateTimer(nextSleepInterval);

    // 使用更新的状态从头开始重新启动
    context.ContinueAsNew(myEntityState);
    return null;
}

public class MonitorWorkflow extends Workflow {

  @Override
  public WorkflowStub create() {
    return ctx -> {

      Duration nextSleepInterval;

      var status = ctx.callActivity(DemoWorkflowStatusActivity.class.getName(), DemoStatusActivityOutput.class).await();
      var isHealthy = status.getIsHealthy();

      if (isHealthy) {
        // 在健康状态下检查频率较低
        nextSleepInterval = Duration.ofMinutes(60);
      } else {

        ctx.callActivity(DemoWorkflowAlertActivity.class.getName()).await();

        // 在不健康状态下检查频率较高
        nextSleepInterval = Duration.ofMinutes(5);
      }

      // 将工作流置于休眠状态，直到确定的时间
      try {
        ctx.createTimer(nextSleepInterval);
      } catch (InterruptedException e) {
        throw new RuntimeException(e);
      }

      // 使用更新的状态从头开始重新启动
      ctx.continueAsNew();
    }
  }
}

type JobStatus struct {
	JobID     string `json:"job_id"`
	IsHealthy bool   `json:"is_healthy"`
}
func StatusMonitorWorkflow(ctx *workflow.WorkflowContext) (any, error) {
	var sleepInterval time.Duration
	var job JobStatus
	if err := ctx.GetInput(&job); err != nil {
		return "", err
	}
	var status string
	if err := ctx.CallActivity(CheckStatus, workflow.ActivityInput(job)).Await(&status); err != nil {
		return "", err
	}
	if status == "healthy" {
		job.IsHealthy = true
		sleepInterval = time.Minutes * 60
	} else {
		if job.IsHealthy {
			job.IsHealthy = false
			err := ctx.CallActivity(SendAlert, workflow.ActivityInput(fmt.Sprintf("Job '%s' is unhealthy!", job.JobID))).Await(nil)
			if err != nil {
				return "", err
			}
		}
		sleepInterval = time.Minutes * 5
	}
	if err := ctx.CreateTimer(sleepInterval).Await(nil); err != nil {
		return "", err
	}
	ctx.ContinueAsNew(job, false)
	return "", nil
}
func CheckStatus(ctx workflow.ActivityContext) (any, error) {
	statuses := []string{"healthy", "unhealthy"}
	return statuses[rand.Intn(1)], nil
}
func SendAlert(ctx workflow.ActivityContext) (any, error) {
	var message string
	if err := ctx.GetInput(&message); err != nil {
		return "", err
	}
	fmt.Printf("*** Alert: %s", message)
	return "", nil
}

import {
  Task,
  DaprWorkflowClient,
  WorkflowActivityContext,
  WorkflowContext,
  WorkflowRuntime,
  TWorkflow,
} from "@dapr/dapr";
import * as readlineSync from "readline-sync";

// 将整个代码包装在一个立即调用的异步函数中
async function start() {
  class Order {
    cost: number;
    product: string;
    quantity: number;
    constructor(cost: number, product: string, quantity: number) {
      this.cost = cost;
      this.product = product;
      this.quantity = quantity;
    }
  }

  function sleep(ms: number): Promise<void> {
    return new Promise((resolve) => setTimeout(resolve, ms));
  }

  // 更新 gRPC 客户端和工作者以使用本地地址和端口
  const daprHost = "localhost";
  const daprPort = "50001";
  const workflowClient = new DaprWorkflowClient({
    daprHost,
    daprPort,
  });
  const workflowRuntime = new WorkflowRuntime({
    daprHost,
    daprPort,
  });

  // 发送审批请求给经理的活动函数
  const sendApprovalRequest = async (_: WorkflowActivityContext, order: Order) => {
    // 模拟一些需要时间的工作
    await sleep(3000);
    console.log(`发送审批请求: ${order.product}`);
  };

  // 下订单的活动函数
  const placeOrder = async (_: WorkflowActivityContext, order: Order) => {
    console.log(`下订单: ${order.product}`);
  };

  // 表示采购订单工作流的编排函数
  const purchaseOrderWorkflow: TWorkflow = async function* (ctx: WorkflowContext, order: Order): any {
    // 低于 $1000 的订单自动批准
    if (order.cost < 1000) {
      return "Auto-approved";
    }

    // $1000 或以上的订单需要经理批准
    yield ctx.callActivity(sendApprovalRequest, order);

    // 必须在 24 小时内收到批准，否则将被取消。
    const tasks: Task<any>[] = [];
    const approvalEvent = ctx.waitForExternalEvent("approval_received");
    const timeoutEvent = ctx.createTimer(24 * 60 * 60);
    tasks.push(approvalEvent);
    tasks.push(timeoutEvent);
    const winner = ctx.whenAny(tasks);

    if (winner == timeoutEvent) {
      return "Cancelled";
    }

    yield ctx.callActivity(placeOrder, order);
    const approvalDetails = approvalEvent.getResult();
    return `Approved by ${approvalDetails.approver}`;
  };

  workflowRuntime
    .registerWorkflow(purchaseOrderWorkflow)
    .registerActivity(sendApprovalRequest)
    .registerActivity(placeOrder);

  // 将工作者启动包装在 try-catch 块中以处理启动期间的任何错误
  try {
    await workflowRuntime.start();
    console.log("工作者启动成功");
  } catch (error) {
    console.error("启动工作者时出错:", error);
  }

  // 调度新的编排
  try {
    const cost = readlineSync.questionInt("输入订单金额:");
    const approver = readlineSync.question("输入审批人:");
    const timeout = readlineSync.questionInt("输入订单超时时间（秒）:");
    const order = new Order(cost, "MyProduct", 1);
    const id = await workflowClient.scheduleNewWorkflow(purchaseOrderWorkflow, order);
    console.log(`编排已调度，ID: ${id}`);

    // 异步提示批准
    promptForApproval(approver, workflowClient, id);

    // 等待编排完成
    const state = await workflowClient.waitForWorkflowCompletion(id, undefined, timeout + 2);

    console.log(`编排完成！结果: ${state?.serializedOutput}`);
  } catch (error) {
    console.error("调度或等待编排时出错:", error);
  }

  // 停止工作者和客户端
  await workflowRuntime.stop();
  await workflowClient.stop();

  // 停止 dapr sidecar
  process.exit(0);
}

async function promptForApproval(approver: string, workflowClient: DaprWorkflowClient, id: string) {
  if (readlineSync.keyInYN("按 [Y] 批准订单... Y/是, N/否")) {
    const approvalEvent = { approver: approver };
    await workflowClient.raiseEvent(id, "approval_received", approvalEvent);
  } else {
    return "订单被拒绝";
  }
}

start().catch((e) => {
  console.error(e);
  process.exit(1);
});

public override async Task<OrderResult> RunAsync(WorkflowContext context, OrderPayload order)
{
    // ...(其他步骤)...

    // 需要对超过某个阈值的订单进行批准
    if (order.TotalCost > OrderApprovalThreshold)
    {
        try
        {
            // 请求人类批准此订单
            await context.CallActivityAsync(nameof(RequestApprovalActivity), order);

            // 暂停并等待人类批准订单
            ApprovalResult approvalResult = await context.WaitForExternalEventAsync<ApprovalResult>(
                eventName: "ManagerApproval",
                timeout: TimeSpan.FromDays(3));
            if (approvalResult == ApprovalResult.Rejected)
            {
                // 订单被拒绝，在此结束工作流
                return new OrderResult(Processed: false);
            }
        }
        catch (TaskCanceledException)
        {
            // 批准超时会导致自动取消订单
            return new OrderResult(Processed: false);
        }
    }

    // ...(其他步骤)...

    // 以成功结果结束工作流
    return new OrderResult(Processed: true);
}

public class ExternalSystemInteractionWorkflow extends Workflow {
    @Override
    public WorkflowStub create() {
        return ctx -> {
            // ...其他步骤...
            Integer orderCost = ctx.getInput(int.class);
            // 需要对超过某个阈值的订单进行批准
            if (orderCost > ORDER_APPROVAL_THRESHOLD) {
                try {
                    // 请求人类批准此订单
                    ctx.callActivity("RequestApprovalActivity", orderCost, Void.class).await();
                    // 暂停并等待人类批准订单
                    boolean approved = ctx.waitForExternalEvent("ManagerApproval", Duration.ofDays(3), boolean.class).await();
                    if (!approved) {
                        // 订单被拒绝，在此结束工作流
                        ctx.complete("Process reject");
                    }
                } catch (TaskCanceledException e) {
                    // 批准超时会导致自动取消订单
                    ctx.complete("Process cancel");
                }
            }
            // ...其他步骤...

            // 以成功结果结束工作流
            ctx.complete("Process approved");
        };
    }
}