websocket 三 hub 去重：抽 runWritePump 共享写泵 + 合并 agent 广播函数为 broadcastAgent

Result: {"status":"keep","total_issues":8,"eslint_errors":0,"eslint_problems":0,"eslint_warnings":0,"golint_canonicalheader":0,"golint_errname":0,"golint_errorlint":1,"golint_exhaustive":0,"golint_forcetypeassert":0,"golint_gosec":0,"golint_intrange":0,"golint_modernize":3,"golint_nilnil":3,"golint_perfsprint":0,"golint_prealloc":0,"golint_recvcheck":1,"golint_test_testifylint":0,"golint_test_thelper":0,"golint_test_total":0,"golint_test_usetesting":0,"golint_total":8,"golint_usestdlibvars":0,"golint_vetx_total":0,"golint_wastedassign":0,"measure_s":77,"tsc_errors":0,"vitest_failed":0,"vitest_total":126}

internal/apps/openflare/websocket/agent_hub.go

@@ -132,32 +132,19 @@ func SendAgentWAFIPGroups(nodeID string, payload any) bool {
 
 // BroadcastWAFIPGroups pushes changed WAF IP groups to all connected agents.
 func BroadcastWAFIPGroups(payload any) int {
-	if payload == nil {
-		return 0
-	}
-	message := Message{Type: agentMessageTypeWAFIPGroups, Payload: payload}
-	defaultAgentHub.mu.RLock()
-	clients := make([]*agentClient, 0, len(defaultAgentHub.clients))
-	for _, client := range defaultAgentHub.clients {
-		clients = append(clients, client)
-	}
-	defaultAgentHub.mu.RUnlock()
-
-	success := 0
-	for _, client := range clients {
-		if client.enqueue(message) {
-			success++
-		}
-	}
-	return success
+	return broadcastAgent(agentMessageTypeWAFIPGroups, payload)
 }
 
 // BroadcastActiveConfig pushes active config metadata to all connected agents.
 func BroadcastActiveConfig(payload any) int {
+	return broadcastAgent(agentMessageTypeActiveConfig, payload)
+}
+
+func broadcastAgent(messageType string, payload any) int {
 	if payload == nil {
 		return 0
 	}
-	message := Message{Type: agentMessageTypeActiveConfig, Payload: payload}
+	message := Message{Type: messageType, Payload: payload}
 	defaultAgentHub.mu.RLock()
 	clients := make([]*agentClient, 0, len(defaultAgentHub.clients))
 	for _, client := range defaultAgentHub.clients {
@@ -233,31 +220,8 @@ func agentWSReadTimeout() time.Duration {
 }
 
 func (c *agentClient) writePump() {
-	ticker := time.NewTicker(wsPingInterval)
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-c.done:
-			return
-		case message := <-c.send:
-			_ = c.conn.SetWriteDeadline(time.Now().Add(wsWriteDeadline))
-			if err := c.conn.WriteJSON(message); err != nil {
-				slog.Debug("agent ws write failed", "node_id", c.nodeID, "error", err)
-				c.close()
-				return
-			}
-		case <-ticker.C:
-			select {
-			case <-c.done:
-				return
-			case c.send <- Message{Type: messageTypePing}:
-			default:
-			}
-		}
-	}
+	runWritePump(c.nodeID, c.conn, c.done, c.send, c.close, "agent ws")
 }
-
 func (c *agentClient) enqueue(message Message) bool {
 	select {
 	case <-c.done:

internal/apps/openflare/websocket/flared_hub.go

@@ -6,7 +6,6 @@ package websocket
 import (
 	"log/slog"
 	"sync"
-	"time"
 
 	"github.com/gin-gonic/gin"
 	"github.com/gorilla/websocket"
@@ -139,27 +138,5 @@ func (c *flaredClient) readPump() {
 }
 
 func (c *flaredClient) writePump() {
-	ticker := time.NewTicker(wsPingInterval)
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-c.done:
-			return
-		case message := <-c.send:
-			_ = c.conn.SetWriteDeadline(time.Now().Add(wsWriteDeadline))
-			if err := c.conn.WriteJSON(message); err != nil {
-				slog.Debug("flared ws write failed", "node_id", c.nodeID, "error", err)
-				c.close()
-				return
-			}
-		case <-ticker.C:
-			select {
-			case <-c.done:
-				return
-			case c.send <- Message{Type: messageTypePing}:
-			default:
-			}
-		}
-	}
+	runWritePump(c.nodeID, c.conn, c.done, c.send, c.close, "flared ws")
 }

internal/apps/openflare/websocket/relay_hub.go

@@ -6,7 +6,6 @@ package websocket
 import (
 	"log/slog"
 	"sync"
-	"time"
 
 	"github.com/gin-gonic/gin"
 	"github.com/gorilla/websocket"
@@ -120,27 +119,5 @@ func (c *relayClient) readPump() {
 }
 
 func (c *relayClient) writePump() {
-	ticker := time.NewTicker(wsPingInterval)
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-c.done:
-			return
-		case message := <-c.send:
-			_ = c.conn.SetWriteDeadline(time.Now().Add(wsWriteDeadline))
-			if err := c.conn.WriteJSON(message); err != nil {
-				slog.Debug("relay ws write failed", "node_id", c.nodeID, "error", err)
-				c.close()
-				return
-			}
-		case <-ticker.C:
-			select {
-			case <-c.done:
-				return
-			case c.send <- Message{Type: messageTypePing}:
-			default:
-			}
-		}
-	}
+	runWritePump(c.nodeID, c.conn, c.done, c.send, c.close, "relay ws")
 }

internal/apps/openflare/websocket/write_pump.go

@@ -0,0 +1,47 @@
+// Copyright 2026 Arctel.net
+// SPDX-License-Identifier: Apache-2.0
+
+package websocket
+
+import (
+	"log/slog"
+	"time"
+
+	"github.com/gorilla/websocket"
+)
+
+// runWritePump drains send onto conn until done is closed, emitting
+// JSON pings at wsPingInterval. Shared by agent/relay/flared clients;
+// closeFn must be idempotent.
+func runWritePump(
+	nodeID string,
+	conn *websocket.Conn,
+	done <-chan struct{},
+	send chan Message,
+	closeFn func(),
+	logLabel string,
+) {
+	ticker := time.NewTicker(wsPingInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-done:
+			return
+		case message := <-send:
+			_ = conn.SetWriteDeadline(time.Now().Add(wsWriteDeadline))
+			if err := conn.WriteJSON(message); err != nil {
+				slog.Debug(logLabel+" write failed", "node_id", nodeID, "error", err)
+				closeFn()
+				return
+			}
+		case <-ticker.C:
+			select {
+			case <-done:
+				return
+			case send <- Message{Type: messageTypePing}:
+			default:
+			}
+		}
+	}
+}