sse获取未读消息

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.

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,