diff --git a/auto/api/v1/relax.go b/auto/api/v1/relax.go index e8bb2021..dca678b1 100644 --- a/auto/api/v1/relax.go +++ b/auto/api/v1/relax.go @@ -19,6 +19,7 @@ type Relax interface { Chain() gin.HandlersChain GetUnreadMsgCount(*web.GetUnreadMsgCountReq) (*web.GetUnreadMsgCountResp, error) + StreamUnreadMsgCount(*gin.Context) mustEmbedUnimplementedRelaxServant() } @@ -62,6 +63,16 @@ func RegisterRelaxServant(e *gin.Engine, s Relax, m ...RelaxChain) { var rv _render_ = resp rv.Render(c) })...) + + // Register SSE route for streaming unread message count + router.Handle("GET", "user/msgcount/stream", append(middlewares, func(c *gin.Context) { + select { + case <-c.Request.Context().Done(): + return + default: + } + s.StreamUnreadMsgCount(c) + })...) } // UnimplementedRelaxServant can be embedded to have forward compatible implementations. diff --git a/internal/servants/web/relax.go b/internal/servants/web/relax.go index b92f167d..b4048ee3 100644 --- a/internal/servants/web/relax.go +++ b/internal/servants/web/relax.go @@ -5,6 +5,7 @@ package web import ( + "encoding/json" "github.com/gin-gonic/gin" "github.com/redis/rueidis" api "github.com/rocboss/paopao-ce/auto/api/v1" @@ -13,6 +14,7 @@ import ( "github.com/rocboss/paopao-ce/internal/servants/base" "github.com/rocboss/paopao-ce/internal/servants/chain" "github.com/sirupsen/logrus" + "time" ) var ( @@ -51,6 +53,55 @@ func (s *relaxSrv) GetUnreadMsgCount(req *web.GetUnreadMsgCountReq) (*web.GetUnr return &web.GetUnreadMsgCountResp{}, nil } +func (s *relaxSrv) StreamUnreadMsgCount(c *gin.Context) { + c.Header("Content-Type", "text/event-stream") + c.Header("Cache-Control", "no-cache") + c.Header("Connection", "keep-alive") + c.Header("X-Accel-Buffering", "no") + + uid, exists := c.Get("uid") + if !exists { + c.AbortWithStatus(401) + return + } + uidInt, ok := uid.(int64) + if !ok { + c.AbortWithStatus(400) + return + } + + var lastCount int64 = -1 + ticker := time.NewTicker(1 * time.Second) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + data, xerr := s.wc.GetUnreadMsgCountResp(uidInt) + count := int64(0) + + if xerr == nil && len(data) > 0 { + var msgCount struct { + Count int64 `json:"count"` + } + if err := json.Unmarshal(data, &msgCount); err == nil { + count = msgCount.Count + } + } + + if count != lastCount { + c.SSEvent("message", map[string]int64{"count": count}) + lastCount = count + } else { + c.SSEvent("", "") + } + c.Writer.Flush() + case <-c.Writer.CloseNotify(): + return + } + } +} + func newRelaxSrv(s *base.DaoServant, wc core.WebCache) api.Relax { return &relaxSrv{ DaoServant: s,