235 lines
8.2 KiB
Go
235 lines
8.2 KiB
Go
// server 演示将 Orbit SDK 接入现有 HTTP 服务；平台仍独立部署。
|
||
package main
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"net"
|
||
"net/http"
|
||
"os"
|
||
"os/signal"
|
||
"strings"
|
||
"sync/atomic"
|
||
"syscall"
|
||
"time"
|
||
|
||
outbox "git.yuntuo.hk/k2safe/orbit/adapters/outbox/mysql"
|
||
"git.yuntuo.hk/k2safe/orbit/examples/http-service/internal/inbox"
|
||
"git.yuntuo.hk/k2safe/orbit/examples/http-service/internal/jobs"
|
||
orbit "git.yuntuo.hk/k2safe/orbit/sdk/go"
|
||
)
|
||
|
||
func main() {
|
||
if err := run(); err != nil {
|
||
fmt.Fprintln(os.Stderr, err)
|
||
os.Exit(1)
|
||
}
|
||
}
|
||
|
||
type config struct {
|
||
Token string `json:"token"`
|
||
Namespace string `json:"namespace_id"`
|
||
Queue string `json:"queue_id"`
|
||
}
|
||
|
||
func client() (*orbit.Client, config, error) {
|
||
c := config{Token: os.Getenv("ORBIT_TOKEN"), Namespace: os.Getenv("ORBIT_NAMESPACE_ID"), Queue: os.Getenv("ORBIT_QUEUE_ID")}
|
||
if file := os.Getenv("ORBIT_CREDENTIALS_FILE"); file != "" {
|
||
raw, err := os.ReadFile(file)
|
||
if err != nil {
|
||
return nil, c, errors.New("cannot read credential file")
|
||
}
|
||
var saved config
|
||
if json.Unmarshal(raw, &saved) != nil {
|
||
return nil, c, errors.New("invalid credential file")
|
||
}
|
||
if c.Token == "" {
|
||
c.Token = saved.Token
|
||
}
|
||
if c.Namespace == "" {
|
||
c.Namespace = saved.Namespace
|
||
}
|
||
if c.Queue == "" {
|
||
c.Queue = saved.Queue
|
||
}
|
||
}
|
||
if c.Token == "" || c.Namespace == "" || c.Queue == "" {
|
||
return nil, c, errors.New("Orbit token, namespace and queue are required")
|
||
}
|
||
api, err := orbit.NewClient(os.Getenv("ORBIT_URL"), orbit.Options{Token: c.Token})
|
||
return api, c, err
|
||
}
|
||
func run() error {
|
||
if len(os.Args) != 2 {
|
||
return errors.New("usage: server init|provision|serve")
|
||
}
|
||
command := os.Args[1]
|
||
if command != "init" && command != "provision" && command != "serve" {
|
||
return errors.New("unknown command")
|
||
}
|
||
process, signalStop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||
defer signalStop()
|
||
startup, stop := context.WithTimeout(process, 10*time.Second)
|
||
defer stop()
|
||
if command == "provision" {
|
||
api, c, err := client()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
// 首次使用专属队列前创建，键和内容固定；部署重试不可改名/换键。运行期间从配置复用ID。
|
||
result, err := api.CreateSubscription(startup, "http-service-sub-"+c.Queue, orbit.CreateSubscriptionRequest{Name: "http-service-" + c.Queue, QueueID: c.Queue, MaxPending: 100, MaxAttempts: 20})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
fmt.Println(result.Subscription.ID)
|
||
return nil
|
||
}
|
||
db, err := outbox.Open(startup, os.Getenv("APP_MYSQL_DSN"))
|
||
if err != nil {
|
||
return err
|
||
}
|
||
// 所有组件完成关闭后才关闭共享连接池；初始化/构造/监听失败也走同一回收路径。
|
||
defer db.Close()
|
||
if command == "init" {
|
||
if err = jobs.Init(startup, db); err != nil {
|
||
return err
|
||
}
|
||
if err = inbox.Init(startup, db); err != nil {
|
||
return err
|
||
}
|
||
fmt.Println("business_schema_initialized=true")
|
||
return nil
|
||
}
|
||
api, c, err := client()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
store, err := jobs.New(db, c.Namespace, c.Queue)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
// 尽早核验业务迁移；不会自动建表，也不依赖Orbit在线才能启动。
|
||
for _, query := range []string{
|
||
`SELECT namespace_id,queue_id,operation_id,state FROM sender_outbox LIMIT 0`,
|
||
`SELECT namespace_id,operation_id,source_ref,task_id FROM sender_jobs LIMIT 0`,
|
||
`SELECT consumer_name,event_id,event_digest FROM receiver_inbox LIMIT 0`,
|
||
`SELECT consumer_name,queue_id,task_id,last_revision,result_json FROM receiver_task_views LIMIT 0`,
|
||
} {
|
||
if _, err = db.ExecContext(startup, query); err != nil {
|
||
return errors.New("business schema missing")
|
||
}
|
||
}
|
||
receiver, err := inbox.New(db, "http-service", c.Queue)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
worker, err := orbit.NewWorker(api, orbit.WorkerOptions{QueueID: c.Queue, Capacity: 4, ShutdownTimeout: 20 * time.Second, Handlers: []orbit.WorkerHandler{{Capability: orbit.WorkerCapability{Handler: "http.object.prepare", HandlerVersion: "v1", PayloadVersion: "v1"}, Handle: prepare}}})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
consumer, err := orbit.NewEventConsumer(api, orbit.EventConsumerOptions{SubscriptionID: os.Getenv("ORBIT_SUBSCRIPTION_ID"), Capacity: 4, ShutdownTimeout: 10 * time.Second, Handle: receiver.Handle})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
ready := new(atomic.Bool)
|
||
mux := http.NewServeMux()
|
||
mux.HandleFunc("GET /health/live", func(w http.ResponseWriter, r *http.Request) { w.Write([]byte("live\n")) })
|
||
mux.HandleFunc("GET /health/ready", func(w http.ResponseWriter, r *http.Request) {
|
||
ctx, cancel := context.WithTimeout(r.Context(), 500*time.Millisecond)
|
||
defer cancel()
|
||
if !ready.Load() || db.PingContext(ctx) != nil {
|
||
http.Error(w, "not ready", 503)
|
||
return
|
||
}
|
||
w.Write([]byte("ready\n"))
|
||
})
|
||
mux.HandleFunc("POST /jobs", func(w http.ResponseWriter, r *http.Request) {
|
||
if !ready.Load() {
|
||
http.Error(w, "stopping", 503)
|
||
return
|
||
}
|
||
r.Body = http.MaxBytesReader(w, r.Body, 4096)
|
||
var in struct {
|
||
OperationID string `json:"operation_id"`
|
||
SourceRef string `json:"source_ref"`
|
||
}
|
||
d := json.NewDecoder(r.Body)
|
||
d.DisallowUnknownFields()
|
||
if d.Decode(&in) != nil || d.Decode(new(any)) != io.EOF {
|
||
http.Error(w, "invalid request", 400)
|
||
return
|
||
}
|
||
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
|
||
defer cancel()
|
||
payload, _ := json.Marshal(map[string]string{"source_ref": in.SourceRef})
|
||
duplicate, err := store.Enqueue(ctx, in.OperationID, in.SourceRef, orbit.SubmitTaskRequest{QueueID: c.Queue, Handler: "http.object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: payload, BusinessKey: in.OperationID, TaskTimeoutMS: 86400000, RecoverySafety: orbit.RecoveryIdempotent, Retry: orbit.RetryPolicy{MaxAttempts: 3, AttemptTimeoutMS: 10000}})
|
||
if err != nil {
|
||
httpFailure(w, err)
|
||
return
|
||
}
|
||
writeJSON(w, 202, map[string]any{"operation_id": in.OperationID, "local_committed": true, "duplicate": duplicate})
|
||
})
|
||
mux.HandleFunc("GET /jobs/{id}", func(w http.ResponseWriter, r *http.Request) {
|
||
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
|
||
defer cancel()
|
||
result, err := store.View(ctx, r.PathValue("id"))
|
||
if err != nil {
|
||
httpFailure(w, err)
|
||
return
|
||
}
|
||
writeJSON(w, 200, result)
|
||
})
|
||
addr := os.Getenv("APP_HTTP_ADDR")
|
||
if addr == "" {
|
||
addr = "127.0.0.1:8090"
|
||
}
|
||
host, _, err := net.SplitHostPort(addr)
|
||
if err != nil || net.ParseIP(host) == nil || !net.ParseIP(host).IsLoopback() {
|
||
return errors.New("example requires a loopback APP_HTTP_ADDR; mount handlers behind your application authentication for production")
|
||
}
|
||
listener, err := net.Listen("tcp", addr)
|
||
if err != nil {
|
||
return errors.New("cannot listen on APP_HTTP_ADDR")
|
||
}
|
||
defer listener.Close()
|
||
server := &http.Server{Handler: mux, ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 10 * time.Second, WriteTimeout: 10 * time.Second, IdleTimeout: 60 * time.Second, MaxHeaderBytes: 16 << 10}
|
||
fmt.Printf("http_listening=%s\n", listener.Addr())
|
||
return serve(process, listener, server, ready, []component{{"outbox", func(ctx context.Context) error {
|
||
return store.Run(ctx, api, outbox.RunOptions{ShutdownTimeout: 20 * time.Second})
|
||
}}, {"worker", worker.Run}}, component{"events", consumer.Run}, 25*time.Second)
|
||
}
|
||
func httpFailure(w http.ResponseWriter, err error) {
|
||
code, text := 503, "unconfirmed; retry original operation and content"
|
||
if errors.Is(err, outbox.ErrInvalid) {
|
||
code, text = 400, "invalid request"
|
||
}
|
||
if errors.Is(err, outbox.ErrConflict) {
|
||
code, text = 409, "operation content conflict"
|
||
}
|
||
if errors.Is(err, outbox.ErrNotFound) {
|
||
code, text = 404, "operation not found"
|
||
}
|
||
http.Error(w, text, code)
|
||
}
|
||
func writeJSON(w http.ResponseWriter, status int, v any) {
|
||
w.Header().Set("Content-Type", "application/json")
|
||
w.WriteHeader(status)
|
||
_ = json.NewEncoder(w).Encode(v)
|
||
}
|
||
func prepare(ctx context.Context, e orbit.WorkerExecution) (json.RawMessage, error) {
|
||
var in struct {
|
||
SourceRef string `json:"source_ref"`
|
||
}
|
||
if json.Unmarshal(e.Task.Payload, &in) != nil || in.SourceRef == "" {
|
||
return nil, &orbit.WorkerFailure{Code: "INVALID_SOURCE"}
|
||
}
|
||
if err := ctx.Err(); err != nil {
|
||
return nil, err
|
||
}
|
||
// 纯计算可重复；实际外部副作用必须由业务实现幂等或核对，不能照搬幂等声明。
|
||
return json.Marshal(map[string]string{"source_ref": in.SourceRef, "normalized_ref": strings.ToUpper(in.SourceRef)})
|
||
}
|