NixMsg/sdk/go/transport_fake.go

package nixmsg

import (
	"context"
	"encoding/json"
	"sync"
	"sync/atomic"
)

// FakeTransport 单测用假 MQTT：不启真实网络。
type FakeTransport struct {
	mu       sync.Mutex
	cfg      transportConfig
	cred     atomic.Value
	up       [][]byte
	stopped  atomic.Bool
	online   bool
	connects []fakeConnect

	// AutoHello 为 true 时自动回复 hello（默认 true）。
	AutoHello bool
	// HelloToken 自动 hello 响应里的 session_token。
	HelloToken string
	// HelloServerTimeMs 自动 hello 的 server_time_ms。
	HelloServerTimeMs int64
	MaxBodyBytes      int
	MaxMetaBytes      int
	MaxFrameBytes     int
}

type fakeConnect struct {
	CleanStart    bool
	SessionExpiry uint32
	Password      string
}

// NewFakeTransport 创建假传输；默认自动回复 hello。
func NewFakeTransport() *FakeTransport {
	f := &FakeTransport{
		AutoHello:         true,
		HelloToken:        "nst_test_token",
		HelloServerTimeMs: 1_750_000_000_000,
		MaxBodyBytes:      262144,
		MaxMetaBytes:      4096,
		MaxFrameBytes:     786432,
	}
	f.cred.Store("")
	return f
}

func (f *FakeTransport) SetCredential(passwordOrToken string) {
	f.cred.Store(passwordOrToken)
}

func (f *FakeTransport) Start(_ context.Context, cfg transportConfig) error {
	f.mu.Lock()
	f.cfg = cfg
	f.mu.Unlock()
	return f.SimulateConnectOK()
}

func (f *FakeTransport) PublishUp(payload []byte) error {
	f.mu.Lock()
	cp := append([]byte(nil), payload...)
	f.up = append(f.up, cp)
	auto := f.AutoHello
	f.mu.Unlock()

	var head struct {
		Type string `json:"type"`
		RID  string `json:"rid"`
	}
	_ = unmarshalJSON(payload, &head)
	if auto && head.Type == "hello" && head.RID != "" {
		f.replyHello(head.RID)
	}
	return nil
}

func (f *FakeTransport) replyHello(rid string) {
	f.mu.Lock()
	token := f.HelloToken
	st := f.HelloServerTimeMs
	mb, mm, mf := f.MaxBodyBytes, f.MaxMetaBytes, f.MaxFrameBytes
	f.mu.Unlock()
	resp, _ := marshalJSON(map[string]any{
		"v": 1, "type": "resp", "rid": rid, "ok": true,
		"data": map[string]any{
			"server_time_ms":       st,
			"server_version":       "0.1.0",
			"max_body_bytes":       mb,
			"max_meta_bytes":       mm,
			"max_frame_bytes":      mf,
			"max_ttl_seconds":      2592000,
			"max_schedule_seconds": 31536000,
			"ack_timeout_seconds":  300,
			"session_token":        token,
		},
	})
	f.InjectDown(resp)
}

func (f *FakeTransport) Stop(_ context.Context) error {
	f.stopped.Store(true)
	f.mu.Lock()
	cfg := f.cfg
	f.online = false
	f.mu.Unlock()
	if cfg.OnOffline != nil {
		cfg.OnOffline()
	}
	return nil
}

// SimulateConnectOK 模拟一次成功 CONNECT（含 Clean Start 记录）并完成订阅就绪。
func (f *FakeTransport) SimulateConnectOK() error {
	clean, expiry := buildCleanConnectFlags()
	pass, _ := f.cred.Load().(string)
	f.mu.Lock()
	f.connects = append(f.connects, fakeConnect{CleanStart: clean, SessionExpiry: expiry, Password: pass})
	cfg := f.cfg
	f.online = true
	f.mu.Unlock()
	if cfg.OnConnectPacket != nil {
		cfg.OnConnectPacket(clean, expiry)
	}
	if cfg.Backoff != nil {
		cfg.Backoff.MarkOnline()
	}
	if cfg.MQTTReady != nil {
		if err := cfg.MQTTReady(context.Background()); err != nil {
			return err
		}
	}
	if cfg.OnOnline != nil {
		cfg.OnOnline()
	}
	return nil
}

// SimulateReconnect 模拟断线再连（再次 Clean Start）。
func (f *FakeTransport) SimulateReconnect() error {
	f.mu.Lock()
	cfg := f.cfg
	f.online = false
	f.mu.Unlock()
	if cfg.Backoff != nil {
		cfg.Backoff.MarkOffline()
	}
	if cfg.OnOffline != nil {
		cfg.OnOffline()
	}
	return f.SimulateConnectOK()
}

// SimulateAuthFail 模拟 CONNACK 认证失败。
func (f *FakeTransport) SimulateAuthFail(reason AuthReason) {
	f.stopped.Store(true)
	f.mu.Lock()
	cfg := f.cfg
	f.mu.Unlock()
	if cfg.OnAuthFailed != nil {
		cfg.OnAuthFailed(reason)
	}
}

// SimulateKick 模拟 0x8E。
func (f *FakeTransport) SimulateKick() {
	f.stopped.Store(true)
	f.mu.Lock()
	cfg := f.cfg
	f.mu.Unlock()
	if cfg.OnKicked != nil {
		cfg.OnKicked()
	}
}

// InjectDown 注入下行帧。
func (f *FakeTransport) InjectDown(payload []byte) {
	f.mu.Lock()
	cfg := f.cfg
	f.mu.Unlock()
	if cfg.OnDown != nil {
		cfg.OnDown(payload)
	}
}

// ReplyOK 对指定 rid 回成功 resp。
func (f *FakeTransport) ReplyOK(rid string, data any) {
	var raw json.RawMessage
	if data != nil {
		raw, _ = marshalJSON(data)
	} else {
		raw = json.RawMessage(`{}`)
	}
	resp, _ := marshalJSON(map[string]any{
		"v": 1, "type": "resp", "rid": rid, "ok": true, "data": raw,
	})
	f.InjectDown(resp)
}

// ReplyErr 对指定 rid 回失败 resp。
func (f *FakeTransport) ReplyErr(rid, code, message string) {
	resp, _ := marshalJSON(map[string]any{
		"v": 1, "type": "resp", "rid": rid, "ok": false,
		"error": map[string]any{"code": code, "message": message},
	})
	f.InjectDown(resp)
}

// UpFrames 已发布的上行帧副本。
func (f *FakeTransport) UpFrames() [][]byte {
	f.mu.Lock()
	defer f.mu.Unlock()
	out := make([][]byte, len(f.up))
	for i, p := range f.up {
		out[i] = append([]byte(nil), p...)
	}
	return out
}

// Connects 历次 CONNECT 参数。
func (f *FakeTransport) Connects() []fakeConnect {
	f.mu.Lock()
	defer f.mu.Unlock()
	out := make([]fakeConnect, len(f.connects))
	copy(out, f.connects)
	return out
}

// DrainUp 取出并清空上行。
func (f *FakeTransport) DrainUp() [][]byte {
	f.mu.Lock()
	defer f.mu.Unlock()
	out := f.up
	f.up = nil
	return out
}

// LastUp 最后一条上行（解析为通用 map）。
func (f *FakeTransport) LastUp() map[string]any {
	f.mu.Lock()
	defer f.mu.Unlock()
	if len(f.up) == 0 {
		return nil
	}
	var m map[string]any
	_ = unmarshalJSON(f.up[len(f.up)-1], &m)
	return m
}

// FindUp 找指定 type 的上行帧。
func (f *FakeTransport) FindUp(typ string) []map[string]any {
	f.mu.Lock()
	defer f.mu.Unlock()
	var out []map[string]any
	for _, p := range f.up {
		var m map[string]any
		if unmarshalJSON(p, &m) == nil {
			if m["type"] == typ {
				out = append(out, m)
			}
		}
	}
	return out
}