Method dispatch

internal/nodehub/hub.go:295–337

(c *conn, msg *nodev1.NodeMessage)

}

func (h *Hub) dispatch(c *conn, msg *nodev1.NodeMessage) {
	h.metrics.markSeen(c.nodeID)
	event := msg.GetEvent()
	if event != "" {
		// 优先派发到流订阅；命中则不再走 PushHandler。
		if id := msg.GetId(); id != "" {
			if v, ok := c.streamSubs.Load(id); ok {
				s := v.(*Stream)
				s.deliver(StreamFrame{Event: event, Body: append([]byte(nil), msg.GetBody()...)})
				return
			}
		}
		h.handleEvent(c, msg)
		return
	}
	id := msg.GetId()
	if id == "" {
		// 既无 id 也无 event → 丢弃
		return
	}
	// 流终态：node 在流结束时回一条普通响应（Ok=true 或 Ok=false+Error）
	if v, ok := c.streamSubs.Load(id); ok {
		s := v.(*Stream)
		var endErr error
		if !msg.GetOk() {
			es := msg.GetError()
			if es == "" {
				es = "node returned not-ok with empty error"
			}
			endErr = errors.New(es)
		}
		s.closeFromHub(endErr)
		return
	}
	if v, ok := c.pending.Load(id); ok {
		ch := v.(chan *nodev1.NodeMessage)
		select {
		case ch <- msg:
		default:
			// 通道已满（不应发生，buffer=1 且只投递一次）
		}
	}
}

func (h *Hub) handleEvent(c *conn, msg *nodev1.NodeMessage) {
	body := msg.GetBody()