Шаблон взаимодействия с человеком

• Завершите чтение быстрой статьи.
• Клонировать или скачать проект с примерами из GitHub (использует внутрипроцессную модель)

• Завершите чтение быстрой статьи.
• Клонируйте или скачайте проект с образцами из GitHub

• Завершите чтение быстрой статьи.
• Клонируйте или скачайте проект с образцами из GitHub

• Завершите чтение быстрой статьи.

Функция E4_SmsPhoneVerification использует стандартный файл function.json для функций оркестратора.

Ниже приведен код, реализующий функцию:

Пример PowerShell скоро появится.

Пример Java скоро появится.

using Microsoft.DurableTask;
using System;
using System.Threading;
using System.Threading.Tasks;

[DurableTask(nameof(ApprovalOrchestration))]
public class ApprovalOrchestration : TaskOrchestrator<ApprovalRequestData, ApprovalResult>
{
    public override async Task<ApprovalResult> RunAsync(
        TaskOrchestrationContext context, ApprovalRequestData input)
    {
        string requestId = input.RequestId;
        double timeoutHours = input.TimeoutHours;

        // Step 1: Submit the approval request (notify approver)
        SubmissionResult submissionResult = await context.CallActivityAsync<SubmissionResult>(
            nameof(SubmitApprovalRequestActivity), input);

        // Make the status available via custom status
        context.SetCustomStatus(submissionResult);

        // Step 2: Create a durable timer for the timeout
        DateTime timeoutDeadline = context.CurrentUtcDateTime.AddHours(timeoutHours);

        using var timeoutCts = new CancellationTokenSource();
        Task timeoutTask = context.CreateTimer(timeoutDeadline, timeoutCts.Token);

        // Step 3: Wait for an external event (approval/rejection)
        Task<ApprovalResponseData> approvalTask = context.WaitForExternalEvent<ApprovalResponseData>(
            "approval_response");

        // Step 4: Wait for either the timeout or the approval response
        Task completedTask = await Task.WhenAny(approvalTask, timeoutTask);

        // Step 5: Process based on which task completed
        ApprovalResult result;

        if (completedTask == approvalTask)
        {
            // Human responded in time - cancel the timeout timer
            timeoutCts.Cancel();

            ApprovalResponseData approvalData = approvalTask.Result;

            // Process the approval
            result = await context.CallActivityAsync<ApprovalResult>(
                nameof(ProcessApprovalActivity),
                new ProcessApprovalInput
                {
                    RequestId = requestId,
                    IsApproved = approvalData.IsApproved,
                    Approver = approvalData.Approver
                });
        }
        else
        {
            // Timeout occurred
            result = new ApprovalResult
            {
                RequestId = requestId,
                Status = "Timeout",
                ProcessedAt = context.CurrentUtcDateTime.ToString("o")
            };
        }

        return result;
    }
}

import {
  OrchestrationContext,
  TOrchestrator,
  whenAny,
} from "@microsoft/durabletask-js";

const approvalOrchestrator: TOrchestrator = async function* (
  ctx: OrchestrationContext,
  amount: number
): any {
  // Step 1: Submit the request
  const requestId: string = yield ctx.callActivity(submitRequest, { amount });

  ctx.setCustomStatus({ stage: "Awaiting approval", requestId });

  // Step 2: Race external event vs timer
  const approvalEvent = ctx.waitForExternalEvent<{ approved: boolean }>(
    "approval"
  );
  const timeout = ctx.createTimer(5); // 5-second timeout

  const winner = yield whenAny([approvalEvent, timeout]);

  let result: string;
  if (winner === approvalEvent) {
    // Human responded in time
    const decision = approvalEvent.getResult();
    ctx.setCustomStatus({
      stage: "Processing",
      requestId,
      approved: decision.approved,
    });
    result = yield ctx.callActivity(processApproval, {
      requestId,
      approved: decision.approved,
    });
  } else {
    // Timer fired first — timed out
    ctx.setCustomStatus({ stage: "Timed out", requestId });
    result = yield ctx.callActivity(notifyTimeout, requestId);
  }

  ctx.setCustomStatus({ stage: "Completed", requestId });
  return result;
};

import datetime
from durabletask import task

def human_interaction_orchestrator(ctx: task.OrchestrationContext, input_data: dict) -> dict:
    """
    Orchestrator that demonstrates the human interaction pattern.

    Submits an approval request, then waits for a human to approve or reject.
    """
    request_id = input_data.get("request_id")
    timeout_hours = input_data.get("timeout_hours", 24)

    # Step 1: Submit the approval request (notify approver)
    request_data = {
        "request_id": request_id,
        "requester": input_data.get("requester"),
        "item": input_data.get("item")
    }

    submission_result = yield ctx.call_activity("submit_approval_request", input=request_data)

    # Make the status available via custom status
    ctx.set_custom_status(submission_result)

    # Step 2: Create a durable timer for the timeout
    timeout_deadline = ctx.current_utc_datetime + datetime.timedelta(hours=timeout_hours)
    timeout_task = ctx.create_timer(timeout_deadline)

    # Step 3: Wait for an external event (approval/rejection)
    approval_task = ctx.wait_for_external_event("approval_response")

    # Step 4: Wait for either the timeout or the approval response
    winner_task = yield task.when_any([approval_task, timeout_task])

    # Step 5: Process based on which task completed
    if winner_task == approval_task:
        # Human responded in time
        approval_data = yield approval_task

        # Process the approval
        result = yield ctx.call_activity("process_approval", input={
            "request_id": request_id,
            "is_approved": approval_data.get("is_approved", False),
            "approver": approval_data.get("approver", "Unknown")
        })
    else:
        # Timeout occurred
        result = {
            "request_id": request_id,
            "status": "Timeout",
            "timed_out_at": ctx.current_utc_datetime.isoformat()
        }

    return result

Этот пример показан для .NET, JavaScript, Java и Python.

import com.microsoft.durabletask.*;
import com.microsoft.durabletask.azuremanaged.DurableTaskSchedulerWorkerExtensions;
import java.time.Duration;

DurableTaskGrpcWorker worker = DurableTaskSchedulerWorkerExtensions.createWorkerBuilder(connectionString)
    .addOrchestration(new TaskOrchestrationFactory() {
        @Override
        public String getName() { return "ApprovalWorkflow"; }

        @Override
        public TaskOrchestration create() {
            return ctx -> {
                // Get the approval request
                ApprovalRequest request = ctx.getInput(ApprovalRequest.class);

                // Set initial status
                ctx.setCustomStatus(new WorkflowStatus("Waiting for approval", null));

                // Create a timeout timer
                Task timeoutTask = ctx.createTimer(
                    Duration.ofHours((long) request.timeoutHours));

                // Wait for external event (approval response)
                Task<ApprovalResponse> approvalTask = ctx.waitForExternalEvent(
                    "approval_response", ApprovalResponse.class);

                // Wait for either approval or timeout
                Task<?> winner = ctx.anyOf(approvalTask, timeoutTask).await();

                if (winner == approvalTask) {
                    // Human responded in time
                    ApprovalResponse response = approvalTask.await();
                    String status = response.isApproved ? "APPROVED" : "REJECTED";
                    ctx.setCustomStatus(new WorkflowStatus(status, response));
                    ctx.complete(new WorkflowResult(
                        status,
                        response.isApproved ? "Request approved" : "Request rejected"));
                } else {
                    // Timeout occurred
                    ctx.setCustomStatus(new WorkflowStatus("Timed out", null));
                    ctx.complete(new WorkflowResult("TIMEOUT", "Request timed out"));
                }
            };
        }
    })
    .build();

Файл function.json определяется следующим образом:

Этот код создает четырехзначный код подтверждения и отправляет SMS-сообщение.

Пример PowerShell скоро появится.

Пример Java скоро появится.

using Microsoft.DurableTask;
using Microsoft.Extensions.Logging;
using System;
using System.Threading.Tasks;

[DurableTask(nameof(SubmitApprovalRequestActivity))]
public class SubmitApprovalRequestActivity : TaskActivity<ApprovalRequestData, SubmissionResult>
{
    private readonly ILogger<SubmitApprovalRequestActivity> _logger;

    public SubmitApprovalRequestActivity(ILogger<SubmitApprovalRequestActivity> logger)
    {
        _logger = logger;
    }

    public override Task<SubmissionResult> RunAsync(
        TaskActivityContext context, ApprovalRequestData input)
    {
        _logger.LogInformation(
            "Submitting approval request {RequestId} from {Requester} for {Item}",
            input.RequestId, input.Requester, input.Item);

        // In a real system, this would send an email, notification, or update a database
        var result = new SubmissionResult
        {
            RequestId = input.RequestId,
            Status = "Pending",
            SubmittedAt = DateTime.UtcNow.ToString("o"),
            ApprovalUrl = $"http://localhost:8000/api/approvals/{input.RequestId}"
        };

        return Task.FromResult(result);
    }
}

import { ActivityContext } from "@microsoft/durabletask-js";

const submitRequest = async (
  _ctx: ActivityContext,
  request: { amount: number }
): Promise<string> => {
  console.log(
    `[submitRequest] Purchase request submitted: $${request.amount}`
  );
  return `REQ-${Date.now()}`;
};

import datetime
from durabletask import task

def submit_approval_request(ctx: task.ActivityContext, request_data: dict) -> dict:
    """
    Activity that submits an approval request.
    In a real application, this would notify a human approver via email, message, etc.
    """
    request_id = request_data.get("request_id")
    requester = request_data.get("requester")
    item = request_data.get("item")

    # In a real system, this would send an email, notification, or update a database
    return {
        "request_id": request_id,
        "status": "Pending",
        "submitted_at": datetime.datetime.now().isoformat(),
        "approval_url": f"http://localhost:8000/api/approvals/{request_id}"
    }

Этот пример показан для .NET, JavaScript, Java и Python.

Пример Java отправляет встроенный запрос на утверждение. Для рабочего приложения создайте отдельное действие:

.addActivity(new TaskActivityFactory() {
    @Override
    public String getName() { return "SubmitApprovalRequest"; }

    @Override
    public TaskActivity create() {
        return ctx -> {
            ApprovalRequest request = ctx.getInput(ApprovalRequest.class);

            // In a real system, send email/notification
            logger.info("Submitting approval request for " + request.item);

            return new SubmissionResult(
                "Pending",
                Instant.now().toString(),
                "http://localhost:8000/api/approvals/" + request.requester);
        };
    }
})

using Microsoft.DurableTask;
using Microsoft.Extensions.Logging;
using System;
using System.Threading.Tasks;

[DurableTask(nameof(ProcessApprovalActivity))]
public class ProcessApprovalActivity : TaskActivity<ProcessApprovalInput, ApprovalResult>
{
    private readonly ILogger<ProcessApprovalActivity> _logger;

    public ProcessApprovalActivity(ILogger<ProcessApprovalActivity> logger)
    {
        _logger = logger;
    }

    public override Task<ApprovalResult> RunAsync(
        TaskActivityContext context, ProcessApprovalInput input)
    {
        string status = input.IsApproved ? "Approved" : "Rejected";
        _logger.LogInformation(
            "Processing {Status} request {RequestId} by {Approver}",
            status, input.RequestId, input.Approver);

        // In a real system, this would update a database, trigger workflows, etc.
        var result = new ApprovalResult
        {
            RequestId = input.RequestId,
            Status = status,
            ProcessedAt = DateTime.UtcNow.ToString("o"),
            Approver = input.Approver
        };

        return Task.FromResult(result);
    }
}

// Data classes
public class ApprovalRequestData
{
    public string RequestId { get; set; } = string.Empty;
    public string Requester { get; set; } = string.Empty;
    public string Item { get; set; } = string.Empty;
    public double TimeoutHours { get; set; } = 24.0;
}

public class ApprovalResponseData
{
    public bool IsApproved { get; set; }
    public string Approver { get; set; } = string.Empty;
}

public class SubmissionResult
{
    public string RequestId { get; set; } = string.Empty;
    public string Status { get; set; } = string.Empty;
    public string SubmittedAt { get; set; } = string.Empty;
    public string ApprovalUrl { get; set; } = string.Empty;
}

public class ProcessApprovalInput
{
    public string RequestId { get; set; } = string.Empty;
    public bool IsApproved { get; set; }
    public string Approver { get; set; } = string.Empty;
}

public class ApprovalResult
{
    public string RequestId { get; set; } = string.Empty;
    public string Status { get; set; } = string.Empty;
    public string ProcessedAt { get; set; } = string.Empty;
    public string? Approver { get; set; }
}

import { ActivityContext } from "@microsoft/durabletask-js";

const processApproval = async (
  _ctx: ActivityContext,
  data: { requestId: string; approved: boolean }
): Promise<string> => {
  console.log(
    `[processApproval] Request ${data.requestId}: ${
      data.approved ? "APPROVED" : "REJECTED"
    }`
  );
  return data.approved ? "Order placed" : "Order cancelled";
};

const notifyTimeout = async (
  _ctx: ActivityContext,
  requestId: string
): Promise<string> => {
  console.log(
    `[notifyTimeout] Request ${requestId} timed out — auto-rejected`
  );
  return "Timed out — auto-rejected";
};

import datetime
from durabletask import task

def process_approval(ctx: task.ActivityContext, approval_data: dict) -> dict:
    """
    Activity that processes the approval once received.
    """
    request_id = approval_data.get("request_id")
    is_approved = approval_data.get("is_approved")
    approver = approval_data.get("approver")

    status = "Approved" if is_approved else "Rejected"

    # In a real system, this would update a database, trigger workflows, etc.
    return {
        "request_id": request_id,
        "status": status,
        "processed_at": datetime.datetime.now().isoformat(),
        "approver": approver
    }

Этот пример показан для .NET, JavaScript, Java и Python.

Пример Java обрабатывает одобрения непосредственно в оркестраторе. Для рабочего приложения создайте отдельное действие.

using System;
using System.Threading.Tasks;

var client = DurableTaskClientBuilder.UseDurableTaskScheduler(connectionString).Build();

// Schedule the approval workflow
var input = new ApprovalRequestData
{
    RequestId = "request-" + Guid.NewGuid().ToString(),
    Requester = "john.doe@example.com",
    Item = "Vacation Request - 5 days",
    TimeoutHours = 24
};

string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(
    nameof(ApprovalOrchestration), input);

Console.WriteLine($"Started approval workflow: {instanceId}");

// Simulate human approving the request
Console.WriteLine("Simulating approval...");
await Task.Delay(2000);

// Raise the approval event
var approvalResponse = new ApprovalResponseData
{
    IsApproved = true,
    Approver = "manager@example.com"
};

await client.RaiseEventAsync(instanceId, "approval_response", approvalResponse);

// Wait for completion
var result = await client.WaitForInstanceCompletionAsync(instanceId, getInputsAndOutputs: true);
Console.WriteLine($"Result: {result.ReadOutputAs<ApprovalResult>().Status}");

import {
  DurableTaskAzureManagedClientBuilder,
  DurableTaskAzureManagedWorkerBuilder,
} from "@microsoft/durabletask-js-azuremanaged";

const client = new DurableTaskAzureManagedClientBuilder()
  .connectionString(connectionString)
  .build();

const worker = new DurableTaskAzureManagedWorkerBuilder()
  .connectionString(connectionString)
  .addOrchestrator(approvalOrchestrator)
  .addActivity(submitRequest)
  .addActivity(processApproval)
  .addActivity(notifyTimeout)
  .build();

await worker.start();

// Schedule the approval workflow
const approvalId = await client.scheduleNewOrchestration(
  approvalOrchestrator,
  500 // amount
);

console.log(`Orchestration started: ${approvalId}`);

// Wait for it to reach "Awaiting approval", then send approval
await new Promise((r) => setTimeout(r, 3000));

await client.raiseOrchestrationEvent(approvalId, "approval", {
  approved: true,
});
console.log("Sent approval event");

const result = await client.waitForOrchestrationCompletion(
  approvalId,
  true,
  60
);
console.log(`Result: ${result?.serializedOutput}`);

await worker.stop();
await client.stop();

from durabletask.azuremanaged.client import DurableTaskSchedulerClient
import time
import uuid

client = DurableTaskSchedulerClient(
    host_address=endpoint,
    secure_channel=endpoint != "http://localhost:8080",
    taskhub=taskhub,
    token_credential=credential
)

# Schedule the approval workflow
input_data = {
    "request_id": f"request-{uuid.uuid4()}",
    "requester": "john.doe@example.com",
    "item": "Vacation Request - 5 days",
    "timeout_hours": 24
}

instance_id = client.schedule_new_orchestration(
    human_interaction_orchestrator,
    input=input_data
)

print(f"Started approval workflow: {instance_id}")

# Simulate human approving the request
print("Simulating approval...")
time.sleep(2)

# Raise the approval event
approval_response = {
    "is_approved": True,
    "approver": "manager@example.com"
}

client.raise_orchestration_event(instance_id, "approval_response", data=approval_response)

# Wait for completion
result = client.wait_for_orchestration_completion(instance_id, timeout=60)
print(f"Result: {result.serialized_output}")

Этот пример показан для .NET, JavaScript, Java и Python.

import java.time.Duration;
import java.time.Instant;
import java.util.UUID;

DurableTaskClient client = DurableTaskSchedulerClientExtensions
    .createClientBuilder(connectionString).build();

// Schedule the approval workflow
ApprovalRequest request = new ApprovalRequest(
    "john.doe@example.com",
    "Vacation Request - 5 days",
    24  // timeout hours
);

String instanceId = client.scheduleNewOrchestrationInstance(
    "ApprovalWorkflow",
    new NewOrchestrationInstanceOptions()
        .setInput(request)
        .setInstanceId("request-" + UUID.randomUUID().toString()));

System.out.println("Started approval workflow: " + instanceId);

// Simulate human approving the request
System.out.println("Simulating approval...");
Thread.sleep(2000);

// Raise the approval event
ApprovalResponse response = new ApprovalResponse(
    true,  // isApproved
    "manager@example.com",
    "Approved",
    Instant.now().toString()
);

client.raiseEvent(instanceId, "approval_response", response);

// Wait for completion
OrchestrationMetadata result = client.waitForInstanceCompletion(
    instanceId, Duration.ofSeconds(60), true);

System.out.println("Result: " + result.readOutputAs(WorkflowResult.class).status);