You've already forked agentic-coding-workflow
85 lines
2.5 KiB
Go
85 lines
2.5 KiB
Go
// stream.go 实现通知 inbox 的实时推送（spec §11 步骤8）。
|
|
//
|
|
// StreamHub 是 per-user 的订阅扇出中心：每个浏览器标签页 Subscribe 一个有界缓冲
|
|
// channel；Dispatcher 落库成功后 Publish 到该用户的全部订阅者。慢订阅者（缓冲满）
|
|
// 直接丢弃该消息（drop-on-slow），绝不阻塞 Publish / Dispatch，避免 goroutine /
|
|
// 内存泄漏（镜像 chat StreamerHub 的丢弃纪律）。
|
|
package notify
|
|
|
|
import (
|
|
"sync"
|
|
|
|
"github.com/google/uuid"
|
|
)
|
|
|
|
// StreamHub 是 Dispatcher 推送实时通知所需的窄接口。
|
|
type StreamHub interface {
|
|
Subscribe(userID uuid.UUID) (<-chan Message, func())
|
|
Publish(userID uuid.UUID, m Message)
|
|
}
|
|
|
|
// streamSubBuffer 是单个订阅者的缓冲深度。满时新消息被丢弃（客户端可用 inbox
|
|
// 列表接口补偿，与 chat WS onopen 主动拉取同理）。
|
|
const streamSubBuffer = 16
|
|
|
|
// memHub 是 StreamHub 的进程内实现。
|
|
type memHub struct {
|
|
mu sync.Mutex
|
|
subs map[uuid.UUID]map[int]chan Message
|
|
next int
|
|
}
|
|
|
|
// NewStreamHub 构造进程内的 per-user 通知推送中心。
|
|
func NewStreamHub() StreamHub {
|
|
return &memHub{subs: map[uuid.UUID]map[int]chan Message{}}
|
|
}
|
|
|
|
// Subscribe 为某用户注册一个订阅者，返回只读 channel 与取消函数。取消函数幂等：
|
|
// 关闭 channel 并从订阅表移除；调用方应在请求结束时 defer 调用。
|
|
func (h *memHub) Subscribe(userID uuid.UUID) (<-chan Message, func()) {
|
|
ch := make(chan Message, streamSubBuffer)
|
|
h.mu.Lock()
|
|
if h.subs[userID] == nil {
|
|
h.subs[userID] = map[int]chan Message{}
|
|
}
|
|
id := h.next
|
|
h.next++
|
|
h.subs[userID][id] = ch
|
|
h.mu.Unlock()
|
|
|
|
var once sync.Once
|
|
cancel := func() {
|
|
once.Do(func() {
|
|
h.mu.Lock()
|
|
if m, ok := h.subs[userID]; ok {
|
|
if c, ok := m[id]; ok {
|
|
delete(m, id)
|
|
close(c)
|
|
}
|
|
if len(m) == 0 {
|
|
delete(h.subs, userID)
|
|
}
|
|
}
|
|
h.mu.Unlock()
|
|
})
|
|
}
|
|
return ch, cancel
|
|
}
|
|
|
|
// Publish 向某用户的全部订阅者非阻塞投递一条消息。缓冲满的订阅者被跳过。
|
|
//
|
|
// 关键：整个投递循环持有 h.mu。因为发送是非阻塞的（select default），持锁开销极小，
|
|
// 但这让 Publish 与 cancel()（同样持 h.mu 后 close channel）互斥——否则快照 channel
|
|
// 释放锁后、cancel 并发 close，会 panic "send on closed channel"。
|
|
func (h *memHub) Publish(userID uuid.UUID, m Message) {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
for _, c := range h.subs[userID] {
|
|
select {
|
|
case c <- m:
|
|
default:
|
|
// drop-on-slow：缓冲满，丢弃，避免阻塞 Dispatch。
|
|
}
|
|
}
|
|
}
|