awareness.go
1package sync
2
3import (
4 "encoding/json"
5 "sync"
6 "time"
7)
8
9// AwarenessState represents the state of a single client in the awareness protocol.
10// It contains the client's ID, a clock for ordering updates, and arbitrary state data.
11type AwarenessState struct {
12 ClientID uint64
13 Clock uint64
14 State map[string]interface{}
15}
16
17// AwarenessClient tracks connected clients using the y-sweet awareness protocol.
18// It maintains a map of all known clients and triggers a callback when no other
19// clients remain after a timeout period.
20type AwarenessClient struct {
21 clientID uint64
22 clients map[uint64]AwarenessState
23 mu sync.RWMutex
24 lastUpdate time.Time
25 timeout time.Duration
26 onNoOtherClients func()
27 timer *time.Timer
28}
29
30// NewAwarenessClient creates a new awareness client with the given local client ID.
31// The onNoOtherClients callback is triggered when all other clients disconnect
32// and the specified timeout expires.
33func NewAwarenessClient(clientID uint64, timeout time.Duration, onNoOtherClients func()) *AwarenessClient {
34 if timeout <= 0 {
35 timeout = 5 * time.Minute // default
36 }
37 c := &AwarenessClient{
38 clientID: clientID,
39 clients: make(map[uint64]AwarenessState),
40 lastUpdate: time.Now(),
41 timeout: timeout,
42 onNoOtherClients: onNoOtherClients,
43 }
44 c.resetTimer()
45 return c
46}
47
48// States returns a snapshot of all known client states.
49// This includes the local client and any remote clients.
50func (c *AwarenessClient) States() []AwarenessState {
51 c.mu.RLock()
52 defer c.mu.RUnlock()
53
54 states := make([]AwarenessState, 0, len(c.clients))
55 for _, state := range c.clients {
56 states = append(states, state)
57 }
58 return states
59}
60
61func (c *AwarenessClient) resetTimer() {
62 if c.timer != nil {
63 c.timer.Stop()
64 }
65 c.timer = time.AfterFunc(c.timeout, func() {
66 c.checkTimeout()
67 })
68}
69
70func (c *AwarenessClient) checkTimeout() {
71 c.mu.Lock()
72 defer c.mu.Unlock()
73
74 if len(c.clients) == 0 && c.onNoOtherClients != nil {
75 c.onNoOtherClients()
76 }
77}
78
79// HandleUpdate processes an awareness update from the server.
80// It parses the update data and updates the local client map.
81// The timeout is reset whenever an update is received.
82func (c *AwarenessClient) HandleUpdate(data []byte) error {
83 c.mu.Lock()
84 defer c.mu.Unlock()
85
86 if len(data) == 0 {
87 return nil
88 }
89
90 count, n, err := readVarUint(data)
91 if err != nil {
92 return err
93 }
94
95 offset := n
96 c.clients = make(map[uint64]AwarenessState)
97
98 for i := uint64(0); i < count; i++ {
99 if offset >= len(data) {
100 return nil
101 }
102
103 clientID, n, err := readVarUint(data[offset:])
104 if err != nil {
105 return err
106 }
107 offset += n
108
109 if offset >= len(data) {
110 return nil
111 }
112
113 clock, n, err := readVarUint(data[offset:])
114 if err != nil {
115 return err
116 }
117 offset += n
118
119 if offset >= len(data) {
120 return nil
121 }
122
123 stateJSON := data[offset:]
124 var state map[string]interface{}
125 if err := json.Unmarshal(stateJSON, &state); err != nil {
126 // Non-JSON awareness state is valid (e.g., empty or binary)
127 // Treat as empty state rather than failing
128 state = nil
129 }
130
131 c.clients[clientID] = AwarenessState{
132 ClientID: clientID,
133 Clock: clock,
134 State: state,
135 }
136
137 offset = len(data)
138 }
139
140 c.lastUpdate = time.Now()
141 c.resetTimer()
142
143 return nil
144}
145
146// ClientID returns the local client ID for this awareness client.
147func (c *AwarenessClient) ClientID() uint64 {
148 c.mu.RLock()
149 defer c.mu.RUnlock()
150 return c.clientID
151}
