| 157 | } |
| 158 | |
| 159 | func (s *session) dispatch(msg *nodev1.ServerMessage) { |
| 160 | // 1) cancel 帧：取消 inflight。 |
| 161 | if cid := msg.GetCancelId(); cid != "" { |
| 162 | s.mu.Lock() |
| 163 | cancel, ok := s.inflight[cid] |
| 164 | s.mu.Unlock() |
| 165 | if ok { |
| 166 | cancel() |
| 167 | } |
| 168 | return |
| 169 | } |
| 170 | |
| 171 | method := msg.GetMethod() |
| 172 | |
| 173 | // 2) ack 帧（method=="ack"）：唤醒等待。 |
| 174 | if method == "ack" { |
| 175 | s.handleAck(msg.GetBody()) |
| 176 | return |
| 177 | } |
| 178 | |
| 179 | if method == "" { |
| 180 | // 既无 method 也无 cancel_id，忽略。 |
| 181 | return |
| 182 | } |
| 183 | |
| 184 | // 3) 普通 method 调用：开 goroutine 处理，注册 cancel。 |
| 185 | reqID := msg.GetId() |
| 186 | reqCtx, reqCancel := context.WithCancel(s.ctx) |
| 187 | if reqID != "" { |
| 188 | reqCtx = context.WithValue(reqCtx, reqIDCtxKey{}, reqID) |
| 189 | s.mu.Lock() |
| 190 | s.inflight[reqID] = reqCancel |
| 191 | s.mu.Unlock() |
| 192 | } |
| 193 | |
| 194 | body := msg.GetBody() |
| 195 | s.wg.Add(1) |
| 196 | go func() { |
| 197 | defer s.wg.Done() |
| 198 | defer reqCancel() |
| 199 | defer func() { |
| 200 | if reqID != "" { |
| 201 | s.mu.Lock() |
| 202 | delete(s.inflight, reqID) |
| 203 | s.mu.Unlock() |
| 204 | } |
| 205 | }() |
| 206 | |
| 207 | respBody, err := s.cfg.Dispatcher.Handle(reqCtx, method, body) |
| 208 | if reqID == "" { |
| 209 | // 一次性、无需响应的调用。 |
| 210 | return |
| 211 | } |
| 212 | resp := &nodev1.NodeMessage{Id: reqID} |
| 213 | if err != nil { |
| 214 | resp.Ok = false |
| 215 | resp.Error = err.Error() |
| 216 | } else { |