✨ abnormal message notification
This commit is contained in:
landaiqing
@@ -1,197 +0,0 @@
|
||||
package websocket_api
|
||||
|
||||
import (
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/gorilla/websocket"
|
||||
"log"
|
||||
"net/http"
|
||||
"schisandra-cloud-album/common/result"
|
||||
"schisandra-cloud-album/global"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
var (
|
||||
// 消息通道
|
||||
msg = make(map[string]chan interface{})
|
||||
// websocket客户端链接池
|
||||
client = make(map[string]*websocket.Conn)
|
||||
// 互斥锁，防止程序对统一资源同时进行读写
|
||||
mux sync.Mutex
|
||||
)
|
||||
|
||||
// NewSocketClient 创建websocket服务
|
||||
// @Summary 创建websocket服务(gorilla)
|
||||
// @Description 创建websocket服务
|
||||
// @Tags websocket
|
||||
// @Router /api/ws/socket [get]
|
||||
func (WebsocketAPI) NewSocketClient(context *gin.Context) {
|
||||
id := context.Query("client_id")
|
||||
global.LOG.Println(id + "websocket链接")
|
||||
// 升级为websocket长链接
|
||||
WsHandler(context.Writer, context.Request, id)
|
||||
}
|
||||
|
||||
// DeleteClient api:/deleteClient接口处理函数
|
||||
func (WebsocketAPI) DeleteClient(context *gin.Context) {
|
||||
id := context.Query("client_id")
|
||||
// 关闭websocket链接
|
||||
conn, exist := getClient(id)
|
||||
if exist {
|
||||
err := conn.Close()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
deleteClient(id)
|
||||
} else {
|
||||
result.FailWithMessage("客户端不存在", context)
|
||||
}
|
||||
// 关闭其消息通道
|
||||
_, exist = getMsgChannel(id)
|
||||
if exist {
|
||||
deletemsgChannel(id)
|
||||
}
|
||||
}
|
||||
|
||||
// SendMessageData 发送消息接口处理函数
|
||||
func SendMessageData(clientId string, data interface{}) bool {
|
||||
m, exist := getMsgChannel(clientId)
|
||||
if !exist {
|
||||
log.Println("未找到该客户端的消息通道")
|
||||
return false
|
||||
}
|
||||
// 向消息通道发送消息
|
||||
select {
|
||||
case m <- data:
|
||||
global.LOG.Println("发送消息给客户端：" + clientId)
|
||||
return true
|
||||
default:
|
||||
global.LOG.Println("消息通道已满，消息发送失败")
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// websocket Upgrader
|
||||
var wsupgrader = websocket.Upgrader{
|
||||
ReadBufferSize: 1024,
|
||||
WriteBufferSize: 1024,
|
||||
HandshakeTimeout: 5 * time.Second,
|
||||
// 取消ws跨域校验
|
||||
CheckOrigin: func(r *http.Request) bool {
|
||||
return true
|
||||
},
|
||||
}
|
||||
|
||||
// WsHandler 处理ws请求
|
||||
func WsHandler(w http.ResponseWriter, r *http.Request, id string) {
|
||||
var conn *websocket.Conn
|
||||
var err error
|
||||
var exist bool
|
||||
// 创建一个定时器用于服务端心跳
|
||||
pingTicker := time.NewTicker(time.Second * 10)
|
||||
conn, err = wsupgrader.Upgrade(w, r, nil)
|
||||
if err != nil {
|
||||
global.LOG.Println(err)
|
||||
return
|
||||
}
|
||||
// 把与客户端的链接添加到客户端链接池中
|
||||
addClient(id, conn)
|
||||
|
||||
// 获取该客户端的消息通道
|
||||
m, exist := getMsgChannel(id)
|
||||
if !exist {
|
||||
m = make(chan interface{})
|
||||
addMsgChannel(id, m)
|
||||
}
|
||||
// 设置客户端关闭ws链接回调函数
|
||||
conn.SetCloseHandler(func(code int, text string) error {
|
||||
deleteClient(id)
|
||||
return nil
|
||||
})
|
||||
defer conn.Close()
|
||||
for {
|
||||
_, _, err := conn.ReadMessage()
|
||||
if err != nil {
|
||||
global.LOG.Error(err)
|
||||
return
|
||||
}
|
||||
select {
|
||||
case content, _ := <-m:
|
||||
// 从消息通道接收消息，然后推送给前端
|
||||
err = conn.WriteJSON(content)
|
||||
if err != nil {
|
||||
err = conn.Close()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
deleteClient(id)
|
||||
break
|
||||
}
|
||||
case <-pingTicker.C:
|
||||
// 服务端心跳:每20秒ping一次客户端，查看其是否在线
|
||||
err := conn.SetWriteDeadline(time.Now().Add(time.Second * 20))
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
err = conn.WriteMessage(websocket.PingMessage, []byte{})
|
||||
if err != nil {
|
||||
log.Println("send pong err:", err)
|
||||
err := conn.Close()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
deleteClient(id)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
// 将客户端添加到客户端链接池
|
||||
func addClient(id string, conn *websocket.Conn) {
|
||||
mux.Lock()
|
||||
client[id] = conn
|
||||
mux.Unlock()
|
||||
}
|
||||
|
||||
// 获取指定客户端链接
|
||||
func getClient(id string) (conn *websocket.Conn, exist bool) {
|
||||
mux.Lock()
|
||||
conn, exist = client[id]
|
||||
mux.Unlock()
|
||||
return
|
||||
}
|
||||
|
||||
// 删除客户端链接
|
||||
func deleteClient(id string) {
|
||||
mux.Lock()
|
||||
delete(client, id)
|
||||
log.Println(id + "websocket退出")
|
||||
mux.Unlock()
|
||||
}
|
||||
|
||||
// 添加用户消息通道
|
||||
func addMsgChannel(id string, m chan interface{}) {
|
||||
mux.Lock()
|
||||
msg[id] = m
|
||||
mux.Unlock()
|
||||
}
|
||||
|
||||
// 获取指定用户消息通道
|
||||
func getMsgChannel(id string) (m chan interface{}, exist bool) {
|
||||
mux.Lock()
|
||||
defer mux.Unlock()
|
||||
m, exist = msg[id]
|
||||
return
|
||||
}
|
||||
|
||||
// 删除指定消息通道
|
||||
func deletemsgChannel(id string) {
|
||||
mux.Lock()
|
||||
if m, ok := msg[id]; ok {
|
||||
close(m)
|
||||
delete(msg, id)
|
||||
}
|
||||
mux.Unlock()
|
||||
}
|
||||
@@ -5,4 +5,5 @@ type Wechat struct {
|
||||
AppSecret string `yaml:"app-secret"`
|
||||
Token string `yaml:"token"`
|
||||
AESKey string `yaml:"aes-key"`
|
||||
OpenID string `yaml:"openid"`
|
||||
}
|
||||
|
||||
@@ -60,3 +60,4 @@ DuplicateLogin = "duplicate login!"
|
||||
PermissionDenied = "permission denied!"
|
||||
LogoutFailed = "logout failed!"
|
||||
LogoutSuccess = "logout successfully!"
|
||||
SystemError = "system error, please contact the administrator!"
|
||||
|
||||
@@ -60,3 +60,4 @@ DuplicateLogin = "重复登录!"
|
||||
PermissionDenied = "权限不足!"
|
||||
LogoutFailed = "登出失败!"
|
||||
LogoutSuccess = "登出成功!"
|
||||
SystemError = "系统错误!,请联系管理员!"
|
||||
|
||||
40
middleware/exception_notification.go
Normal file
40
middleware/exception_notification.go
Normal file
@@ -0,0 +1,40 @@
|
||||
package middleware
|
||||
|
||||
import (
|
||||
"github.com/ArtisanCloud/PowerWeChat/v3/src/kernel/messages"
|
||||
ginI18n "github.com/gin-contrib/i18n"
|
||||
"github.com/gin-gonic/gin"
|
||||
"schisandra-cloud-album/common/result"
|
||||
"schisandra-cloud-album/global"
|
||||
"schisandra-cloud-album/utils"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ExceptionNotification 异常通知中间件
|
||||
func ExceptionNotification() gin.HandlerFunc {
|
||||
return func(c *gin.Context) {
|
||||
defer func() {
|
||||
if err := recover(); err != nil {
|
||||
openID := global.CONFIG.Wechat.OpenID
|
||||
content := `
|
||||
系统异常通知：
|
||||
请求时间：` + time.Now().Format("2006-01-02 15:04:05") + `
|
||||
请求IP：` + utils.GetClientIP(c) + `
|
||||
请求地址：` + c.Request.URL.String() + `
|
||||
请求方法：` + c.Request.Method + `
|
||||
请求参数：` + c.Request.Form.Encode() + `
|
||||
错误信息：` + err.(error).Error() + `
|
||||
`
|
||||
messages.NewRaw(`
|
||||
{
|
||||
"touser":"` + openID + `",
|
||||
"msgtype":"text",
|
||||
"text":{"content":"` + content + `"}"}
|
||||
}
|
||||
`)
|
||||
result.FailWithMessage(ginI18n.MustGetMessage(c, "SystemError"), c)
|
||||
}
|
||||
}()
|
||||
c.Next()
|
||||
}
|
||||
}
|
||||
@@ -10,8 +10,6 @@ var websocketAPI = api.Api.WebsocketApi
|
||||
func WebsocketRouter(router *gin.RouterGroup) {
|
||||
group := router.Group("/ws")
|
||||
{
|
||||
group.GET("/socket", websocketAPI.NewSocketClient)
|
||||
group.GET("/delete", websocketAPI.DeleteClient)
|
||||
group.GET("/gws", websocketAPI.NewGWSServer)
|
||||
}
|
||||
|
||||
|
||||
@@ -30,7 +30,7 @@ func InitRouter() *gin.Engine {
|
||||
MaxAge: 12 * time.Hour,
|
||||
}))
|
||||
// 国际化设置
|
||||
router.Use(middleware.I18n())
|
||||
router.Use(middleware.I18n(), middleware.ExceptionNotification())
|
||||
|
||||
publicGroup := router.Group("api") // 不需要鉴权的路由组
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user