Sakurasan
a234afe448
- internal/hub：按文章 ID 的进程内发布/订阅，信号不携带内容（未审核评论不外泄） - GET /api/comments/stream?post_id=N：SSE 长连 + 25s 心跳 + X-Accel-Buffering off - 广播点：发评论、作者编辑/删除、后台审核通过、后台删除（与公开端共享同一 hub） - 前端 EventSource 静默刷新（不亮加载态、不干扰正在输入/编辑的状态） - requestLog 的 statusRecorder 补 Flush 透传——包装器没实现 Flusher 时流式端点整体不可用
50 lines
1.3 KiB
Go
50 lines
1.3 KiB
Go
// Package hub 是一个极简的进程内发布/订阅：按主题（文章 ID）广播
|
|
// 「有变更」信号。订阅者只拿到一个空 struct 信号——不携带内容，
|
|
// 消费方（SSE 前端）收到后自己 refetch，未审核的评论内容不会经由
|
|
// 这条通道外泄。
|
|
package hub
|
|
|
|
import "sync"
|
|
|
|
type Hub struct {
|
|
mu sync.Mutex
|
|
subs map[int64]map[chan struct{}]struct{}
|
|
}
|
|
|
|
func New() *Hub {
|
|
return &Hub{subs: make(map[int64]map[chan struct{}]struct{})}
|
|
}
|
|
|
|
// Subscribe 订阅某主题；返回信号 channel 和退订函数。
|
|
// channel 容量 1：订阅者处理不过来时新信号直接丢弃（合并刷新）。
|
|
func (h *Hub) Subscribe(topic int64) (<-chan struct{}, func()) {
|
|
ch := make(chan struct{}, 1)
|
|
h.mu.Lock()
|
|
if h.subs[topic] == nil {
|
|
h.subs[topic] = make(map[chan struct{}]struct{})
|
|
}
|
|
h.subs[topic][ch] = struct{}{}
|
|
h.mu.Unlock()
|
|
off := func() {
|
|
h.mu.Lock()
|
|
delete(h.subs[topic], ch)
|
|
if len(h.subs[topic]) == 0 {
|
|
delete(h.subs, topic)
|
|
}
|
|
h.mu.Unlock()
|
|
}
|
|
return ch, off
|
|
}
|
|
|
|
// Broadcast 唤醒某主题的全部订阅者；积压的订阅者不阻塞。
|
|
func (h *Hub) Broadcast(topic int64) {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
for ch := range h.subs[topic] {
|
|
select {
|
|
case ch <- struct{}{}:
|
|
default:
|
|
}
|
|
}
|
|
}
|