NixMsg/sdk/go/send.go

package nixmsg

import (
	"context"
	"encoding/json"
	"time"
)

// Send 发送消息；可在未连接时入队，重连后按原内容重交。
func (c *Client) Send(ctx context.Context, to Target, body Body, opt SendOptions) (SendResult, error) {
	c.mu.Lock()
	limits := c.limits
	skew := c.clockSkew
	maxQ := c.opts.SendQueueSize
	qLen := len(c.sendQ)
	stopped := c.stopReconnect || c.closed
	stopErr := c.stopErrLocked()
	c.mu.Unlock()

	if stopped {
		return SendResult{}, stopErr
	}

	if body.Enc == "" {
		body.Enc = "utf8"
	}
	if body.ContentType == "" && opt.ContentType != "" {
		body.ContentType = opt.ContentType
	} else if body.ContentType == "" {
		if body.Enc == "base64" {
			body.ContentType = "application/octet-stream"
		} else {
			body.ContentType = "text/plain; charset=utf-8"
		}
	} else if opt.ContentType != "" {
		body.ContentType = opt.ContentType
	}

	n, err := bodyDecodedLen(body)
	if err != nil {
		return SendResult{}, err
	}
	maxBody := limits.MaxBodyBytes
	if maxBody <= 0 {
		maxBody = 262144
	}
	if n > maxBody {
		return SendResult{}, apiErr(CodeBodyTooLarge, "正文超限")
	}

	id := opt.ID
	if id == "" {
		id = newMessageID()
	}

	frame := map[string]any{
		"v":    1,
		"type": "send",
		"rid":  c.nextRID(),
		"id":   id,
		"to":   to,
		"body": body,
	}
	if opt.Meta != nil {
		frame["meta"] = opt.Meta
	}
	if opt.TalkPassword != "" {
		frame["talk_password"] = opt.TalkPassword
	}
	if opt.Receipt != nil {
		frame["receipt"] = *opt.Receipt
	}
	if opt.Keep {
		off := map[string]any{"keep": true}
		if opt.TTL != nil {
			off["ttl_seconds"] = *opt.TTL
		}
		frame["offline"] = off
	}

	var sendAtMs *int64
	if opt.SendAt != nil && opt.Delay != nil {
		return SendResult{}, apiErr(CodeBadRequest, "sendAt 与 delay 互斥")
	}
	if opt.SendAt != nil {
		ms := opt.SendAt.UnixMilli() + skew
		sendAtMs = &ms
		frame["send_at_ms"] = ms
	} else if opt.Delay != nil {
		frame["delay_ms"] = opt.Delay.Milliseconds()
	}

	payload, err := marshalJSON(frame)
	if err != nil {
		return SendResult{}, err
	}
	maxFrame := limits.MaxFrameBytes
	if maxFrame <= 0 {
		maxFrame = 786432
	}
	if len(payload) > maxFrame {
		return SendResult{}, apiErr(CodeFrameTooLarge, "整帧超限")
	}

	item := &sendItem{
		frame:    frame,
		payload:  payload,
		id:       id,
		sendAtMs: sendAtMs,
		result:   make(chan sendOutcome, 1),
	}

	c.mu.Lock()
	if c.closed || c.stopReconnect {
		err := c.stopErrLocked()
		c.mu.Unlock()
		return SendResult{}, err
	}
	if qLen >= maxQ || len(c.sendQ) >= maxQ {
		c.mu.Unlock()
		return SendResult{}, apiErr(CodeQueueFull, "发送队列已满")
	}
	c.sendQ = append(c.sendQ, item)
	c.mu.Unlock()

	c.drainSendQueue()

	select {
	case <-ctx.Done():
		c.mu.Lock()
		if !item.inflight && !item.abandoned {
			nq := c.sendQ[:0]
			for _, it := range c.sendQ {
				if it != item {
					nq = append(nq, it)
				}
			}
			c.sendQ = nq
			c.mu.Unlock()
			return SendResult{}, ctx.Err()
		}
		item.abandoned = true
		c.mu.Unlock()
		return SendResult{}, apiErr(CodeResultUnknown, "结果未知，请用同一消息号重试")
	case out := <-item.result:
		return out.res, out.err
	}
}

func (c *Client) drainSendQueue() {
	for {
		c.mu.Lock()
		if !c.handshook || c.transport == nil || c.stopReconnect {
			c.mu.Unlock()
			return
		}
		maxFrame := c.limits.MaxFrameBytes
		if maxFrame <= 0 {
			maxFrame = 786432
		}
		var next *sendItem
		for _, it := range c.sendQ {
			if !it.inflight && !it.abandoned {
				next = it
				break
			}
		}
		if next == nil || c.inflight >= c.opts.MaxInflight {
			c.mu.Unlock()
			return
		}
		if len(next.payload) > maxFrame {
			c.mu.Unlock()
			c.finishSend(next, SendResult{}, apiErr(CodeFrameTooLarge, "整帧超限"))
			continue
		}
		next.epoch++
		captured := next.epoch
		next.inflight = true
		c.inflight++
		tr := c.transport
		payload := next.payload
		rid, _ := next.frame["rid"].(string)
		item := next
		lost := c.connLost
		c.mu.Unlock()

		go c.dispatchSend(tr, item, rid, payload, captured, lost)
	}
}

func (c *Client) dispatchSend(tr transport, item *sendItem, rid string, payload []byte, epoch uint64, lost <-chan struct{}) {
	ch := make(chan respFrame, 1)
	c.mu.Lock()
	if item.epoch != epoch || !item.inflight {
		c.mu.Unlock()
		return
	}
	c.pending[rid] = &pendingReq{rid: rid, ch: ch, isSend: true}
	c.mu.Unlock()

	if err := tr.PublishUp(payload); err != nil {
		c.mu.Lock()
		delete(c.pending, rid)
		if item.epoch != epoch || !item.inflight {
			c.mu.Unlock()
			return
		}
		item.inflight = false
		if c.inflight > 0 {
			c.inflight--
		}
		if c.stopReconnect {
			errStop := c.stopErrLocked()
			c.mu.Unlock()
			c.finishSend(item, SendResult{}, errStop)
			return
		}
		// 保留 id/body/send_at_ms，换新 rid 后继续泵，避免 Send 永久挂起。
		c.regenerateSendLocked(item)
		c.mu.Unlock()
		c.drainSendQueue()
		return
	}

	if lost == nil {
		lost = make(chan struct{})
	}

	var rf respFrame
	select {
	case rf = <-ch:
	case <-lost:
		return
	case <-c.ctx.Done():
		return
	}

	c.mu.Lock()
	valid := item.epoch == epoch && item.inflight
	c.mu.Unlock()
	if !valid {
		return
	}

	if !rf.OK {
		code, msg := CodeBadRequest, "发送失败"
		if rf.Error != nil {
			code, msg = rf.Error.Code, rf.Error.Message
		}
		if code == CodeRateLimited {
			c.mu.Lock()
			item.inflight = false
			if c.inflight > 0 {
				c.inflight--
			}
			delete(c.pending, rid)
			item.rateN++
			n := item.rateN
			c.regenerateSendLocked(item)
			c.mu.Unlock()
			d := backoffJitter(nominalDelay(n))
			time.AfterFunc(d, func() { c.drainSendQueue() })
			return
		}
		c.finishSend(item, SendResult{}, apiErr(code, msg))
		return
	}
	var sd SendResult
	_ = json.Unmarshal(rf.Data, &sd)
	if sd.ID == "" {
		sd.ID = item.id
	}
	c.finishSend(item, sd, nil)
}

func (c *Client) finishSend(item *sendItem, res SendResult, err error) {
	c.mu.Lock()
	out := item.result
	abandoned := item.abandoned
	nq := c.sendQ[:0]
	for _, it := range c.sendQ {
		if it != item {
			nq = append(nq, it)
		}
	}
	c.sendQ = nq
	if item.inflight {
		if c.inflight > 0 {
			c.inflight--
		}
		item.inflight = false
	}
	rid, _ := item.frame["rid"].(string)
	delete(c.pending, rid)
	c.mu.Unlock()
	if !abandoned {
		select {
		case out <- sendOutcome{res: res, err: err}:
		default:
		}
	}
	c.drainSendQueue()
}

// ResendPayloadForTest 返回队列中第一条发送帧的编码（测试用）。
func (c *Client) ResendPayloadForTest() []byte {
	c.mu.Lock()
	defer c.mu.Unlock()
	if len(c.sendQ) == 0 {
		return nil
	}
	return append([]byte(nil), c.sendQ[0].payload...)
}