orbit/examples/task-outbox/cmd/sender/main.go

// sender 演示业务事务Outbox与可重启的有限发送器。
package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"os/signal"
	"syscall"
	"time"

	"git.yuntuo.hk/k2safe/orbit/examples/task-outbox/internal/outbox"
	orbit "git.yuntuo.hk/k2safe/orbit/sdk/go"
)

func main() {
	if err := run(); err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
}
func run() error {
	if len(os.Args) < 2 {
		return errors.New("usage: sender init|enqueue <operation_id> <source_ref>|send|status <operation_id>|demo")
	}
	command := os.Args[1]
	if !(command == "init" && len(os.Args) == 2 || command == "send" && len(os.Args) == 2 || command == "demo" && len(os.Args) == 2 || command == "enqueue" && len(os.Args) == 4 || command == "status" && len(os.Args) == 3) {
		return errors.New("invalid sender command or arguments")
	}
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()
	open, cancel := context.WithTimeout(ctx, 10*time.Second)
	db, err := outbox.Open(open, os.Getenv("ORBIT_SENDER_DSN"))
	cancel()
	if err != nil {
		return err
	}
	defer db.Close()
	if command == "init" {
		bounded, done := context.WithTimeout(ctx, 10*time.Second)
		defer done()
		if err = outbox.Init(bounded, db); err != nil {
			return err
		}
		fmt.Println("sender_schema_initialized=true foreign_keys=0")
		return nil
	}
	api, ns, queue, err := client()
	if err != nil {
		return err
	}
	store, err := outbox.New(db, ns, queue)
	if err != nil {
		return err
	}
	switch command {
	case "enqueue":
		id, source := os.Args[2], os.Args[3]
		request := request(queue, id, source)
		duplicate, err := store.Enqueue(ctx, id, source, request)
		if err != nil {
			return err
		}
		fmt.Printf("operation=%s local_committed=true duplicate=%t\n", id, duplicate)
		return nil
	case "status":
		r, err := store.Status(ctx, os.Args[2])
		if err != nil {
			return err
		}
		printRecord(r)
		return nil
	case "demo":
		return demo(ctx, store, api, queue)
	case "send":
		for ctx.Err() == nil {
			r, err := store.SendOnce(ctx, api)
			wait := 500 * time.Millisecond
			if err == nil {
				printRecord(r)
			} else if !errors.Is(err, outbox.ErrIdle) {
				fmt.Fprintln(os.Stderr, err)
				wait = 2 * time.Second
			}
			timer := time.NewTimer(wait)
			select {
			case <-ctx.Done():
				timer.Stop()
			case <-timer.C:
			}
		}
		return nil
	}
	return nil
}
func client() (*orbit.Client, string, string, error) {
	token, ns, queue := os.Getenv("ORBIT_TOKEN"), os.Getenv("ORBIT_NAMESPACE_ID"), os.Getenv("ORBIT_QUEUE_ID")
	if path := os.Getenv("ORBIT_CREDENTIALS_FILE"); path != "" {
		raw, err := os.ReadFile(path)
		if err != nil {
			return nil, "", "", errors.New("cannot read sender credential file")
		}
		var c struct {
			Token       string `json:"token"`
			NamespaceID string `json:"namespace_id"`
			QueueID     string `json:"queue_id"`
		}
		if json.Unmarshal(raw, &c) != nil {
			return nil, "", "", errors.New("invalid sender credential file")
		}
		if token == "" {
			token = c.Token
		}
		if ns == "" {
			ns = c.NamespaceID
		}
		if queue == "" {
			queue = c.QueueID
		}
	}
	if token == "" {
		return nil, "", "", errors.New("sender credential is required")
	}
	c, err := orbit.NewClient(os.Getenv("ORBIT_URL"), orbit.Options{Token: token})
	return c, ns, queue, err
}
func request(queue, id, source string) orbit.SubmitTaskRequest {
	payload, _ := json.Marshal(struct {
		SourceRef string `json:"source_ref"`
		Sequence  uint64 `json:"sequence"`
	}{source, 9007199254740993})
	return orbit.SubmitTaskRequest{QueueID: queue, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: payload, BusinessKey: id, TaskTimeoutMS: 60000}
}
func printRecord(r outbox.Record) {
	task := ""
	if r.TaskID != nil {
		task = *r.TaskID
	}
	fmt.Printf("operation=%s state=%s sends=%d code=%s task=%s\n", r.OperationID, r.State, r.SendCount, r.LastCode, task)
}
func demo(parent context.Context, s *outbox.Store, api *orbit.Client, queue string) error {
	ctx, stop := context.WithTimeout(parent, 20*time.Second)
	defer stop()
	id, err := orbit.NewEventMessageID()
	if err != nil {
		return err
	}
	// 演示每次新建一个操作；真实接入必须先由业务方保存ID再调用Enqueue。
	fmt.Printf("demo_operation=%s\n", id)
	in := request(queue, id, "objects/outbox-demo.txt")
	if _, err = s.Enqueue(ctx, id, "objects/outbox-demo.txt", in); err != nil {
		return err
	}
	duplicate, err := s.Enqueue(ctx, id, "objects/outbox-demo.txt", in)
	if err != nil || !duplicate {
		return errors.New("outbox local duplicate was not recovered")
	}
	for {
		r, err := s.Status(ctx, id)
		if err != nil {
			return err
		}
		if r.State == "REVIEW" {
			return errors.New("outbox demo needs review using its printed operation identity")
		}
		if r.State == "ACCEPTED" {
			task, err := api.GetTask(ctx, *r.TaskID)
			if err != nil {
				return err
			}
			if _, err = api.CancelTask(ctx, task.ID, "outbox-demo-cancel-"+id, orbit.CancelTaskRequest{ExpectedRevision: task.Revision, Reason: "outbox submission demo completed"}); err != nil {
				return err
			}
			printRecord(r)
			fmt.Println("local_duplicate=true accepted_once=true execution_success_not_implied=true")
			return nil
		}
		_, err = s.SendOnce(ctx, api)
		if err != nil && !errors.Is(err, outbox.ErrIdle) {
			return err
		}
		timer := time.NewTimer(100 * time.Millisecond)
		select {
		case <-ctx.Done():
			timer.Stop()
			return ctx.Err()
		case <-timer.C:
		}
	}
}