| 337 | } |
| 338 | |
| 339 | func (h *Hub) handleEvent(c *conn, msg *nodev1.NodeMessage) { |
| 340 | body := msg.GetBody() |
| 341 | switch msg.GetEvent() { |
| 342 | case "hello": |
| 343 | h.pushHandler.OnHello(c.nodeID, body) |
| 344 | case "usage_push": |
| 345 | seq := msg.GetSeq() |
| 346 | h.metrics.pushUsageTotal.Add(1) |
| 347 | if err := h.pushHandler.OnUsagePush(c.nodeID, seq, body); err != nil { |
| 348 | h.logger.Warn("nodehub: usage_push handler failed; not acking", |
| 349 | "node_id", c.nodeID, "seq", seq, "err", err) |
| 350 | return |
| 351 | } |
| 352 | ackBody := []byte(fmt.Sprintf(`{"seq":%d}`, seq)) |
| 353 | if err := c.send(&nodev1.ServerMessage{ |
| 354 | Method: "ack", |
| 355 | Body: ackBody, |
| 356 | }); err != nil { |
| 357 | h.logger.Warn("nodehub: send usage ack failed", |
| 358 | "node_id", c.nodeID, "seq", seq, "err", err) |
| 359 | return |
| 360 | } |
| 361 | h.metrics.pushUsageAckTotal.Add(1) |
| 362 | case "log": |
| 363 | h.pushHandler.OnLog(c.nodeID, msg.GetId(), body) |
| 364 | case "traceroute_hop": |
| 365 | h.pushHandler.OnTracerouteHop(c.nodeID, msg.GetId(), body) |
| 366 | default: |
| 367 | h.logger.Debug("nodehub: unknown event", "node_id", c.nodeID, "event", msg.GetEvent()) |
| 368 | } |
| 369 | } |
| 370 | |
| 371 | func (c *conn) send(msg *nodev1.ServerMessage) error { |
| 372 | c.sendMu.Lock() |