yumusb
35becefe1e
Add support for bundling metrics and enable WebSocket compression. Introduces protocol.BundlePayload and MessageTypeBundle, and handles bundled messages on the server side (handleBundleMessage) by iterating items and forwarding each to metricService.HandleMetricData. On the agent side, add bundleWriter to collect outgoing MetricsPayloads, switch collectors to write into the bundle writer, and send a single bundle message when items exist. Also enable compression on the WebSocket upgrader and create a custom websocket.Dialer (with Proxy, HandshakeTimeout and EnableCompression). Minor logging/messages and import adjustments included.
57 lines
1.8 KiB
Go
57 lines
1.8 KiB
Go
package handler
|
|
|
|
import (
|
|
"github.com/dushixiang/pika/internal/service"
|
|
ws "github.com/dushixiang/pika/internal/websocket"
|
|
"github.com/gorilla/websocket"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
type AgentHandler struct {
|
|
logger *zap.Logger
|
|
agentService *service.AgentService
|
|
trafficService *service.TrafficService
|
|
metricService *service.MetricService
|
|
monitorSvc *service.MonitorService
|
|
tamperService *service.TamperService
|
|
ddnsService *service.DDNSService
|
|
sshLoginService *service.SSHLoginService
|
|
apiKeyService *service.ApiKeyService
|
|
propertyService *service.PropertyService
|
|
wsManager *ws.Manager
|
|
upgrader websocket.Upgrader
|
|
}
|
|
|
|
func NewAgentHandler(logger *zap.Logger, agentService *service.AgentService, trafficService *service.TrafficService,
|
|
metricService *service.MetricService, monitorService *service.MonitorService, tamperService *service.TamperService,
|
|
ddnsService *service.DDNSService, sshLoginService *service.SSHLoginService, apiKeyService *service.ApiKeyService,
|
|
propertyService *service.PropertyService, wsManager *ws.Manager) *AgentHandler {
|
|
|
|
h := &AgentHandler{
|
|
logger: logger,
|
|
agentService: agentService,
|
|
trafficService: trafficService,
|
|
metricService: metricService,
|
|
monitorSvc: monitorService,
|
|
tamperService: tamperService,
|
|
ddnsService: ddnsService,
|
|
sshLoginService: sshLoginService,
|
|
apiKeyService: apiKeyService,
|
|
propertyService: propertyService,
|
|
wsManager: wsManager,
|
|
}
|
|
|
|
// 初始化upgrader，需要在创建handler之后因为需要引用h.checkOrigin
|
|
h.upgrader = websocket.Upgrader{
|
|
ReadBufferSize: 1024 * 32,
|
|
WriteBufferSize: 1024 * 32,
|
|
EnableCompression: true,
|
|
}
|
|
|
|
// 设置WebSocket消息处理器
|
|
wsManager.SetMessageHandler(h.handleWebSocketMessage)
|
|
wsManager.SetPongHandler(h.handleWebSocketPong)
|
|
|
|
return h
|
|
}
|