工作流插件

type Engine interface {
    Start(ctx context.Context) error
    Stop(ctx context.Context) error
    Submit(ctx context.Context, task *Task) (TaskID, error)
    GetResult(ctx context.Context, id TaskID) (*Result, error)
    Cancel(ctx context.Context, id TaskID) error
}

import asynqPlugin "github.com/tx7do/go-wind-plugins/workflow/asynq"

engine := asynqPlugin.New(
    asynqPlugin.WithAddr("localhost:6379"),
    asynqPlugin.WithConcurrency(10),
    asynqPlugin.WithQueues(
        asynqPlugin.Queue{Name: "critical", Priority: 6},
        asynqPlugin.Queue{Name: "default", Priority: 3},
        asynqPlugin.Queue{Name: "low", Priority: 1},
    ),
)

type EmailTask struct {
    To      string
    Subject string
    Body    string
}

func HandleEmailTask(ctx context.Context, task *asynqPlugin.Task) error {
    var payload EmailTask
    json.Unmarshal(task.Payload, &payload)
    return sendEmail(payload.To, payload.Subject, payload.Body)
}

// 注册 Handler
engine.Register("send:email", HandleEmailTask)

payload, _ := json.Marshal(EmailTask{
    To: "user@example.com",
    Subject: "Welcome",
    Body: "Welcome to GoWind!",
})

taskID, _ := engine.Submit(ctx, &workflow.Task{
    Type:    "send:email",
    Payload: payload,
    Queue:   "default",
    Delay:   5 * time.Minute,    // 延迟执行
    MaxRetry: 3,
    Timeout:  30 * time.Second,
})

// Cron 表达式
engine.Schedule("0 9 * * *", &workflow.Task{
    Type: "daily:report",
    Payload: reportPayload,
})

import temporalPlugin "github.com/tx7do/go-wind-plugins/workflow/temporal"

client, _ := temporalPlugin.New(
    temporalPlugin.WithHostPort("localhost:7233"),
    temporalPlugin.WithNamespace("default"),
    temporalPlugin.WithTaskQueue("my-queue"),
)

func OrderWorkflow(ctx workflow.Context, order Order) error {
    // 步骤 1：扣减库存
    var inventoryOK bool
    workflow.ExecuteActivity(ctx, DeductInventoryActivity, order).Get(ctx, &inventoryOK)
    if !inventoryOK {
        return errors.New("inventory deduction failed")
    }

    // 步骤 2：处理支付
    var paymentResult PaymentResult
    workflow.ExecuteActivity(ctx, PaymentActivity, order).Get(ctx, &paymentResult)

    // 步骤 3：发送确认邮件
    workflow.ExecuteActivity(ctx, SendConfirmationActivity, order).Get(ctx, nil)

    return nil
}

worker := temporalPlugin.NewWorker(client, "my-queue")
worker.RegisterWorkflow(OrderWorkflow)
worker.RegisterActivity(DeductInventoryActivity)
worker.RegisterActivity(PaymentActivity)
worker.RegisterActivity(SendConfirmationActivity)

worker.Start(ctx)

workflowID := "order-" + uuid.New().String()
client.ExecuteWorkflow(ctx, temporalPlugin.StartWorkflowOptions{
    ID:        workflowID,
    TaskQueue: "my-queue",
}, OrderWorkflow, order)