syncthing 源码阅读：events pubsub 模型

# syncthing/lib/events/events.go

## syncthing 事件管理

入口函数部分看到了 apiSub 和 diskSub

```go
apiSub := events.NewBufferedSubscription(events.Default.Subscribe(events.AllEvents&^events.LocalChangeDetected), 1000)
diskSub := events.NewBufferedSubscription(events.Default.Subscribe(events.LocalChangeDetected), 1000)
```

syncthing 的 pubsub 模块是通过 events 包来管理的，下面咱来完整的看看 events 的设计

## events 包结构

- EventType
- Event
- Logger
- Default -> Logger 实例
- Subscription
- BufferedSubscription

### EventType

定义 event 的类型常量和对应的字符串

```go
type EventType int

const (
Ping EventType = 1 << iota
Starting
....

AllEvents = (1 << iota) - 1
)
```

这里通过 iota 表达式技巧来定义常量，对应

```log
Ping 00000000
Starting 00000010
...
AllEvents 0111...11
```
常量采用类位运算的方式

1 << iota 表示向左移动 iota 位

mask & eventType != 0 来判断 mask 时候包好对应的事件类型

mask = Starting : Starting 事件
mask = Starting & Ping : Staring 和 Ping 事件
mask = AllEvents&^Starting : Starting 除外的其他事件 (&^ 表示后者先取反再执行 & 运算)

EventType String 方法返回对应常量的字符串 slice

### Event

每次时间对应一个 Event 实例 ，结构

```go
type Event struct {
// Per-subscription sequential event ID. Named "id" for backwards compatibility with the REST API
// 一个订阅者创建后，后面所有订阅者订阅的事件都会传递给订阅者-对应事件序列，SubscriptionID 是这个序列中事件对应的 ID
SubscriptionID int `json:"id"`

// 事件全局递增 ID
GlobalID int `json:"globalID"`

// 事件创建时间
Time time.Time `json:"time"`

// 事件类型
Type EventType `json:"type"`

// 时间附带数据 (interface{} 为万能类型）
Data interface{} `json:"data"`
}
```

### Logger

Logger 管理了事件的创建和订阅，结构：

```go
type Logger struct {

// 订阅者数组
subs []*Subscription

// 用于产生下次事件的 e.SubscriptionId ,对应每一个 sub 一个 int 值
nextSubscriptionIDs []int

// 事件的全局 ID 管理
nextGlobalID int

// Logger 可能会被多个 goroutine 调用，所以必须要设置互斥
mutex sync.Mutex
}

// Log 创建事件
func (l *Logger) Log(t EventType, data interface{}){}

// Subscribe 方法创建对应事件的订阅者
func (l *Logger) Subscribe(mask EventType) *Subscription {}

// 取消事件的订阅
func (l *Logger) Unsubscribe(s *Subscription) {}
```

### Default

对外暴露的默认 Logger ，调用 Default.Log 创建事件，Default.Subscribe 创建订阅者

### Subscription

订阅者

```go
const BufferSize = 64
type Subscription struct {
// 订阅的事件 mask
mask EventType
// 事件 channel , 对应一个有缓冲区的通道，缓冲区大小为 BufferSize
events chan Event
// 接收事件设置一个 timeout ,在 timeout 之内不会重复接收
timeout *time.Timer
}

// Poll returns an event from the subscription or an error if the poll times
// out of the event channel is closed. Poll should not be called concurrently
// from multiple goroutines for a single subscription.
// Poll 方法将本 goroutine 停住，等待一个事件产生，或者超时事件产生
func (s *Subscription) Poll(timeout time.Duration) (Event, error) {}

// 返回事件通道
func (s *Subscription) C() <-chan Event {}
```

### BufferedSubscription

相当于一个订阅者的事件缓冲区（感觉名字应该叫 SubscriptionEventBuffer）

```go
type bufferedSubscription struct {
// 订阅者
sub *Subscription
// 事件缓冲数组
buf []Event
// 缓冲区是有大小限制的，next 每次获得一个事件过后要放在缓冲区的什么位置
next int
// 当前的 event 对应的 SubscriptionID
cur int // Current SubscriptionID
// 多个 goroutine 场景要使用互斥和信号量来
mut sync.Mutex
cond *stdsync.Cond
}

// 创建缓冲区并启动一个 goroutine 运行 pollingLoop
func NewBufferedSubscription(s *Subscription, size int) BufferedSubscription {
bs := &bufferedSubscription{
sub: s,
buf: make([]Event, size),
mut: sync.NewMutex(),
}
bs.cond = stdsync.NewCond(bs.mut)
go bs.pollingLoop()
return bs
}

// pollingLoop 方法会启动一个循环，不断的调用 sub.Poll 方法获取事件
func (s *bufferedSubscription) pollingLoop() {}

// Since 方法会获取 id 后缓冲区内发生的所有事件
func (s *bufferedSubscription) Since(id int, into []Event) []Event {
s.mut.Lock()
defer s.mut.Unlock()

// 无线循环直到等到有发生在 id 后的事件产生
// Q: 为什么没有非要一直等呢，万一一直没有事件发生不是卡死了么
for id >= s.cur {
s.cond.Wait()
}

for i := s.next; i < len(s.buf); i++ {
if s.buf[i].SubscriptionID > id {
into = append(into, s.buf[i])
}
}
for i := 0; i < s.next; i++ {
if s.buf[i].SubscriptionID > id {
into = append(into, s.buf[i])
}
}

return into
}
```

## 完整流程

events 模块中最重要的就是 Default 和 BufferedSubscription 了，通过 go 插件可以看到调用方：

1. sub = events.Default.Subscribe 创建订阅者
2. bsub = events.NewBufferedSubscription 创建一个事件订阅缓冲区
3. events.Default.Log 产生事件
4. bsub.Since 方法获取缓冲区内的事件列表

The issue is a code walkthrough of syncthing's events pubsub model in lib/events/events.go. Start by reading the linked source file to understand EventType, Event, Logger, Subscription, and BufferedSubscription. Trace how Default.Subscribe and NewBufferedSubscription are used in the codebase. The goal is to document the flow of event creation, subscription, and buffering for educational purposes.

由索引模型根據 Issue 內容生成。