有选择地向特定客户端发送事件

如何解决有选择地向特定客户端发送事件

我正在构建一个 gin 应用程序，其中我利用 SSE 根据 this tutorial 通知我的用户对数据库的任何相关更改。目前，我已经实现了这样的功能，即在 POST 处理程序期间通过通道向 GET 处理程序发送消息以触发所有客户端的事件发送。终点是 /events/:uid/，其中 uid 指的是一个 param 变量，它是我设置的客户端 UID。我的问题是，我对如何将 SSE 广播给特定的一组用户感到困惑，这些用户每个都在自己独特的端点上收听。现在所有客户端都被添加到地图中，但我不确定如何基于此进行过滤。

GET 端点处理程序

var eventsstream = broker.NewServer()

func GetRelevantMarkers(c *gin.Context) {
    c.Writer.Header().Set("Content-Type","text/event-stream")
    c.Writer.Header().Set("Cache-Control","no-cache")
    c.Writer.Header().Set("Connection","keep-alive")
    c.Writer.Header().Set("transfer-encoding","chunked")

    clientChan := make(broker.ClientChan)

    // Send new connection to event server
    eventsstream.NewClients <- clientChan

    defer func() {
        // Send closed connection to event server
        eventsstream.ClosedClients <- clientChan
    }()

    go func() {
        // Send connection that is closed by client to event server
        <-c.Done()
        eventsstream.ClosedClients <- clientChan
    }()


    c.Stream(func(w io.Writer) bool {
        // Stream message to client from message channel
        if msg,ok := <-eventsstream.Message; ok {
            c.SSEvent("message",msg)
            return true
        }
        return false
    })
}

事件代理

type Event struct {
    // Events are pushed to this channel by the main events-gathering routine
    Message chan string

    // New client connections
    NewClients chan chan string

    // Closed client connections
    ClosedClients chan chan string

    // Total client connections
    TotalClients map[chan string]bool
}

type ClientChan chan string

func NewServer() (event *Event) {

    event = &Event{
        Message:       make(chan string),NewClients:    make(chan chan string),ClosedClients: make(chan chan string),TotalClients:  make(map[chan string]bool),}

    go event.listen()

    return
}

func (stream *Event) listen() {
    for {
        select {
        // Add new available client
        case client := <-stream.NewClients:
            stream.TotalClients[client] = true
            log.Printf("Client added. %d registered clients",len(stream.TotalClients))

        // Remove closed client
        case client := <-stream.ClosedClients:
            delete(stream.TotalClients,client)
            log.Printf("Removed client. %d registered clients",len(stream.TotalClients))

        // broadcast message to client
        case eventMsg := <-stream.Message:
            for clientMessageChan := range stream.TotalClients {
                clientMessageChan <- eventMsg
            }
        }
    }
}

POST 端点处理程序

func DeleteEvent(c *gin.Context) {
    eventsstream.Message <- "event-deleted"
}