pig/pig-farm-controller
1
0
Fork 0
Code Issues 11 Pull Requests Actions Packages Projects Releases Wiki Activity

合并websocket逻辑

Browse Source
This commit is contained in:
huang
2025-09-09 15:27:53 +08:00
parent adb9a12a9d
commit 4c9f059af2
5 changed files with 27 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
internal/api/api.go
View File

@@ -53,9 +53,6 @@ type API struct {
// websocketManager WebSocket管理器 // websocketManager WebSocket管理器
websocketManager *websocket.Manager websocketManager *websocket.Manager
// websocketService WebSocket服务
websocketService *service.WebSocketService
// heartbeatService 心跳服务 // heartbeatService 心跳服务
heartbeatService *service.HeartbeatService heartbeatService *service.HeartbeatService
@@ -68,7 +65,7 @@ type API struct {
// NewAPI 创建并返回一个新的API实例 // NewAPI 创建并返回一个新的API实例
// 初始化Gin引擎和相关配置 // 初始化Gin引擎和相关配置
func NewAPI(cfg *config.Config, userRepo repository.UserRepo, operationHistoryRepo repository.OperationHistoryRepo, deviceControlRepo repository.DeviceControlRepo, deviceRepo repository.DeviceRepo, websocketService *service.WebSocketService, heartbeatService *service.HeartbeatService, deviceStatusPool *service.DeviceStatusPool) *API { func NewAPI(cfg *config.Config, userRepo repository.UserRepo, operationHistoryRepo repository.OperationHistoryRepo, deviceControlRepo repository.DeviceControlRepo, deviceRepo repository.DeviceRepo, websocketManager *websocket.Manager, heartbeatService *service.HeartbeatService, deviceStatusPool *service.DeviceStatusPool) *API {
// 设置Gin为发布模式 // 设置Gin为发布模式
gin.SetMode(gin.DebugMode) gin.SetMode(gin.DebugMode)
@@ -99,13 +96,10 @@ func NewAPI(cfg *config.Config, userRepo repository.UserRepo, operationHistoryRe
operationController := operation.NewController(operationHistoryRepo) operationController := operation.NewController(operationHistoryRepo)
// 创建设备控制控制器 // 创建设备控制控制器
deviceController := device.NewController(deviceControlRepo, deviceRepo, websocketService, heartbeatService, deviceStatusPool) deviceController := device.NewController(deviceControlRepo, deviceRepo, websocketManager, heartbeatService, deviceStatusPool)
// 创建WebSocket管理器
websocketManager := websocket.NewManager(websocketService, deviceRepo)
// 创建远程控制控制器 // 创建远程控制控制器
remoteController := remote.NewController(websocketService) remoteController := remote.NewController(websocketManager)
// 创建鉴权中间件 // 创建鉴权中间件
authMiddleware := middleware.NewAuthMiddleware(userRepo) authMiddleware := middleware.NewAuthMiddleware(userRepo)
@@ -119,7 +113,6 @@ func NewAPI(cfg *config.Config, userRepo repository.UserRepo, operationHistoryRe
remoteController: remoteController, remoteController: remoteController,
authMiddleware: authMiddleware, authMiddleware: authMiddleware,
websocketManager: websocketManager, websocketManager: websocketManager,
websocketService: websocketService,
heartbeatService: heartbeatService, heartbeatService: heartbeatService,
deviceStatusPool: deviceStatusPool, deviceStatusPool: deviceStatusPool,
logger: logs.NewLogger(), logger: logs.NewLogger(),

9
internal/controller/device/device.go
View File

@@ -12,6 +12,7 @@ import (
"git.huangwc.com/pig/pig-farm-controller/internal/model" "git.huangwc.com/pig/pig-farm-controller/internal/model"
"git.huangwc.com/pig/pig-farm-controller/internal/service" "git.huangwc.com/pig/pig-farm-controller/internal/service"
"git.huangwc.com/pig/pig-farm-controller/internal/storage/repository" "git.huangwc.com/pig/pig-farm-controller/internal/storage/repository"
"git.huangwc.com/pig/pig-farm-controller/internal/websocket"
"github.com/gin-gonic/gin" "github.com/gin-gonic/gin"
) )
@@ -121,18 +122,18 @@ func (req *DeviceRequest) BindAndValidate(data []byte) error {
type Controller struct { type Controller struct {
deviceControlRepo repository.DeviceControlRepo deviceControlRepo repository.DeviceControlRepo
deviceRepo repository.DeviceRepo deviceRepo repository.DeviceRepo
websocketService *service.WebSocketService websocketManager *websocket.Manager
heartbeatService *service.HeartbeatService heartbeatService *service.HeartbeatService
deviceStatusPool *service.DeviceStatusPool deviceStatusPool *service.DeviceStatusPool
logger *logs.Logger logger *logs.Logger
} }
// NewController 创建设备控制控制器实例 // NewController 创建设备控制控制器实例
func NewController(deviceControlRepo repository.DeviceControlRepo, deviceRepo repository.DeviceRepo, websocketService *service.WebSocketService, heartbeatService *service.HeartbeatService, deviceStatusPool *service.DeviceStatusPool) *Controller { func NewController(deviceControlRepo repository.DeviceControlRepo, deviceRepo repository.DeviceRepo, websocketManager *websocket.Manager, heartbeatService *service.HeartbeatService, deviceStatusPool *service.DeviceStatusPool) *Controller {
return &Controller{ return &Controller{
deviceControlRepo: deviceControlRepo, deviceControlRepo: deviceControlRepo,
deviceRepo: deviceRepo, deviceRepo: deviceRepo,
websocketService: websocketService, websocketManager: websocketManager,
heartbeatService: heartbeatService, heartbeatService: heartbeatService,
deviceStatusPool: deviceStatusPool, deviceStatusPool: deviceStatusPool,
logger: logs.NewLogger(), logger: logs.NewLogger(),
@@ -367,7 +368,7 @@ func (c *Controller) Switch(ctx *gin.Context) {
} }
// 发送指令并等待响应 // 发送指令并等待响应
response, err := c.websocketService.SendCommandAndWait("relay-001", "control_device", controlData, 0) response, err := c.websocketManager.SendCommandAndWait("relay-001", "control_device", controlData, 0)
if err != nil { if err != nil {
c.logger.Error("通过WebSocket发送设备控制指令失败: " + err.Error()) c.logger.Error("通过WebSocket发送设备控制指令失败: " + err.Error())
controller.SendErrorResponse(ctx, controller.InternalServerErrorCode, "设备控制失败: "+err.Error()) controller.SendErrorResponse(ctx, controller.InternalServerErrorCode, "设备控制失败: "+err.Error())

12
internal/controller/remote/remote.go
View File

@@ -5,20 +5,20 @@ package remote
import ( import (
"git.huangwc.com/pig/pig-farm-controller/internal/controller" "git.huangwc.com/pig/pig-farm-controller/internal/controller"
"git.huangwc.com/pig/pig-farm-controller/internal/logs" "git.huangwc.com/pig/pig-farm-controller/internal/logs"
"git.huangwc.com/pig/pig-farm-controller/internal/service" "git.huangwc.com/pig/pig-farm-controller/internal/websocket"
"github.com/gin-gonic/gin" "github.com/gin-gonic/gin"
) )
// Controller 远程控制控制器 // Controller 远程控制控制器
type Controller struct { type Controller struct {
websocketService *service.WebSocketService websocketManager *websocket.Manager
logger *logs.Logger logger *logs.Logger
} }
// NewController 创建远程控制控制器实例 // NewController 创建远程控制控制器实例
func NewController(websocketService *service.WebSocketService) *Controller { func NewController(websocketManager *websocket.Manager) *Controller {
return &Controller{ return &Controller{
websocketService: websocketService, websocketManager: websocketManager,
logger: logs.NewLogger(), logger: logs.NewLogger(),
} }
} }
@@ -69,7 +69,7 @@ func (c *Controller) SendCommand(ctx *gin.Context) {
} }
// 发送指令并等待响应 // 发送指令并等待响应
response, err := c.websocketService.SendCommandAndWait(req.DeviceID, req.Command, commandData, 0) response, err := c.websocketManager.SendCommandAndWait(req.DeviceID, req.Command, commandData, 0)
if err != nil { if err != nil {
c.logger.Error("发送指令失败: " + err.Error()) c.logger.Error("发送指令失败: " + err.Error())
controller.SendErrorResponse(ctx, controller.InternalServerErrorCode, "发送指令失败: "+err.Error()) controller.SendErrorResponse(ctx, controller.InternalServerErrorCode, "发送指令失败: "+err.Error())
@@ -100,7 +100,7 @@ type ListConnectedDevicesResponseData struct {
// @Router /api/v1/remote/devices [get] // @Router /api/v1/remote/devices [get]
func (c *Controller) ListConnectedDevices(ctx *gin.Context) { func (c *Controller) ListConnectedDevices(ctx *gin.Context) {
// 获取已连接的设备列表 // 获取已连接的设备列表
devices := c.websocketService.GetConnectedDevices() devices := c.websocketManager.GetConnectedDevices()
data := ListConnectedDevicesResponseData{ data := ListConnectedDevicesResponseData{
Devices: devices, Devices: devices,

15
internal/core/application.go
View File

@@ -13,6 +13,7 @@ import (
"git.huangwc.com/pig/pig-farm-controller/internal/storage/db" "git.huangwc.com/pig/pig-farm-controller/internal/storage/db"
"git.huangwc.com/pig/pig-farm-controller/internal/storage/repository" "git.huangwc.com/pig/pig-farm-controller/internal/storage/repository"
"git.huangwc.com/pig/pig-farm-controller/internal/task" "git.huangwc.com/pig/pig-farm-controller/internal/task"
"git.huangwc.com/pig/pig-farm-controller/internal/websocket"
) )
// Application 代表核心应用结构 // Application 代表核心应用结构
@@ -39,8 +40,8 @@ type Application struct {
// DeviceRepo 设备仓库实例 // DeviceRepo 设备仓库实例
DeviceRepo repository.DeviceRepo DeviceRepo repository.DeviceRepo
// WebSocketService WebSocket服务实例 // WebsocketManager WebSocket管理器
WebSocketService *service.WebSocketService WebsocketManager *websocket.Manager
// DeviceStatusPool 设备状态池实例 // DeviceStatusPool 设备状态池实例
DeviceStatusPool *service.DeviceStatusPool DeviceStatusPool *service.DeviceStatusPool
@@ -100,17 +101,15 @@ func (app *Application) Start() error {
app.DeviceStatusPool = service.NewDeviceStatusPool() app.DeviceStatusPool = service.NewDeviceStatusPool()
// 初始化WebSocket服务 // 初始化WebSocket服务
app.WebSocketService = service.NewWebSocketService(app.DeviceRepo) app.WebsocketManager = websocket.NewManager(app.DeviceRepo)
// 设置设备状态池
app.WebSocketService.SetDeviceStatusPool(app.DeviceStatusPool)
// 设置WebSocket超时时间 // 设置WebSocket超时时间
app.WebSocketService.SetDefaultTimeout(app.Config.GetWebSocketTimeout()) app.WebsocketManager.SetDefaultTimeout(app.Config.GetWebSocketConfig().Timeout)
// 初始化心跳服务 // 初始化心跳服务
app.HeartbeatService = service.NewHeartbeatService(app.WebSocketService, app.DeviceStatusPool, app.DeviceRepo, app.Config) app.HeartbeatService = service.NewHeartbeatService(app.WebsocketManager, app.DeviceStatusPool, app.DeviceRepo, app.Config)
// 初始化API组件 // 初始化API组件
app.API = api.NewAPI(app.Config, app.UserRepo, app.OperationHistoryRepo, app.DeviceControlRepo, app.DeviceRepo, app.WebSocketService, app.HeartbeatService, app.DeviceStatusPool) app.API = api.NewAPI(app.Config, app.UserRepo, app.OperationHistoryRepo, app.DeviceControlRepo, app.DeviceRepo, app.WebsocketManager, app.HeartbeatService, app.DeviceStatusPool)
// 初始化任务执行器组件(使用5个工作协程) // 初始化任务执行器组件(使用5个工作协程)
app.TaskExecutor = task.NewExecutor(5) app.TaskExecutor = task.NewExecutor(5)

11
internal/service/heartbeat.go
View File

@@ -12,13 +12,14 @@ import (
"git.huangwc.com/pig/pig-farm-controller/internal/logs" "git.huangwc.com/pig/pig-farm-controller/internal/logs"
"git.huangwc.com/pig/pig-farm-controller/internal/model" "git.huangwc.com/pig/pig-farm-controller/internal/model"
"git.huangwc.com/pig/pig-farm-controller/internal/storage/repository" "git.huangwc.com/pig/pig-farm-controller/internal/storage/repository"
"git.huangwc.com/pig/pig-farm-controller/internal/websocket"
"github.com/panjf2000/ants/v2" "github.com/panjf2000/ants/v2"
) )
// HeartbeatService 心跳服务，负责管理设备的心跳检测 // HeartbeatService 心跳服务，负责管理设备的心跳检测
type HeartbeatService struct { type HeartbeatService struct {
// websocketService WebSocket服务 // websocketManager WebSocket管理器
websocketService *WebSocketService websocketManager *websocket.Manager
// deviceStatusPool 设备状态池 // deviceStatusPool 设备状态池
deviceStatusPool *DeviceStatusPool deviceStatusPool *DeviceStatusPool
@@ -52,7 +53,7 @@ type HeartbeatService struct {
} }
// NewHeartbeatService 创建心跳服务实例 // NewHeartbeatService 创建心跳服务实例
func NewHeartbeatService(websocketService *WebSocketService, deviceStatusPool *DeviceStatusPool, deviceRepo repository.DeviceRepo, config *config.Config) *HeartbeatService { func NewHeartbeatService(websocketManager *websocket.Manager, deviceStatusPool *DeviceStatusPool, deviceRepo repository.DeviceRepo, config *config.Config) *HeartbeatService {
interval := config.GetHeartbeatConfig().Interval interval := config.GetHeartbeatConfig().Interval
if interval <= 0 { if interval <= 0 {
@@ -65,7 +66,7 @@ func NewHeartbeatService(websocketService *WebSocketService, deviceStatusPool *D
} }
return &HeartbeatService{ return &HeartbeatService{
websocketService: websocketService, websocketManager: websocketManager,
deviceStatusPool: deviceStatusPool, deviceStatusPool: deviceStatusPool,
deviceRepo: deviceRepo, deviceRepo: deviceRepo,
logger: logs.NewLogger(), logger: logs.NewLogger(),
@@ -241,7 +242,7 @@ func (hs *HeartbeatService) handleHeartbeatWithStatus(deviceID string, tempStatu
} }
// 发送心跳包到设备 // 发送心跳包到设备
response, err := hs.websocketService.SendCommandAndWait(deviceID, "heartbeat", heartbeatData, 0) response, err := hs.websocketManager.SendCommandAndWait(deviceID, "heartbeat", heartbeatData, 0)
if err != nil { if err != nil {
hs.logger.Error(fmt.Sprintf("向设备 %s 发送心跳包失败: %v", deviceID, err)) hs.logger.Error(fmt.Sprintf("向设备 %s 发送心跳包失败: %v", deviceID, err))
// 更新设备状态为离线 // 更新设备状态为离线