Method CallStream

internal/nodehub/stream.go:119–181

CallStream 发起一次流式调用：发送 ServerMessage{Method, Body} 后，把后续 node 推回的同 reqID push 帧（event="log" / "traceroute_hop"）通过 Stream.Frames() 暴露。node 端在流自然结束时回一条普通响应（Ok=true 或 Ok=false+Error）， 触发 Stream.Done()；caller 也可通过 Stream.Close() 主动取消（向 node 发送 cancel_id 帧）。 reqBody 用 json 编码（nil 表示空 body）。

(ctx context.Context, nodeID, method string, reqBody any)

//
// reqBody 用 json 编码（nil 表示空 body）。
func (h *Hub) CallStream(ctx context.Context, nodeID, method string, reqBody any) (*Stream, error) {
	h.mu.RLock()
	c, ok := h.conns[nodeID]
	h.mu.RUnlock()
	if !ok {
		return nil, ErrNodeOffline
	}

	var bodyBytes []byte
	if reqBody != nil {
		b, err := json.Marshal(reqBody)
		if err != nil {
			return nil, err
		}
		bodyBytes = b
	}

	reqID, err := newReqID()
	if err != nil {
		return nil, err
	}

	s := &Stream{
		conn:   c,
		reqID:  reqID,
		frames: make(chan StreamFrame, streamFramesBuffer),
		done:   make(chan struct{}),
	}

	c.streamSubs.Store(reqID, s)

	if err := c.send(&nodev1.ServerMessage{
		Id:     reqID,
		Method: method,
		Body:   bodyBytes,
	}); err != nil {
		c.streamSubs.Delete(reqID)
		return nil, err
	}

	// ctx 取消 → 主动 Close（best-effort 通知 node）
	go func() {
		select {
		case <-ctx.Done():
			// 仅在尚未结束时 Close
			s.mu.Lock()
			closed := s.closed
			s.mu.Unlock()
			if !closed {
				// 标记 ctx 错误，再 Close
				s.mu.Lock()
				if s.err == nil {
					s.err = ctx.Err()
				}
				s.mu.Unlock()
				s.Close()
			}
		case <-s.done: