Compare commits
No commits in common. "d6b1788ae5b45543c0bbadf57ea79e6d79b7da5f" and "b9025cf5d1d3ec0857ddf7af811bc4b8490290b2" have entirely different histories.
d6b1788ae5
...
b9025cf5d1
26
dao/cid.go
26
dao/cid.go
|
|
@ -10,7 +10,6 @@ type CID struct {
|
|||
Auth_id int `gorm:"column:auth_id"`
|
||||
Name string `gorm:"column:name"`
|
||||
Url string `gorm:"column:url"`
|
||||
Time int `gorm:"column:time"` // 定时任务，单位秒，大于0表示定时任务
|
||||
Script string `gorm:"column:script"`
|
||||
Token string `gorm:"column:token"` // 用于外部回调
|
||||
}
|
||||
|
|
@ -25,8 +24,8 @@ type CIDRunLog struct {
|
|||
}
|
||||
|
||||
// CreateCID 创建持续集成、部署
|
||||
func CreateCID(name, url, script, token string, time, auth_id int) uint {
|
||||
cid := CID{Name: name, Url: url, Script: script, Token: token, Auth_id: auth_id, Time: time}
|
||||
func CreateCID(name, url, script, token string, auth_id int) uint {
|
||||
cid := CID{Name: name, Url: url, Script: script, Token: token, Auth_id: auth_id}
|
||||
result := DB.Create(&cid)
|
||||
if result.Error != nil {
|
||||
return 0
|
||||
|
|
@ -58,7 +57,7 @@ func FindCIDByAuthID(auth_id int) []CID {
|
|||
}
|
||||
|
||||
// UpdateCIDByID 更新持续集成、部署
|
||||
func UpdateCIDByID(id, auth_id, time int, name, url, script, token string) bool {
|
||||
func UpdateCIDByID(id, auth_id int, name, url, script, token string) bool {
|
||||
pd := FindCIDByID(id, auth_id)
|
||||
if pd.ID == 0 {
|
||||
return false
|
||||
|
|
@ -67,7 +66,7 @@ func UpdateCIDByID(id, auth_id, time int, name, url, script, token string) bool
|
|||
if token == "" {
|
||||
token = pd.Token
|
||||
}
|
||||
result := DB.Model(&CID{}).Where("id = ? and auth_id = ?", id, auth_id).Updates(CID{Name: name, Url: url, Script: script, Token: token, Time: time})
|
||||
result := DB.Model(&CID{}).Where("id = ? and auth_id = ?", id, auth_id).Updates(CID{Name: name, Url: url, Script: script, Token: token})
|
||||
if result.Error != nil {
|
||||
return false
|
||||
}
|
||||
|
|
@ -93,12 +92,12 @@ func FindRunLogByAuthID(auth_id int) []CIDRunLog {
|
|||
|
||||
func FindRunLogByID(auth_id, cid_id int) []CIDRunLog {
|
||||
var cidRunLog []CIDRunLog
|
||||
DB.Where("cid_id = ? and auth_id = ?", cid_id, auth_id).Order("created_at desc").Limit(30).Find(&cidRunLog)
|
||||
DB.Where("cid_id = ? and auth_id = ?", cid_id, auth_id).Order("created_at desc").Find(&cidRunLog)
|
||||
return cidRunLog
|
||||
}
|
||||
func FindRunLogByCIDLogID(id, auth_id int) []CIDRunLog {
|
||||
var cidRunLogs []CIDRunLog
|
||||
DB.Where("id = ? and auth_id = ?", id, auth_id).Order("created_at desc").Limit(30).Find(&cidRunLogs)
|
||||
DB.Where("id = ? and auth_id = ?", id, auth_id).Order("created_at desc").Find(&cidRunLogs)
|
||||
return cidRunLogs
|
||||
}
|
||||
|
||||
|
|
@ -107,16 +106,3 @@ func FindCIDByIDAndToken(id int, token string) CID {
|
|||
DB.Where("id = ? and token = ?", id, token).First(&cid)
|
||||
return cid
|
||||
}
|
||||
|
||||
func FindCIDByTime() []CID {
|
||||
var cids []CID
|
||||
DB.Where("time > 0").Find(&cids)
|
||||
return cids
|
||||
}
|
||||
|
||||
// FindCIDByID 查找持续集成、部署
|
||||
func FindCIDByCID(id uint) CID {
|
||||
var cid CID
|
||||
DB.Debug().Where("id = ? ", id).First(&cid)
|
||||
return cid
|
||||
}
|
||||
|
|
|
|||
1
go.mod
1
go.mod
|
|
@ -8,7 +8,6 @@ require (
|
|||
github.com/golang-jwt/jwt v3.2.2+incompatible
|
||||
github.com/google/uuid v1.6.0
|
||||
github.com/gorilla/websocket v1.5.3
|
||||
github.com/robfig/cron/v3 v3.0.1
|
||||
gorm.io/driver/mysql v1.5.6
|
||||
gorm.io/driver/postgres v1.5.9
|
||||
gorm.io/gorm v1.25.10
|
||||
|
|
|
|||
2
go.sum
2
go.sum
|
|
@ -86,8 +86,6 @@ github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6
|
|||
github.com/pelletier/go-toml/v2 v2.2.2/go.mod h1:1t835xjRzz80PqgE6HHgN2JOsmgYu/h4qDAS4n929Rs=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
|
||||
github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
|
||||
github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8=
|
||||
github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4=
|
||||
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||
|
|
|
|||
153
handler/cid.go
153
handler/cid.go
|
|
@ -2,7 +2,6 @@ package handler
|
|||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"github.com/gin-gonic/gin"
|
||||
"os/exec"
|
||||
|
|
@ -10,13 +9,11 @@ import (
|
|||
"strings"
|
||||
"videoplayer/dao"
|
||||
"videoplayer/proto"
|
||||
"videoplayer/worker"
|
||||
)
|
||||
|
||||
type CIDCreateReq struct {
|
||||
Name string `json:"name" form:"name"`
|
||||
Url string `json:"url" form:"url"`
|
||||
Time int `json:"time" form:"time"` // 定时任务，单位秒，大于0表示定时任务
|
||||
Script string `json:"script" form:"script"`
|
||||
}
|
||||
|
||||
|
|
@ -36,14 +33,10 @@ type CIDUpdateReq struct {
|
|||
ID int `json:"id" form:"id"`
|
||||
Name string `json:"name" form:"name"`
|
||||
Url string `json:"url" form:"url"`
|
||||
Time int `json:"time" form:"time"` // 定时任务，单位秒，大于0表示定时任务
|
||||
Script string `json:"script" form:"script"`
|
||||
Token string `json:"cidtoken" form:"cidtoken"`
|
||||
}
|
||||
|
||||
// 全局变量，记录是否进行cron定时任务的刷新
|
||||
var cron_count int
|
||||
|
||||
func SetUpCIDGroup(router *gin.Engine) {
|
||||
cidGroup := router.Group("/cid") //持续集成、部署
|
||||
cidGroup.POST("/create", CreateCID)
|
||||
|
|
@ -89,11 +82,9 @@ func CreateCID(c *gin.Context) {
|
|||
id, _ := c.Get("id")
|
||||
authID := int(id.(float64))
|
||||
token, _ := generateRandomHexString(32)
|
||||
res := dao.CreateCID(req.Name, req.Url, req.Script, token, req.Time, authID)
|
||||
|
||||
res := dao.CreateCID(req.Name, req.Url, req.Script, token, authID)
|
||||
if res != 0 {
|
||||
if req.Time > 0 {
|
||||
updateCronRedisTime(int(res), req.Time)
|
||||
}
|
||||
c.JSON(200, gin.H{"code": proto.SuccessCode, "message": "success", "data": res})
|
||||
} else {
|
||||
c.JSON(200, gin.H{"error": "CreateCID failed", "code": proto.OperationFailed, "message": "failed"})
|
||||
|
|
@ -127,14 +118,11 @@ func UpdateCID(c *gin.Context) {
|
|||
// 获取用户ID
|
||||
id, _ := c.Get("id")
|
||||
authID := int(id.(float64))
|
||||
cid := dao.UpdateCIDByID(req.ID, authID, req.Time, req.Name, req.Url, req.Script, req.Token)
|
||||
cid := dao.UpdateCIDByID(req.ID, authID, req.Name, req.Url, req.Script, req.Token)
|
||||
if cid == false {
|
||||
c.JSON(200, gin.H{"error": "CID not found", "code": proto.OperationFailed, "message": "failed"})
|
||||
return
|
||||
} else {
|
||||
if req.Time > 0 {
|
||||
updateCronRedisTime(req.ID, req.Time)
|
||||
}
|
||||
c.JSON(200, gin.H{"code": proto.SuccessCode, "message": "success", "data": "success"})
|
||||
}
|
||||
} else {
|
||||
|
|
@ -234,138 +222,3 @@ echo "end"`
|
|||
//fmt.Println("bash content:", scriptContent)
|
||||
dao.CreateRunLog(id, authID, scriptContent, out.String(), err3_info) //添加执行日志
|
||||
}
|
||||
|
||||
// 定时任务处理逻辑
|
||||
func RunCron() {
|
||||
cron_count++
|
||||
if cron_count > 6 {
|
||||
updateCronFromDBToRedis()
|
||||
cron_count = 0
|
||||
}
|
||||
//从redis查看是否有定时任务
|
||||
//如果有定时任务，执行定时任务
|
||||
//如果没有定时任务，查找数据库是否有定时任务，如果有，则加入redis，如果没有，则不做任何操作
|
||||
key := "cron_cid_runs"
|
||||
res := worker.GetRedis(key)
|
||||
//fmt.Println("cid run cron res:", res)
|
||||
if res == "" {
|
||||
readCronFromDBToRedis(key)
|
||||
} else {
|
||||
var cid_workers []proto.CIDRUN
|
||||
err := json.Unmarshal([]byte(res), &cid_workers)
|
||||
if err != nil {
|
||||
fmt.Println("json unmarshal failed")
|
||||
}
|
||||
//fmt.Println("cid_workers:", cid_workers)
|
||||
for i, v := range cid_workers {
|
||||
//查找定时任务
|
||||
if v.Curr-10 <= 0 {
|
||||
cid := dao.FindCIDByCID(v.CID)
|
||||
if cid.ID != 0 {
|
||||
go RunShell("cron", cid.Url, cid.Script, int(cid.ID), cid.Auth_id)
|
||||
}
|
||||
cid_workers[i].Curr = v.Every
|
||||
} else {
|
||||
cid_workers[i].Curr = v.Curr - 10
|
||||
}
|
||||
}
|
||||
//将定时任务加入redis
|
||||
json_data, err := json.Marshal(cid_workers)
|
||||
if err != nil {
|
||||
fmt.Println("json marshal failed")
|
||||
}
|
||||
worker.SetRedis(key, string(json_data))
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
// 将数据库中的定时任务加入redis
|
||||
func readCronFromDBToRedis(key string) {
|
||||
cids := dao.FindCIDByTime()
|
||||
cid_workers := make([]proto.CIDRUN, 0)
|
||||
for _, v := range cids {
|
||||
cid_worker := proto.CIDRUN{CID: v.ID, Curr: v.Time, Every: v.Time}
|
||||
cid_workers = append(cid_workers, cid_worker)
|
||||
}
|
||||
if len(cid_workers) > 0 {
|
||||
//将定时任务加入redis
|
||||
json_data, err := json.Marshal(cid_workers)
|
||||
if err != nil {
|
||||
fmt.Println("json marshal failed")
|
||||
}
|
||||
worker.SetRedis(key, string(json_data))
|
||||
}
|
||||
}
|
||||
|
||||
// 从数据库更新定时任务到redis
|
||||
func updateCronFromDBToRedis() {
|
||||
key := "cron_cid_runs"
|
||||
cids := dao.FindCIDByTime()
|
||||
cid_maps := make(map[uint]int)
|
||||
|
||||
//将数据库中的定时任务加入map，key为cid，value为时间，便于后续查找
|
||||
for _, v := range cids {
|
||||
cid_maps[v.ID] = v.Time
|
||||
}
|
||||
res := worker.GetRedis(key)
|
||||
if res == "" {
|
||||
readCronFromDBToRedis(key)
|
||||
return
|
||||
}
|
||||
var cid_workers_redis []proto.CIDRUN
|
||||
err := json.Unmarshal([]byte(res), &cid_workers_redis)
|
||||
if err != nil {
|
||||
fmt.Println("json unmarshal failed")
|
||||
return
|
||||
}
|
||||
for i, v := range cid_workers_redis {
|
||||
if time, ok := cid_maps[v.CID]; ok {
|
||||
if v.Every != time {
|
||||
cid_workers_redis[i].Every = time
|
||||
cid_workers_redis[i].Curr = time
|
||||
}
|
||||
}
|
||||
}
|
||||
//将定时任务加入redis
|
||||
json_data, err := json.Marshal(cid_workers_redis)
|
||||
if err != nil {
|
||||
fmt.Println("json marshal failed")
|
||||
return
|
||||
}
|
||||
worker.SetRedis(key, string(json_data))
|
||||
}
|
||||
|
||||
// 查看指定定时任务是否存在，如果存在则更新时间，如果不存在则加入
|
||||
func updateCronRedisTime(id int, time int) {
|
||||
key := "cron_cid_runs"
|
||||
res := worker.GetRedis(key)
|
||||
if res == "" {
|
||||
readCronFromDBToRedis(key)
|
||||
return
|
||||
} else {
|
||||
var cid_workers []proto.CIDRUN
|
||||
err := json.Unmarshal([]byte(res), &cid_workers)
|
||||
if err != nil {
|
||||
fmt.Println("json unmarshal failed")
|
||||
}
|
||||
isContain := false
|
||||
for i, v := range cid_workers {
|
||||
if v.CID == uint(id) {
|
||||
//更新时间，不会继续原来的时间
|
||||
cid_workers[i].Curr = time
|
||||
cid_workers[i].Every = time
|
||||
isContain = true
|
||||
}
|
||||
}
|
||||
if isContain == false {
|
||||
cid_worker := proto.CIDRUN{CID: uint(id), Curr: time, Every: time}
|
||||
cid_workers = append(cid_workers, cid_worker)
|
||||
}
|
||||
//将定时任务加入redis
|
||||
json_data, err := json.Marshal(cid_workers)
|
||||
if err != nil {
|
||||
fmt.Println("json marshal failed")
|
||||
}
|
||||
worker.SetRedis(key, string(json_data))
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,11 +1,9 @@
|
|||
package handler
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/google/uuid"
|
||||
"github.com/gorilla/websocket"
|
||||
"net/http"
|
||||
"strconv"
|
||||
|
|
@ -208,96 +206,71 @@ func Restart(ip string) bool {
|
|||
return false
|
||||
}
|
||||
|
||||
// 发送实时视频流
|
||||
// 接收及发送消息
|
||||
func GetRealTimeImage(c *gin.Context) {
|
||||
id, _ := c.Get("id")
|
||||
id1 := int(id.(float64))
|
||||
device_id := c.Query("device_id")
|
||||
//字符串转int
|
||||
device_id_int, _ := strconv.Atoi(device_id)
|
||||
device := service.GetDevice(device_id_int, id1)
|
||||
if device.ID == 0 {
|
||||
c.JSON(http.StatusOK, gin.H{"code": proto.DataNotFound, "message": "device not found"})
|
||||
return
|
||||
}
|
||||
//建立连接
|
||||
// 升级HTTP连接为WebSocket连接
|
||||
ws, err := upgrader.Upgrade(c.Writer, c.Request, nil)
|
||||
clients[ws] = true
|
||||
if err != nil {
|
||||
fmt.Println("connect wss err:", err)
|
||||
// log.Println(err)
|
||||
fmt.Println(err)
|
||||
return
|
||||
}
|
||||
worker.SetRedisWithExpire(strconv.Itoa(int(device.ID))+"_is_play", "1", time.Minute*5)
|
||||
fmt.Println("device_id:", device_id_int, " has set is_play to 1")
|
||||
go subscribeAndHandleMessages(ws, device_id_int)
|
||||
}
|
||||
|
||||
func subscribeAndHandleMessages(ws *websocket.Conn, device_id int) {
|
||||
ctx := context.Background()
|
||||
pubsub := worker.RedisClient.Subscribe(ctx, strconv.Itoa(device_id)+"_frames_msgs")
|
||||
// 生成唯一连接 uuid
|
||||
con_id := uuid.New().String()
|
||||
online_conn_key := "device_" + strconv.Itoa(device_id) + "_online_conn_ids"
|
||||
// 加入设备在线连接集合
|
||||
worker.SetRedisSetAddWithExpire(online_conn_key, con_id, time.Minute*5)
|
||||
defer pubsub.Close()
|
||||
defer ws.Close()
|
||||
ch := pubsub.Channel()
|
||||
worker.SetRedisWithExpire(strconv.Itoa(int(device.ID))+"_is_play", "1", time.Minute*5) //设置播放状态
|
||||
// 接收客户端消息并发送到指定用户
|
||||
|
||||
go func(ws *websocket.Conn, device_id int) {
|
||||
|
||||
}(ws, device_id_int)
|
||||
var check_cnt int
|
||||
var ticker *time.Ticker
|
||||
|
||||
for {
|
||||
select {
|
||||
case msg, _ := <-ch:
|
||||
if v := clients[ws]; v == true {
|
||||
res2 := worker.PopRedisListLeft(device_id + "_frames")
|
||||
var res3 []byte
|
||||
var msgObj proto.Message
|
||||
if msg.Payload != "" {
|
||||
msgObj.Type = "img"
|
||||
msgObj.Msg = msg.Payload
|
||||
msgObj.From_user_id = -1
|
||||
res3, _ = json.Marshal(msgObj)
|
||||
var msg proto.Message
|
||||
if res2 != "" {
|
||||
//若有消息则发送消息
|
||||
msg.Type = "img"
|
||||
msg.Msg = res2
|
||||
msg.From_user_id = id1
|
||||
res3, _ = json.Marshal(msg)
|
||||
} else {
|
||||
//若无消息则发送心跳包
|
||||
if check_cnt < 5 {
|
||||
check_cnt++
|
||||
time.Sleep(time.Millisecond * 200)
|
||||
time.Sleep(time.Millisecond * 200) //设置延时200ms
|
||||
continue
|
||||
}
|
||||
check_cnt = 0
|
||||
msgObj.Type = "check"
|
||||
msgObj.Msg = "check"
|
||||
msgObj.From_user_id = -1
|
||||
res3, _ = json.Marshal(msgObj)
|
||||
msg.Type = "check"
|
||||
msg.Msg = "check"
|
||||
msg.From_user_id = -1
|
||||
res3, _ = json.Marshal(msg)
|
||||
}
|
||||
// fmt.Println("send message to client length:", len(res3))
|
||||
err2 := ws.WriteMessage(websocket.TextMessage, res3)
|
||||
worker.SetRedisWithExpire(strconv.Itoa(int(device.ID))+"_is_play", "1", time.Minute*5) //设置播放状态
|
||||
if err2 != nil {
|
||||
clientsMux.Lock()
|
||||
clients[ws] = false
|
||||
clientsMux.Unlock()
|
||||
fmt.Println("send message to client err:", err2)
|
||||
worker.SetRedisSetRemove(online_conn_key, con_id)
|
||||
goto end
|
||||
}
|
||||
default:
|
||||
if ticker == nil {
|
||||
ticker = time.NewTicker(time.Second)
|
||||
}
|
||||
select {
|
||||
case <-ticker.C:
|
||||
err := ws.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(time.Second))
|
||||
if err != nil {
|
||||
fmt.Println("Connection check failed:", err)
|
||||
worker.SetRedisSetRemove(online_conn_key, con_id)
|
||||
goto end
|
||||
}
|
||||
default:
|
||||
continue
|
||||
//设置ws关闭状态信息
|
||||
worker.SetRedisWithExpire(strconv.Itoa(int(device.ID))+"_is_play", "0", time.Minute*5) //设置播放状态
|
||||
break
|
||||
}
|
||||
time.Sleep(time.Millisecond * 200) //设置延时200ms
|
||||
}
|
||||
}
|
||||
|
||||
end:
|
||||
// 查看是否还有其他连接，没有则设置 is_play 为 0
|
||||
if worker.IsContainKey(online_conn_key) == false {
|
||||
worker.SetRedisWithExpire(strconv.Itoa(device_id)+"_is_play", "0", time.Minute*5)
|
||||
fmt.Println("device_id:", device_id, " has set is_play to 0")
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,7 +1,6 @@
|
|||
package handler
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
|
|
@ -75,7 +74,7 @@ func SetUpIMGroup(router *gin.Engine) {
|
|||
imGroup.POST("/get_group", GetGroups)
|
||||
imGroup.POST("/get_group_req_user", GetFriendRequest)
|
||||
imGroup.GET("/sse_msg", ServerSendMsg)
|
||||
imGroup.GET("/ws_v2", ServerSendMsgV3)
|
||||
imGroup.GET("/ws_v2", ServerSendMsgV2)
|
||||
imGroup.POST("/get_friend_list", GetFriendList) //获取好友列表,包括群聊
|
||||
//获取好友请求
|
||||
imGroup.POST("/get_friend_request", GetFriendRequest)
|
||||
|
|
@ -608,28 +607,6 @@ func ServerSendMsgV2(c *gin.Context) {
|
|||
time.Sleep(time.Second * 1)
|
||||
}
|
||||
}
|
||||
|
||||
func ServerSendMsgV3(c *gin.Context) {
|
||||
//ws
|
||||
id, _ := c.Get("id")
|
||||
user_id := int(id.(float64))
|
||||
|
||||
// 升级HTTP连接为WebSocket连接
|
||||
ws, err1 := upgrader.Upgrade(c.Writer, c.Request, nil)
|
||||
clients[ws] = true
|
||||
if err1 != nil {
|
||||
// log.Println(err)
|
||||
fmt.Println(err1)
|
||||
return
|
||||
}
|
||||
//设置用户在线状态
|
||||
worker.SetRedisWithExpire("user_"+strconv.Itoa(user_id)+"_status_v2", "1", time.Second*60)
|
||||
worker.SetRedisBitmap("im2_online_users", int64(user_id), 1)
|
||||
worker.SetRedisSetAdd("im2_online_users_set", strconv.Itoa(user_id))
|
||||
//发送消息
|
||||
key := "user_" + strconv.Itoa(user_id) + "_msg_ids"
|
||||
go subscribeAndHandleIMMessages(ws, key, user_id)
|
||||
}
|
||||
func ServerSendMsg(c *gin.Context) {
|
||||
//sse
|
||||
c.Writer.Header().Set("Content-Type", "text/event-stream")
|
||||
|
|
@ -677,54 +654,3 @@ func ServerSendMsg(c *gin.Context) {
|
|||
time.Sleep(time.Second * 1)
|
||||
}
|
||||
}
|
||||
|
||||
func subscribeAndHandleIMMessages(ws *websocket.Conn, chanel string, user_id int) {
|
||||
ctx := context.Background()
|
||||
pubsub := worker.RedisClient.Subscribe(ctx, chanel)
|
||||
defer pubsub.Close()
|
||||
defer ws.Close()
|
||||
ch := pubsub.Channel()
|
||||
for m := range ch {
|
||||
msg_id := m.Payload //消息id
|
||||
if msg_id != "" {
|
||||
msg_id_num, _ := strconv.ParseInt(msg_id, 10, 64)
|
||||
msgs := dao.FindMessageByID2(uint(msg_id_num))
|
||||
if len(msgs) > 0 {
|
||||
msg := msgs[0]
|
||||
//发送消息
|
||||
msg_str, _ := json.Marshal(msg)
|
||||
var msg_ proto.Message
|
||||
msg_.Type = "msg"
|
||||
msg_.Msg = string(msg_str)
|
||||
msg_str2, _ := json.Marshal(msg_)
|
||||
|
||||
err2 := ws.WriteMessage(websocket.TextMessage, msg_str2)
|
||||
if err2 != nil {
|
||||
worker.SetRedisWithExpire("user_"+strconv.Itoa(user_id)+"_status_v2", "0", time.Second*3600)
|
||||
clientsMux.Lock()
|
||||
delete(clients, ws)
|
||||
clientsMux.Unlock()
|
||||
break
|
||||
}
|
||||
}
|
||||
} else {
|
||||
var msg proto.Message
|
||||
msg.Type = "check"
|
||||
msg.Msg = "check"
|
||||
msg.From_user_id = -1
|
||||
//发送心跳包
|
||||
res3, _ := json.Marshal(msg)
|
||||
err2 := ws.WriteMessage(websocket.TextMessage, res3)
|
||||
if err2 != nil {
|
||||
worker.SetRedisWithExpire("user_"+strconv.Itoa(user_id)+"_status_v2", "0", time.Second*3600)
|
||||
worker.SetRedisBitmap("im2_online_users", int64(user_id), 0)
|
||||
worker.SetRedisSetRemove("im2_online_users_set", strconv.Itoa(user_id))
|
||||
clientsMux.Lock()
|
||||
delete(clients, ws)
|
||||
clientsMux.Unlock()
|
||||
break
|
||||
}
|
||||
}
|
||||
worker.SetRedisWithExpire("user_"+strconv.Itoa(user_id)+"_status_v2", "1", time.Second*60)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -160,10 +160,6 @@ func GetVideoList(c *gin.Context) {
|
|||
const layout = "2006-01-02 15:04:05"
|
||||
tm1 := time.Unix(gvl_req.StartTime, 0).Format(layout)
|
||||
tm2 := time.Unix(gvl_req.EndTime, 0).Format(layout)
|
||||
if gvl_req.StartTime == 0 || gvl_req.EndTime == 0 {
|
||||
tm1 = ""
|
||||
tm2 = ""
|
||||
}
|
||||
videos := service.GetVideoList(int(id.(float64)), tm1, tm2, gvl_req.Hour)
|
||||
c.JSON(http.StatusOK, gin.H{"data": videos, "code": proto.SuccessCode, "message": "success"})
|
||||
} else {
|
||||
|
|
|
|||
16
main.go
16
main.go
|
|
@ -3,9 +3,7 @@ package main
|
|||
import (
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/golang-jwt/jwt"
|
||||
"github.com/robfig/cron/v3"
|
||||
"io"
|
||||
"log"
|
||||
"os"
|
||||
"strings"
|
||||
"videoplayer/dao"
|
||||
|
|
@ -35,14 +33,6 @@ func main() {
|
|||
handler.SetUpToolGroup(r) // Tool
|
||||
defer dao.Close()
|
||||
defer worker.CloseRedis()
|
||||
//定时任务
|
||||
c := cron.New(cron.WithSeconds())
|
||||
// 添加每 10 秒执行一次的任务
|
||||
_, err = c.AddFunc("@every 10s", myTask)
|
||||
if err != nil {
|
||||
log.Fatal("添加定时任务失败: ", err)
|
||||
}
|
||||
c.Start()
|
||||
r.Run(":8083") // listen and serve on 0.0.0.0:8083
|
||||
}
|
||||
func init() {
|
||||
|
|
@ -148,9 +138,3 @@ func JWTAuthMiddleware() gin.HandlerFunc {
|
|||
c.Next()
|
||||
}
|
||||
}
|
||||
|
||||
func myTask() {
|
||||
// 定时任务
|
||||
//redis中取出数据
|
||||
handler.RunCron()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -11,9 +11,3 @@ type UpdateUserInfoReq struct {
|
|||
Run bool `json:"run" form:"run"` //是否运行
|
||||
Avatar string `json:"avatar" form:"avatar"` //头像
|
||||
}
|
||||
|
||||
type CIDRUN struct {
|
||||
CID uint `json:"cid" form:"cid"` //持续集成ID,查找持续集成任务
|
||||
Curr int `json:"curr" form:"curr"` //当前剩余时间，每次执行减10s小于等于0则执行
|
||||
Every int `json:"every" form:"every"` //每隔多少秒执行一次,小于等于0表示不执行，时间粒度为10s
|
||||
}
|
||||
|
|
|
|||
|
|
@ -45,8 +45,6 @@ func CreateGeneralMessageService(from_id, to_id, msg_type, group_id int, content
|
|||
if res == "1" {
|
||||
//在线,存入redis
|
||||
worker.PushRedisListWithExpire("user_"+strconv.Itoa(to_id)+"_msg_ids", strconv.Itoa(int(id)), time.Second*300)
|
||||
//发布消息
|
||||
worker.Publish("user_"+strconv.Itoa(to_id)+"_msg_ids", strconv.Itoa(int(id)), time.Second*300)
|
||||
}
|
||||
//判断接收方是否是机器人
|
||||
id_str := strconv.Itoa(to_id)
|
||||
|
|
@ -98,8 +96,6 @@ func CreateGeneralMessageService(from_id, to_id, msg_type, group_id int, content
|
|||
}
|
||||
//在线,存入redis
|
||||
worker.PushRedisListWithExpire("user_"+user_id+"_msg_ids", strconv.Itoa(int(id)), time.Second*300)
|
||||
//发布消息
|
||||
worker.Publish("user_"+user_id+"_msg_ids", strconv.Itoa(int(id)), time.Second*300)
|
||||
}
|
||||
case 3:
|
||||
//user := dao.FindUserByID(to_id)
|
||||
|
|
|
|||
|
|
@ -12,19 +12,19 @@ import (
|
|||
"github.com/go-redis/redis/v8"
|
||||
)
|
||||
|
||||
var RedisClient *redis.Client // Redis 客户端, 用于连接 Redis 服务器
|
||||
var redisClient *redis.Client // Redis 客户端, 用于连接 Redis 服务器
|
||||
func InitRedis() error {
|
||||
ctx := context.Background()
|
||||
|
||||
if proto.Config.REDIS_User_PW == false {
|
||||
// 连接redis
|
||||
RedisClient = redis.NewClient(&redis.Options{
|
||||
redisClient = redis.NewClient(&redis.Options{
|
||||
Addr: proto.Config.REDIS_ADDR, // Redis 服务器地址
|
||||
DB: proto.Config.REDIS_DB, // 使用的数据库编号
|
||||
})
|
||||
} else {
|
||||
// 连接redis
|
||||
RedisClient = redis.NewClient(&redis.Options{
|
||||
redisClient = redis.NewClient(&redis.Options{
|
||||
Addr: proto.Config.REDIS_ADDR, // Redis 服务器地址
|
||||
Password: proto.Config.REDIS_PASSWORD, // 如果 Redis 设置了密码
|
||||
DB: proto.Config.REDIS_DB, // 使用的数据库编号
|
||||
|
|
@ -32,7 +32,7 @@ func InitRedis() error {
|
|||
}
|
||||
|
||||
// 验证 Redis 客户端是否可以正常工作
|
||||
_, err := RedisClient.Ping(ctx).Result()
|
||||
_, err := redisClient.Ping(ctx).Result()
|
||||
if err != nil {
|
||||
fmt.Println("Error connecting to Redis: %v", err)
|
||||
}
|
||||
|
|
@ -41,14 +41,14 @@ func InitRedis() error {
|
|||
|
||||
func CloseRedis() {
|
||||
// 关闭 Redis 客户端
|
||||
if err := RedisClient.Close(); err != nil {
|
||||
if err := redisClient.Close(); err != nil {
|
||||
fmt.Println("Error closing Redis client: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func IsContainKey(key string) bool {
|
||||
ctx := context.Background()
|
||||
val, err := RedisClient.Exists(ctx, key).Result() // 检查键是否存在, 如果存在则返回 1, 否则返回 0
|
||||
val, err := redisClient.Exists(ctx, key).Result() // 检查键是否存在, 如果存在则返回 1, 否则返回 0
|
||||
if err != nil {
|
||||
fmt.Println("Error getting key: %v", err)
|
||||
return false
|
||||
|
|
@ -63,7 +63,7 @@ func IsContainKey(key string) bool {
|
|||
func SetRedis(key string, value string) bool {
|
||||
ctx := context.Background()
|
||||
// 设置键值对, 0 表示不设置过期时间, 如果需要设置过期时间, 可以设置为 time.Second * 10 等
|
||||
err := RedisClient.Set(ctx, key, value, time.Minute*30).Err()
|
||||
err := redisClient.Set(ctx, key, value, time.Minute*30).Err()
|
||||
if err != nil {
|
||||
fmt.Println("Error setting key: %v", err)
|
||||
return false
|
||||
|
|
@ -83,12 +83,12 @@ func SetHashWithTime(key string, id int, name, email string, duration time.Durat
|
|||
}
|
||||
|
||||
// 设置哈希表的字段值, 0 表示不设置过期时间, 如果需要设置过期时间, 可以设置为 time.Second * 10 等
|
||||
err := RedisClient.HSet(ctx, key, fields).Err()
|
||||
err := redisClient.HSet(ctx, key, fields).Err()
|
||||
if err != nil {
|
||||
fmt.Println("Error setting key: %v", err)
|
||||
return false
|
||||
}
|
||||
err = RedisClient.Expire(ctx, key, time.Hour*10).Err()
|
||||
err = redisClient.Expire(ctx, key, time.Hour*10).Err()
|
||||
if err != nil {
|
||||
fmt.Println("Error setting key: %v", err)
|
||||
return false
|
||||
|
|
@ -99,12 +99,12 @@ func SetHashWithTime(key string, id int, name, email string, duration time.Durat
|
|||
// 设置redis hash，设置过期时间
|
||||
func SetHash(key string, data map[string]interface{}) bool {
|
||||
ctx := context.Background()
|
||||
err := RedisClient.HSet(ctx, key, data).Err()
|
||||
err := redisClient.HSet(ctx, key, data).Err()
|
||||
if err != nil {
|
||||
fmt.Println("%v :Error setting hash: %v", key, err)
|
||||
return false
|
||||
}
|
||||
err = RedisClient.Expire(ctx, key, time.Minute*30).Err()
|
||||
err = redisClient.Expire(ctx, key, time.Minute*30).Err()
|
||||
if err != nil {
|
||||
fmt.Println("%v :Error setting expire: %v", key, err)
|
||||
return false
|
||||
|
|
@ -114,7 +114,7 @@ func SetHash(key string, data map[string]interface{}) bool {
|
|||
|
||||
func SetHashWithField(key string, field string, value string) bool {
|
||||
ctx := context.Background()
|
||||
err := RedisClient.HSet(ctx, key, field, value).Err()
|
||||
err := redisClient.HSet(ctx, key, field, value).Err()
|
||||
if err != nil {
|
||||
fmt.Println("Error setting key: %v", err)
|
||||
return false
|
||||
|
|
@ -124,7 +124,7 @@ func SetHashWithField(key string, field string, value string) bool {
|
|||
|
||||
func GetHash(key string, field string) string {
|
||||
ctx := context.Background()
|
||||
val, err := RedisClient.HGet(ctx, key, field).Result()
|
||||
val, err := redisClient.HGet(ctx, key, field).Result()
|
||||
if err != nil {
|
||||
fmt.Println("Error getting hash: %v", err)
|
||||
return ""
|
||||
|
|
@ -134,7 +134,7 @@ func GetHash(key string, field string) string {
|
|||
|
||||
func GetHashAll(key string) map[string]string {
|
||||
ctx := context.Background()
|
||||
val, err := RedisClient.HGetAll(ctx, key).Result()
|
||||
val, err := redisClient.HGetAll(ctx, key).Result()
|
||||
if err != nil {
|
||||
fmt.Println("Error getting hash: %v", err)
|
||||
return nil
|
||||
|
|
@ -146,7 +146,7 @@ func GetHashAll(key string) map[string]string {
|
|||
func SetRedisWithExpire(key string, value string, expire time.Duration) bool { // 设置键值对, 0 表示不设置过期时间, 如果需要设置过期时间, 可以设置为 time.Second * 10 等
|
||||
ctx := context.Background()
|
||||
// 设置键值对, 0 表示不设置过期时间, 如果需要设置过期时间, 可以设置为 time.Second * 10 等
|
||||
err := RedisClient.Set(ctx, key, value, expire).Err()
|
||||
err := redisClient.Set(ctx, key, value, expire).Err()
|
||||
if err != nil {
|
||||
fmt.Println("Error setting key: %v", err)
|
||||
return false
|
||||
|
|
@ -157,7 +157,7 @@ func SetRedisWithExpire(key string, value string, expire time.Duration) bool { /
|
|||
// 获取redis
|
||||
func GetRedis(key string) string {
|
||||
ctx := context.Background()
|
||||
val, err := RedisClient.Get(ctx, key).Result() // 从 Redis 读取键值, 如果键不存在则返回空字符串, 如果出现错误则返回错误
|
||||
val, err := redisClient.Get(ctx, key).Result() // 从 Redis 读取键值, 如果键不存在则返回空字符串, 如果出现错误则返回错误
|
||||
if err != nil {
|
||||
fmt.Println(key, " Error getting key: %v", err)
|
||||
return ""
|
||||
|
|
@ -168,7 +168,7 @@ func GetRedis(key string) string {
|
|||
// pop redis list from right,as stack
|
||||
func PopRedisList(key string) string {
|
||||
ctx := context.Background()
|
||||
val, err := RedisClient.RPop(ctx, key).Result() // 从 Redis 读取键值, 如果键不存在则返回空字符串, 如果出现错误则返回错误
|
||||
val, err := redisClient.RPop(ctx, key).Result() // 从 Redis 读取键值, 如果键不存在则返回空字符串, 如果出现错误则返回错误
|
||||
if err != nil {
|
||||
fmt.Println(key, " Error reading from Redis: %v", err)
|
||||
return ""
|
||||
|
|
@ -179,7 +179,7 @@ func PopRedisList(key string) string {
|
|||
// pop redis list from left,as queue
|
||||
func PopRedisListLeft(key string) string {
|
||||
ctx := context.Background()
|
||||
val, err := RedisClient.LPop(ctx, key).Result() // 从 Redis 读取键值, 如果键不存在则返回空字符串, 如果出现错误则返回错误
|
||||
val, err := redisClient.LPop(ctx, key).Result() // 从 Redis 读取键值, 如果键不存在则返回空字符串, 如果出现错误则返回错误
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
|
|
@ -188,7 +188,7 @@ func PopRedisListLeft(key string) string {
|
|||
|
||||
func DelRedis(key string) {
|
||||
ctx := context.Background()
|
||||
err := RedisClient.Del(ctx, key).Err()
|
||||
err := redisClient.Del(ctx, key).Err()
|
||||
if err != nil {
|
||||
fmt.Println("Error deleting key: %v", err)
|
||||
}
|
||||
|
|
@ -197,7 +197,7 @@ func DelRedis(key string) {
|
|||
// push redis list from right
|
||||
func PushRedisList(key string, value string) bool {
|
||||
ctx := context.Background()
|
||||
err := RedisClient.RPush(ctx, key, value).Err()
|
||||
err := redisClient.RPush(ctx, key, value).Err()
|
||||
if err != nil {
|
||||
fmt.Println("Error setting key: %v", err)
|
||||
return false
|
||||
|
|
@ -207,12 +207,12 @@ func PushRedisList(key string, value string) bool {
|
|||
|
||||
func PushRedisListWithExpire(key string, value string, expire time.Duration) bool {
|
||||
ctx := context.Background()
|
||||
err := RedisClient.RPush(ctx, key, value).Err()
|
||||
err := redisClient.RPush(ctx, key, value).Err()
|
||||
if err != nil {
|
||||
fmt.Println("Error setting key: %v", err)
|
||||
return false
|
||||
}
|
||||
err = RedisClient.Expire(ctx, key, expire).Err()
|
||||
err = redisClient.Expire(ctx, key, expire).Err()
|
||||
if err != nil {
|
||||
fmt.Println("Error setting key: %v", err)
|
||||
return false
|
||||
|
|
@ -223,7 +223,7 @@ func PushRedisListWithExpire(key string, value string, expire time.Duration) boo
|
|||
// delete redis key
|
||||
func delRedis(key string) {
|
||||
ctx := context.Background()
|
||||
err := RedisClient.Del(ctx, key).Err()
|
||||
err := redisClient.Del(ctx, key).Err()
|
||||
if err != nil {
|
||||
fmt.Println("Error setting key: %v", err)
|
||||
}
|
||||
|
|
@ -249,7 +249,7 @@ func (u *RUser) toJSONString() string {
|
|||
// put hash to redis
|
||||
func hSetRedis(key string, field string, value string) {
|
||||
ctx := context.Background()
|
||||
err := RedisClient.HSet(ctx, key, field, value).Err()
|
||||
err := redisClient.HSet(ctx, key, field, value).Err()
|
||||
if err != nil {
|
||||
fmt.Println("Error setting key: %v", err)
|
||||
}
|
||||
|
|
@ -258,7 +258,7 @@ func hSetRedis(key string, field string, value string) {
|
|||
// get hash from redis
|
||||
func hGetRedis(key string, field string) string {
|
||||
ctx := context.Background()
|
||||
val, err := RedisClient.HGet(ctx, key, field).Result()
|
||||
val, err := redisClient.HGet(ctx, key, field).Result()
|
||||
if err != nil {
|
||||
fmt.Println("Error getting key: %v", err)
|
||||
}
|
||||
|
|
@ -268,12 +268,12 @@ func hGetRedis(key string, field string) string {
|
|||
// 设置set，有过期时间
|
||||
func SetRedisSet(key string, values []string, expire time.Duration) bool {
|
||||
ctx := context.Background()
|
||||
err := RedisClient.SAdd(ctx, key, values).Err()
|
||||
err := redisClient.SAdd(ctx, key, values).Err()
|
||||
if err != nil {
|
||||
fmt.Println("Error setting key: %v", err)
|
||||
return false
|
||||
}
|
||||
err = RedisClient.Expire(ctx, key, expire).Err()
|
||||
err = redisClient.Expire(ctx, key, expire).Err()
|
||||
if err != nil {
|
||||
fmt.Println("Error setting key: %v", err)
|
||||
return false
|
||||
|
|
@ -284,23 +284,7 @@ func SetRedisSet(key string, values []string, expire time.Duration) bool {
|
|||
// 设置set,添加元素
|
||||
func SetRedisSetAdd(key string, value string) bool {
|
||||
ctx := context.Background()
|
||||
err := RedisClient.SAdd(ctx, key, value).Err()
|
||||
if err != nil {
|
||||
fmt.Println("Error setting key: %v", err)
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// 设置set,添加元素
|
||||
func SetRedisSetAddWithExpire(key string, value string, expire time.Duration) bool {
|
||||
ctx := context.Background()
|
||||
err := RedisClient.SAdd(ctx, key, value).Err()
|
||||
if err != nil {
|
||||
fmt.Println("Error setting key: %v", err)
|
||||
return false
|
||||
}
|
||||
err = RedisClient.Expire(ctx, key, expire).Err()
|
||||
err := redisClient.SAdd(ctx, key, value).Err()
|
||||
if err != nil {
|
||||
fmt.Println("Error setting key: %v", err)
|
||||
return false
|
||||
|
|
@ -311,7 +295,7 @@ func SetRedisSetAddWithExpire(key string, value string, expire time.Duration) bo
|
|||
// 设置set,删除元素
|
||||
func SetRedisSetRemove(key string, value string) bool {
|
||||
ctx := context.Background()
|
||||
err := RedisClient.SRem(ctx, key, value).Err()
|
||||
err := redisClient.SRem(ctx, key, value).Err()
|
||||
if err != nil {
|
||||
fmt.Println("Error setting key: %v", err)
|
||||
return false
|
||||
|
|
@ -322,7 +306,7 @@ func SetRedisSetRemove(key string, value string) bool {
|
|||
// 获取两个set的交集
|
||||
func GetRedisSetIntersect(key1 string, key2 string) []string {
|
||||
ctx := context.Background()
|
||||
val, err := RedisClient.SInter(ctx, key1, key2).Result()
|
||||
val, err := redisClient.SInter(ctx, key1, key2).Result()
|
||||
if err != nil {
|
||||
fmt.Println("Error getting key: %v", err)
|
||||
return nil
|
||||
|
|
@ -333,7 +317,7 @@ func GetRedisSetIntersect(key1 string, key2 string) []string {
|
|||
// 查看set是否包含元素
|
||||
func IsContainSet(key string, value string) bool {
|
||||
ctx := context.Background()
|
||||
val, err := RedisClient.SIsMember(ctx, key, value).Result()
|
||||
val, err := redisClient.SIsMember(ctx, key, value).Result()
|
||||
if err != nil {
|
||||
fmt.Println("Error getting key: %v", err)
|
||||
return false
|
||||
|
|
@ -344,7 +328,7 @@ func IsContainSet(key string, value string) bool {
|
|||
// 查看set的所有元素
|
||||
func GetRedisSetMembers(key string) []string {
|
||||
ctx := context.Background()
|
||||
val, err := RedisClient.SMembers(ctx, key).Result()
|
||||
val, err := redisClient.SMembers(ctx, key).Result()
|
||||
if err != nil {
|
||||
fmt.Println("Error getting key: %v", err)
|
||||
return nil
|
||||
|
|
@ -355,7 +339,7 @@ func GetRedisSetMembers(key string) []string {
|
|||
// BITMAP
|
||||
func SetRedisBitmap(key string, offset int64, value int) bool {
|
||||
ctx := context.Background()
|
||||
err := RedisClient.SetBit(ctx, key, offset, value).Err()
|
||||
err := redisClient.SetBit(ctx, key, offset, value).Err()
|
||||
if err != nil {
|
||||
fmt.Println("Error setting key: %v", err)
|
||||
return false
|
||||
|
|
@ -366,7 +350,7 @@ func SetRedisBitmap(key string, offset int64, value int) bool {
|
|||
// BITMAP获取
|
||||
func GetRedisBitmap(key string, offset int64) int {
|
||||
ctx := context.Background()
|
||||
val, err := RedisClient.GetBit(ctx, key, offset).Result()
|
||||
val, err := redisClient.GetBit(ctx, key, offset).Result()
|
||||
if err != nil {
|
||||
fmt.Println("Error getting key: %v", err)
|
||||
return 0
|
||||
|
|
@ -377,11 +361,11 @@ func GetRedisBitmap(key string, offset int64) int {
|
|||
// 发布订阅者模式-发布消息
|
||||
func Publish(channel string, message string, expire time.Duration) {
|
||||
ctx := context.Background()
|
||||
err := RedisClient.Publish(ctx, channel, message).Err()
|
||||
err := redisClient.Publish(ctx, channel, message).Err()
|
||||
if err != nil {
|
||||
fmt.Println("Error publishing message: %v", err)
|
||||
}
|
||||
err = RedisClient.Expire(ctx, channel, expire).Err()
|
||||
err = redisClient.Expire(ctx, channel, expire).Err()
|
||||
if err != nil {
|
||||
fmt.Println("Error setting key: %v", err)
|
||||
}
|
||||
|
|
@ -390,7 +374,7 @@ func Publish(channel string, message string, expire time.Duration) {
|
|||
// 发布订阅者模式-订阅消息
|
||||
func Subscribe(channel string) []string {
|
||||
ctx := context.Background()
|
||||
pubsub := RedisClient.Subscribe(ctx, channel)
|
||||
pubsub := redisClient.Subscribe(ctx, channel)
|
||||
ch := pubsub.Channel()
|
||||
defer pubsub.Close()
|
||||
var messages []string
|
||||
|
|
|
|||
Loading…
Reference in New Issue