shkong/go-cqhttp
1
0
Fork 0
mirror of https://github.com/Mrs4s/go-cqhttp.git synced 2025-05-06 03:53:50 +08:00
Code Issues Projects Releases Wiki Activity

clean ws client push event.

Browse Source
This commit is contained in:
wdvxdr 2021-07-10 23:29:42 +08:00
parent 34c22bed9c
commit ca8d28f913
No known key found for this signature in database
GPG Key ID: 55FF1414A69CEBA6
3 changed files with 29 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
coolq/bot.go
View File

@ -45,6 +45,7 @@ type CQBot struct {
// MSG 消息Map // MSG 消息Map
type MSG map[string]interface{} type MSG map[string]interface{}
// Event 事件
type Event struct { type Event struct {
RawMsg MSG RawMsg MSG
@ -59,15 +60,15 @@ func (e *Event) marshal() {
_ = json.NewEncoder(e.buffer).Encode(e.RawMsg) _ = json.NewEncoder(e.buffer).Encode(e.RawMsg)
} }
// JsonBytes return byes of json by lazy marshalling. // JSONBytes return byes of json by lazy marshalling.
func (e *Event) JsonBytes() []byte { func (e *Event) JSONBytes() []byte {
e.once.Do(e.marshal) e.once.Do(e.marshal)
return e.buffer.Bytes() return e.buffer.Bytes()
} }
// JsonString return string of json without extra allocation // JSONString return string of json without extra allocation
// by lazy marshalling. // by lazy marshalling.
func (e *Event) JsonString() string { func (e *Event) JSONString() string {
e.once.Do(e.marshal) e.once.Do(e.marshal)
return utils.B2S(e.buffer.Bytes()) return utils.B2S(e.buffer.Bytes())
} }

14
server/http.go
View File

@ -197,20 +197,20 @@ func (c *HTTPClient) onBotPushEvent(e *coolq.Event) {
var res string var res string
if c.filter != "" { if c.filter != "" {
filter := findFilter(c.filter) filter := findFilter(c.filter)
if filter != nil && !filter.Eval(gjson.Parse(e.JsonString())) { if filter != nil && !filter.Eval(gjson.Parse(e.JSONString())) {
log.Debugf("上报Event %v 到 HTTP 服务器 %s 时被过滤.", c.addr, e.JsonBytes()) log.Debugf("上报Event %v 到 HTTP 服务器 %s 时被过滤.", c.addr, e.JSONBytes())
return return
} }
} }
err := gout.POST(c.addr).SetJSON(e.JsonBytes()).BindBody(&res).SetHeader(func() gout.H { err := gout.POST(c.addr).SetJSON(e.JSONBytes()).BindBody(&res).SetHeader(func() gout.H {
h := gout.H{ h := gout.H{
"X-Self-ID": c.bot.Client.Uin, "X-Self-ID": c.bot.Client.Uin,
"User-Agent": "CQHttp/4.15.0", "User-Agent": "CQHttp/4.15.0",
} }
if c.secret != "" { if c.secret != "" {
mac := hmac.New(sha1.New, []byte(c.secret)) mac := hmac.New(sha1.New, []byte(c.secret))
_, err := mac.Write(e.JsonBytes()) _, err := mac.Write(e.JSONBytes())
if err != nil { if err != nil {
log.Error(err) log.Error(err)
return nil return nil
@ -228,12 +228,12 @@ func (c *HTTPClient) onBotPushEvent(e *coolq.Event) {
return nil return nil
}).Do() }).Do()
if err != nil { if err != nil {
log.Warnf("上报Event数据 %s 到 %v 失败: %v", e.JsonBytes(), c.addr, err) log.Warnf("上报Event数据 %s 到 %v 失败: %v", e.JSONBytes(), c.addr, err)
return return
} }
log.Debugf("上报Event数据 %s 到 %v", e.JsonBytes(), c.addr) log.Debugf("上报Event数据 %s 到 %v", e.JSONBytes(), c.addr)
if gjson.Valid(res) { if gjson.Valid(res) {
c.bot.CQHandleQuickOperation(gjson.Parse(e.JsonString()), gjson.Parse(res)) c.bot.CQHandleQuickOperation(gjson.Parse(e.JSONString()), gjson.Parse(res))
} }
} }

44
server/websocket.go
View File

@ -230,39 +230,29 @@ func (c *websocketClient) listenAPI(conn *webSocketConn, u bool) {
func (c *websocketClient) onBotPushEvent(e *coolq.Event) { func (c *websocketClient) onBotPushEvent(e *coolq.Event) {
filter := findFilter(c.filter) filter := findFilter(c.filter)
if filter != nil && !filter.Eval(gjson.Parse(e.JsonString())) { if filter != nil && !filter.Eval(gjson.Parse(e.JSONString())) {
log.Debugf("上报Event %s 到 WS客户端 时被过滤.", e.JsonBytes()) log.Debugf("上报Event %s 到 WS服务器 时被过滤.", e.JSONBytes())
return return
} }
if c.eventConn != nil { push := func(conn *webSocketConn, reconnect func()) {
log.Debugf("向WS服务器 %v 推送Event: %s", c.eventConn.RemoteAddr().String(), e.JsonBytes()) log.Debugf("向WS服务器 %v 推送Event: %s", conn.RemoteAddr().String(), e.JSONBytes())
conn := c.eventConn
conn.Lock() conn.Lock()
defer conn.Unlock() defer conn.Unlock()
_ = c.eventConn.SetWriteDeadline(time.Now().Add(time.Second * 15)) _ = conn.SetWriteDeadline(time.Now().Add(time.Second * 15))
if err := c.eventConn.WriteMessage(websocket.TextMessage, e.JsonBytes()); err != nil { if err := conn.WriteMessage(websocket.TextMessage, e.JSONBytes()); err != nil {
log.Warnf("向WS服务器 %v 推送Event时出现错误: %v", c.eventConn.RemoteAddr().String(), err) log.Warnf("向WS服务器 %v 推送Event时出现错误: %v", conn.RemoteAddr().String(), err)
_ = c.eventConn.Close() _ = conn.Close()
if c.conf.ReconnectInterval != 0 { if c.conf.ReconnectInterval != 0 {
time.Sleep(time.Millisecond * time.Duration(c.conf.ReconnectInterval)) time.Sleep(time.Millisecond * time.Duration(c.conf.ReconnectInterval))
c.connectEvent() reconnect()
} }
} }
} }
if c.eventConn != nil {
push(c.eventConn, c.connectEvent)
}
if c.universalConn != nil { if c.universalConn != nil {
log.Debugf("向WS服务器 %v 推送Event: %s", c.universalConn.RemoteAddr().String(), e.JsonBytes()) push(c.universalConn, c.connectUniversal)
conn := c.universalConn
conn.Lock()
defer conn.Unlock()
_ = c.universalConn.SetWriteDeadline(time.Now().Add(time.Second * 15))
if err := c.universalConn.WriteMessage(websocket.TextMessage, e.JsonBytes()); err != nil {
log.Warnf("向WS服务器 %v 推送Event时出现错误: %v", c.universalConn.RemoteAddr().String(), err)
_ = c.universalConn.Close()
if c.conf.ReconnectInterval != 0 {
time.Sleep(time.Millisecond * time.Duration(c.conf.ReconnectInterval))
c.connectUniversal()
}
}
} }
} }
@ -391,16 +381,16 @@ func (s *webSocketServer) onBotPushEvent(e *coolq.Event) {
defer s.eventConnMutex.Unlock() defer s.eventConnMutex.Unlock()
filter := findFilter(s.filter) filter := findFilter(s.filter)
if filter != nil && !filter.Eval(gjson.Parse(e.JsonString())) { if filter != nil && !filter.Eval(gjson.Parse(e.JSONString())) {
log.Debugf("上报Event %s 到 WS客户端 时被过滤.", e.JsonBytes()) log.Debugf("上报Event %s 到 WS客户端 时被过滤.", e.JSONBytes())
return return
} }
for i, l := 0, len(s.eventConn); i < l; i++ { for i, l := 0, len(s.eventConn); i < l; i++ {
conn := s.eventConn[i] conn := s.eventConn[i]
log.Debugf("向WS客户端 %v 推送Event: %s", conn.RemoteAddr().String(), e.JsonBytes()) log.Debugf("向WS客户端 %v 推送Event: %s", conn.RemoteAddr().String(), e.JSONBytes())
conn.Lock() conn.Lock()
if err := conn.WriteMessage(websocket.TextMessage, e.JsonBytes()); err != nil { if err := conn.WriteMessage(websocket.TextMessage, e.JSONBytes()); err != nil {
_ = conn.Close() _ = conn.Close()
next := i + 1 next := i + 1
if next >= l { if next >= l {