Refresh presence on websocket pong
This commit is contained in:
@@ -12,6 +12,7 @@ package main
|
|||||||
import (
|
import (
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"net/http"
|
"net/http"
|
||||||
|
"sync/atomic"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/gorilla/websocket"
|
"github.com/gorilla/websocket"
|
||||||
@@ -45,6 +46,8 @@ func (sess *Session) readLoop() {
|
|||||||
sess.ws.SetReadLimit(globals.maxMessageSize)
|
sess.ws.SetReadLimit(globals.maxMessageSize)
|
||||||
sess.ws.SetReadDeadline(time.Now().Add(pongWait))
|
sess.ws.SetReadDeadline(time.Now().Add(pongWait))
|
||||||
sess.ws.SetPongHandler(func(string) error {
|
sess.ws.SetPongHandler(func(string) error {
|
||||||
|
atomic.StoreInt64(&sess.lastAction, types.TimeNow().UnixNano())
|
||||||
|
sess.refreshPresence("ws.pong")
|
||||||
sess.ws.SetReadDeadline(time.Now().Add(pongWait))
|
sess.ws.SetReadDeadline(time.Now().Add(pongWait))
|
||||||
return nil
|
return nil
|
||||||
})
|
})
|
||||||
|
|||||||
+15
-10
@@ -519,16 +519,7 @@ func (s *Session) dispatchRaw(raw []byte) {
|
|||||||
func (s *Session) dispatch(msg *ClientComMessage) {
|
func (s *Session) dispatch(msg *ClientComMessage) {
|
||||||
now := types.TimeNow()
|
now := types.TimeNow()
|
||||||
atomic.StoreInt64(&s.lastAction, now.UnixNano())
|
atomic.StoreInt64(&s.lastAction, now.UnixNano())
|
||||||
if !s.uid.IsZero() && s.tenantID.IsValid() {
|
s.refreshPresence("s.dispatch")
|
||||||
if err := store.RefreshUserPresence(s.tenantID, s.uid, s.sid, currentNodeID()); err != nil {
|
|
||||||
logs.Warn.Println("s.dispatch: failed to refresh user presence", s.uid, err, s.sid)
|
|
||||||
}
|
|
||||||
for _, topic := range s.groupSubs() {
|
|
||||||
if err := store.MarkTopicMemberOnline(s.tenantID, topic, s.uid, currentNodeID()); err != nil {
|
|
||||||
logs.Warn.Println("s.dispatch: failed to refresh group member presence", topic, s.uid, err, s.sid)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// This should be the first block here, before any other checks.
|
// This should be the first block here, before any other checks.
|
||||||
var resp *ServerComMessage
|
var resp *ServerComMessage
|
||||||
@@ -701,6 +692,20 @@ func (s *Session) dispatch(msg *ClientComMessage) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (s *Session) refreshPresence(context string) {
|
||||||
|
if s.uid.IsZero() || !s.tenantID.IsValid() {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if err := store.RefreshUserPresence(s.tenantID, s.uid, s.sid, currentNodeID()); err != nil {
|
||||||
|
logs.Warn.Println(context+": failed to refresh user presence", s.uid, err, s.sid)
|
||||||
|
}
|
||||||
|
for _, topic := range s.groupSubs() {
|
||||||
|
if err := store.MarkTopicMemberOnline(s.tenantID, topic, s.uid, currentNodeID()); err != nil {
|
||||||
|
logs.Warn.Println(context+": failed to refresh group member presence", topic, s.uid, err, s.sid)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// Request to subscribe to a topic.
|
// Request to subscribe to a topic.
|
||||||
func (s *Session) subscribe(msg *ClientComMessage) {
|
func (s *Session) subscribe(msg *ClientComMessage) {
|
||||||
if strings.HasPrefix(msg.Original, "new") || strings.HasPrefix(msg.Original, "nch") {
|
if strings.HasPrefix(msg.Original, "new") || strings.HasPrefix(msg.Original, "nch") {
|
||||||
|
|||||||
Reference in New Issue
Block a user