Session 实现 NodeAgentServer：处理 node 的双向流。
(stream nodev1.NodeAgent_SessionServer)
| 197 | |
| 198 | // Session 实现 NodeAgentServer：处理 node 的双向流。 |
| 199 | func (h *Hub) Session(stream nodev1.NodeAgent_SessionServer) error { |
| 200 | ctx := stream.Context() |
| 201 | nodeID, err := h.peerExtractor(ctx) |
| 202 | if err != nil { |
| 203 | h.logger.Warn("nodehub: reject session, peer extract failed", "err", err) |
| 204 | return err |
| 205 | } |
| 206 | if nodeID == "" { |
| 207 | return errors.New("nodehub: empty nodeID") |
| 208 | } |
| 209 | |
| 210 | c := &conn{ |
| 211 | hub: h, |
| 212 | nodeID: nodeID, |
| 213 | stream: stream, |
| 214 | closed: make(chan struct{}), |
| 215 | } |
| 216 | |
| 217 | // 注册（如有旧连接，关闭旧的） |
| 218 | h.mu.Lock() |
| 219 | if old, ok := h.conns[nodeID]; ok { |
| 220 | h.logger.Info("nodehub: node reconnected, closing old session", "node_id", nodeID) |
| 221 | old.close() |
| 222 | h.metrics.reconnectTotal.Add(1) |
| 223 | // 旧连接占用的 onlineGauge 会在其 Session goroutine 退出时减一； |
| 224 | // 这里不重复增减，新连接下面统一 +1。 |
| 225 | h.metrics.onlineGauge.Add(-1) |
| 226 | } |
| 227 | h.conns[nodeID] = c |
| 228 | h.mu.Unlock() |
| 229 | h.metrics.onlineGauge.Add(1) |
| 230 | h.metrics.markSeen(nodeID) |
| 231 | |
| 232 | if h.onNodeConnected != nil { |
| 233 | peerIP := "" |
| 234 | if p, ok := peer.FromContext(ctx); ok && p.Addr != nil { |
| 235 | if host, _, err := net.SplitHostPort(p.Addr.String()); err == nil { |
| 236 | peerIP = host |
| 237 | } |
| 238 | } |
| 239 | go h.onNodeConnected(nodeID, peerIP) |
| 240 | } |
| 241 | |
| 242 | defer func() { |
| 243 | h.mu.Lock() |
| 244 | // 仅当当前注册的还是自己时才删除（避免被新连接挤掉时误删） |
| 245 | stillOurs := h.conns[nodeID] == c |
| 246 | if stillOurs { |
| 247 | delete(h.conns, nodeID) |
| 248 | } |
| 249 | h.mu.Unlock() |
| 250 | if stillOurs { |
| 251 | h.metrics.onlineGauge.Add(-1) |
| 252 | h.metrics.forgetNode(nodeID) |
| 253 | } |
| 254 | c.close() |
| 255 | }() |
| 256 |