84 lines
1.8 KiB
Go
84 lines
1.8 KiB
Go
// Package events 把仓库变更广播给所有在线客户端（WebSocket）。
|
|
package events
|
|
|
|
import (
|
|
"encoding/json"
|
|
"log/slog"
|
|
"sync"
|
|
)
|
|
|
|
// Event 为推送给客户端的消息
|
|
//
|
|
// tree 目录结构变化（新增/删除/移动），客户端重新拉取目录树
|
|
// changed 笔记内容变化，Rev 为新版本
|
|
// deleted 笔记被删除
|
|
// moved 文件或目录被移动（From -> Path），目录事件包含其子项的路径映射
|
|
// settings 应用设置变化，客户端重新读取
|
|
type Event struct {
|
|
Type string `json:"type"`
|
|
Path string `json:"path,omitempty"`
|
|
From string `json:"from,omitempty"`
|
|
Rev string `json:"rev,omitempty"`
|
|
}
|
|
|
|
type Hub struct {
|
|
mu sync.Mutex
|
|
clients map[chan []byte]struct{}
|
|
closed bool
|
|
}
|
|
|
|
func NewHub() *Hub { return &Hub{clients: map[chan []byte]struct{}{}} }
|
|
|
|
// Subscribe 返回消息通道与取消函数；通道在客户端跟不上时会被关闭，客户端应重连并全量同步
|
|
func (h *Hub) Subscribe() (<-chan []byte, func()) {
|
|
ch := make(chan []byte, 64)
|
|
h.mu.Lock()
|
|
if h.closed {
|
|
close(ch)
|
|
h.mu.Unlock()
|
|
return ch, func() {}
|
|
}
|
|
h.clients[ch] = struct{}{}
|
|
h.mu.Unlock()
|
|
return ch, func() {
|
|
h.mu.Lock()
|
|
if _, ok := h.clients[ch]; ok {
|
|
delete(h.clients, ch)
|
|
close(ch)
|
|
}
|
|
h.mu.Unlock()
|
|
}
|
|
}
|
|
|
|
func (h *Hub) Broadcast(events ...Event) {
|
|
if len(events) == 0 {
|
|
return
|
|
}
|
|
msg, err := json.Marshal(events)
|
|
if err != nil {
|
|
slog.Error("序列化事件失败", "err", err)
|
|
return
|
|
}
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
for ch := range h.clients {
|
|
select {
|
|
case ch <- msg:
|
|
default:
|
|
delete(h.clients, ch)
|
|
close(ch)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Close 禁止新订阅，并通知现有连接关闭，供实例恢复重新打开数据库使用。
|
|
func (h *Hub) Close() {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
h.closed = true
|
|
for ch := range h.clients {
|
|
delete(h.clients, ch)
|
|
close(ch)
|
|
}
|
|
}
|