Real-Time Message Push (SSE) in Practice

// app/admin/service/internal/server/sse_server.go
type SSEServer struct {
    clients map[uint32]*client.Player  // userId → 客户端连接
    mu      sync.RWMutex
}

func NewSSEServer() *SSEServer {
    return &SSEServer{
        clients: make(map[uint32]*client.Player),
    }
}

// SSE 端点，前端 EventSource 连接
func (s *SSEServer) HandleSSE(c *gin.Context) {
    userId := c.GetUint32("userId")

    c.Header("Content-Type", "text/event-stream")
    c.Header("Cache-Control", "no-cache")
    c.Header("Connection", "keep-alive")
    c.Header("Access-Control-Allow-Origin", "*")

    // 注册客户端连接
    player := client.NewPlayer(userId, c.Writer)
    s.register(userId, player)
    defer s.unregister(userId)

    // 保持连接
    ticker := time.NewTicker(30 * time.Second)
    defer ticker.Stop()

    for {
        select {
        case <-c.Done():
            return
        case <-ticker.C:
            // 心跳保活
            c.SSEvent("ping", "heartbeat")
            c.Writer.Flush()
        }
    }
}

// 向指定用户推送消息
func (s *SSEServer) Push(userId uint32, event string, data interface{}) {
    s.mu.RLock()
    defer s.mu.RUnlock()

    if client, exists := s.clients[userId]; exists {
        client.Send(event, data)
    }
}

// 广播给所有在线用户
func (s *SSEServer) Broadcast(event string, data interface{}) {
    s.mu.RLock()
    defer s.mu.RUnlock()

    for _, client := range s.clients {
        client.Send(event, data)
    }
}

// app/core/service/internal/service/event_handler.go
func (s *EventHandler) OnCommentCreated(ctx context.Context, event *CommentCreatedEvent) {
    // 1. 通知管理后台：新评论待审核
    s.adminSSEClient.Push(event.PostAuthorId, "comment:pending", map[string]any{
        "commentId": event.CommentId,
        "postId":    event.PostId,
        "content":   event.Content,
        "userName":  event.UserName,
    })

    // 2. 通知前台用户：文章有新评论
    s.appSSEClient.Push(event.PostAuthorId, "comment:new", map[string]any{
        "postId":    event.PostId,
        "postTitle": event.PostTitle,
        "userName":  event.UserName,
    })
}

func (s *EventHandler) OnCommentReplied(ctx context.Context, event *CommentRepliedEvent) {
    // 通知被回复的用户
    s.appSSEClient.Push(event.ReplyToUserId, "comment:reply", map[string]any{
        "commentId":   event.CommentId,
        "postId":      event.PostId,
        "replyByUser": event.UserName,
        "content":     event.Content,
    })
}

// app/core/service/internal/service/service.go
func InitEventHandlers(eventbus *eventbus.EventBus, adminSSE *sse.Client, appSSE *sse.Client) {
    handler := &EventHandler{adminSSEClient: adminSSE, appSSEClient: appSSE}

    // 订阅评论事件
    eventbus.Subscribe("comment.created", handler.OnCommentCreated)
    eventbus.Subscribe("comment.replied", handler.OnCommentReplied)
    eventbus.Subscribe("comment.approved", handler.OnCommentApproved)

    // 订阅帖子事件
    eventbus.Subscribe("post.published", handler.OnPostPublished)
    eventbus.Subscribe("post.updated", handler.OnPostUpdated)
}

// Redis Pub/Sub 广播
func (s *SSEServer) Push(userId uint32, event string, data interface{}) {
    msg := PushMessage{
        UserId: userId,
        Event:  event,
        Data:   data,
    }
    payload, _ := json.Marshal(msg)
    // 发布到 Redis，所有 SSE 实例订阅
    redisClient.Publish(ctx, "sse:push", payload)
}

// 每个 SSE 实例订阅 Redis
func (s *SSEServer) SubscribeRedis() {
    sub := redisClient.Subscribe(ctx, "sse:push")
    for msg := range sub.Channel() {
        var pushMsg PushMessage
        json.Unmarshal([]byte(msg.Payload), &pushMsg)

        // 只推送给本实例连接的用户
        s.localPush(pushMsg.UserId, pushMsg.Event, pushMsg.Data)
    }
}