zmkgC/internal/app/system/controller/device.go

// ==========================================================================
// GFast自动生成controller操作代码。
// 生成日期：2023-11-24 09:57:07
// 生成路径: internal/app/system/controller/device.go
// 生成人：gfast
// desc:安全帽设备（java）
// company:云南奇讯科技有限公司
// ==========================================================================

package controller

import (
	"context"
	"encoding/json"
	"github.com/gogf/gf/v2/database/gredis"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/net/ghttp"
	"github.com/gogf/gf/v2/os/glog"
	"github.com/gogf/gf/v2/os/gmutex"
	"github.com/tiger1103/gfast/v3/api/v1/system"
	"github.com/tiger1103/gfast/v3/internal/app/system/dao"
	"github.com/tiger1103/gfast/v3/internal/app/system/service"
	"github.com/tiger1103/gfast/v3/library/liberr"
	"log"
	"sync"
	"time"
)

type deviceController struct {
	BaseController
}

var Device = new(deviceController)

// 全局map
var clientMap = make(map[string]*wsClient)

const RedisHelmetKey = "safety_helmet:"

type wsClient struct {
	Socket *ghttp.WebSocket
}

type BusUserEntity struct {
	NickName  string ` json:"nickName" dc:"微信名称"`
	HeadIcon  string ` json:"headIcon" dc:"登录头像"`
	UserName  string ` json:"userName" dc:"人员姓名"`
	DevNum    string `json:"devNum" dc:"设备标识"`
	LatAndLon string ` json:"LatAndLon" dc:"纬度经度（格式：29.101,180.101）"`
}

// List 列表
func (c *deviceController) List(ctx context.Context, req *system.DeviceSearchReq) (res *system.DeviceSearchRes, err error) {
	res, err = service.Device().List(ctx, req)
	return
}

// AllList 返回当前项目下。设备和用户关联到的所有设备
func (c *deviceController) AllList(ctx context.Context, req *system.DeviceSearchAllReq) (res *system.DeviceSearchAllRes, err error) {
	res, err = service.Device().AllList(ctx, req)
	return
}

// HelmetListFunc 安全帽列表
func (c *deviceController) HelmetListFunc(ctx context.Context, req *system.HelmetListReq) (res *system.HelmetListRes, err error) {
	res, err = service.Device().HelmetListFunc(ctx, req)
	return
}

// Get 获取安全帽设备（java）
func (c *deviceController) Get(ctx context.Context, req *system.DeviceGetReq) (res *system.DeviceGetRes, err error) {
	res = new(system.DeviceGetRes)
	res, err = service.Device().GetByDevNum(ctx, req)
	return
}

// Add 添加安全帽设备（java）
func (c *deviceController) Add(ctx context.Context, req *system.DeviceAddReq) (res *system.DeviceAddRes, err error) {
	err = service.Device().Add(ctx, req)
	return
}

// RemoveRelationFunc 移除人员
func (c *deviceController) RemoveRelationFunc(ctx context.Context, req *system.RemoveRelationReq) (res *system.RemoveRelationRes, err error) {
	err = service.Device().RemoveRelationFunc(ctx, req)
	return
}

// Delete 删除安全帽设备（java）
func (c *deviceController) Delete(ctx context.Context, req *system.DeviceDeleteReq) (res *system.DeviceDeleteRes, err error) {
	err = service.Device().Delete(ctx, req.DevNums)
	return
}

// ScheduleTimeFunc 根据指定时间段去查询有足迹的日期
func (c *deviceController) ScheduleTimeFunc(ctx context.Context, req *system.ScheduleTimeReq) (res *system.ScheduleTimeRes, err error) {
	res = new(system.ScheduleTimeRes)
	res, err = service.Device().ScheduleTimeFunc(ctx, req)
	return
}

var (
	once        sync.Once
	subscribe   gredis.Conn
	subscribeMu sync.Mutex
)

// 初始化订阅
func InitSubscription() {
	subscribeMu.Lock()
	defer subscribeMu.Unlock()
	if subscribe == nil {
		var err error
		subscribe, _, err = g.Redis("helmetRedis").Subscribe(context.Background(), RedisHelmetKey)
		if err != nil {
			log.Fatal("Failed to subscribe to Redis:", err)
		}
	}
}

func (c *deviceController) WebsocketFunc(ctx context.Context, req *system.WebsocketReq) (res *system.WebsocketRes, err error) {
	err = g.Try(ctx, func(ctx context.Context) {
		once.Do(InitSubscription)
		reqWs := ghttp.RequestFromCtx(ctx)
		ws, err := reqWs.WebSocket()
		if err != nil {
			glog.Error(ctx, err)
			reqWs.Exit()
		}
		var strData = ""
		var i = 0
		mutex := gmutex.New()
		mutex.Lock()
		defer mutex.Unlock()
		for {
			if strData != "" || i == 0 {
				count, err := dao.BusConstructionUser.Ctx(ctx).Where("dev_num", strData).Where("project_id", req.ProjectId).Count()
				if err != nil {
					liberr.ErrIsNil(ctx, err, "WebSocket服务出现问题！")
					return
				}
				if count > 0 || i == 0 {
					i = 1
					var busUser []*BusUserEntity
					err = dao.BusConstructionUser.Ctx(ctx).Where("project_id", req.ProjectId).Where("dev_num <> ''").Fields("head_icon,nick_name,user_name,dev_num").Scan(&busUser)
					if err != nil {
						liberr.ErrIsNil(ctx, err, "WebSocket服务出现问题！")
						return
					}
					for ii := range busUser {
						get, _ := g.Redis("helmetRedis").Get(ctx, RedisHelmetKey+busUser[ii].DevNum)
						if get != nil {
							busUser[ii].LatAndLon = get.String()
						}
					}
					marshal, err := json.Marshal(busUser)
					if err != nil {
						liberr.ErrIsNil(ctx, err, "WebSocket服务出现问题！")
						return
					}
					if err = ws.WriteMessage(1, marshal); err != nil {

						return
					}
				}
			}
			_, _, err := ws.ReadMessage()
			if err != nil {
				return //断开就退出整个代码块
			}
			//第一次先发数据、然后如果前端还活着，那么就继续走（走到redis阻塞，redis如果有消息发布就继续下一次循环，），如果死了就退出、
			msg, err := subscribe.ReceiveMessage(ctx)
			if err != nil {
				log.Println("Error receiving message:", err)
				time.Sleep(time.Second)
				continue
			}
			strData = msg.Payload
		}
	})
	return
}