我使用 Go 已經有一段時間了，但之前從未使用過 SSE。我有一個問題，有人可以提供一個服務器發送事件的工作示例，該示例只會發送給特定用戶（連接）。我正在使用大猩猩 - 會話進行身份驗證，我想使用 UserID 來分隔連接?；蛘呶覒撏ㄟ^ Ajax 使用 5 秒輪詢？非常感謝這是我發現并嘗試過的：https://gist.github.com/ismasan/3fb75381cd2deb6bfa9c它不會發送給單個用戶，如果連接關閉，go func 不會停止https://github.com/striversity/gotr/blob/master/010-server-sent-event-part-2/main.go這正是我所需要的，但一旦連接被刪除它就不會跟蹤。所以現在，一旦您在私人窗口中關閉并打開瀏覽器，它就根本無法工作。此外，如上所述，go 例程繼續進行。
1 回答
慕容森
TA貢獻1853條經驗 獲得超18個贊
創建一個“代理”以將消息分發給連接的用戶：
type Broker struct {
// users is a map where the key is the user id
// and the value is a slice of channels to connections
// for that user id
users map[string][]chan []byte
// actions is a channel of functions to call
// in the broker's goroutine. The broker executes
// everything in that single goroutine to avoid
// data races.
actions chan func()
}
// run executes in a goroutine. It simply gets and
// calls functions.
func (b *Broker) run() {
for a := range b.actions {
a()
}
}
func newBroker() *Broker {
b := &Broker{
users: make(map[string][]chan []byte),
actions: make(chan func()),
}
go b.run()
return b
}
// addUserChan adds a channel for user with given id.
func (b *Broker) addUserChan(id string, ch chan []byte) {
b.actions <- func() {
b.users[id] = append(b.users[id], ch)
}
}
// removeUserchan removes a channel for a user with the given id.
func (b *Broker) removeUserChan(id string, ch chan []byte) {
// The broker may be trying to send to
// ch, but nothing is receiving. Pump ch
// to prevent broker from getting stuck.
go func() { for range ch {} }()
b.actions <- func() {
chs := b.users[id]
i := 0
for _, c := range chs {
if c != ch {
chs[i] = c
i = i + 1
}
}
if i == 0 {
delete(b.users, id)
} else {
b.users[id] = chs[:i]
}
// Close channel to break loop at beginning
// of removeUserChan.
// This must be done in broker goroutine
// to ensure that broker does not send to
// closed goroutine.
close(ch)
}
}
// sendToUser sends a message to all channels for the given user id.
func (b *Broker) sendToUser(id string, data []byte) {
b.actions <- func() {
for _, ch := range b.users[id] {
ch <- data
}
}
}
在包級別使用代理聲明一個變量：
var broker = newBroker()
使用代理編寫 SSE 端點：
func sseEndpoint(w http.ResponseWriter, r *http.Request) {
// I assume that user id is in query string for this example,
// You should use your authentication code to get the id.
id := r.FormValue("id")
// Do the usual SSE setup.
flusher := w.(http.Flusher)
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", "keep-alive")
// Create channel to receive messages for this connection.
// Register that channel with the broker.
// On return from the function, remove the channel
// from the broker.
ch := make(chan []byte)
broker.addUserChan(id, ch)
defer broker.removeUserChan(id, ch)
for {
select {
case <-r.Context().Done():
// User closed the connection. We are out of here.
return
case m := <-ch:
// We got a message. Do the usual SSE stuff.
fmt.Fprintf(w, "data: %s\n\n", m)
flusher.Flush()
}
}
}
將代碼添加到您的應用程序以調用 Broker.sendToUser。
- 1 回答
- 0 關注
- 133 瀏覽
添加回答
舉報
0/150
提交
取消