subscribe listens for new replica information!
(ctx context.Context)
| 178 | |
| 179 | // subscribe listens for new replica information! |
| 180 | func (m *Manager) subscribe(ctx context.Context) error { |
| 181 | var ( |
| 182 | needsUpdate = false |
| 183 | updating = false |
| 184 | updateMutex = sync.Mutex{} |
| 185 | ) |
| 186 | |
| 187 | // This loop will continually update nodes as updates are processed. |
| 188 | // The intent is to always be up to date without spamming the run |
| 189 | // function, so if a new update comes in while one is being processed, |
| 190 | // it will reprocess afterwards. |
| 191 | var update func() |
| 192 | update = func() { |
| 193 | err := m.syncReplicas(ctx) |
| 194 | if err != nil && !errors.Is(err, context.Canceled) { |
| 195 | m.logger.Warn(ctx, "run replica from subscribe", slog.Error(err)) |
| 196 | } |
| 197 | updateMutex.Lock() |
| 198 | if needsUpdate { |
| 199 | needsUpdate = false |
| 200 | updateMutex.Unlock() |
| 201 | update() |
| 202 | return |
| 203 | } |
| 204 | updating = false |
| 205 | updateMutex.Unlock() |
| 206 | } |
| 207 | cancelFunc, err := m.pubsub.Subscribe(PubsubEvent, func(_ context.Context, message []byte) { |
| 208 | updateMutex.Lock() |
| 209 | defer updateMutex.Unlock() |
| 210 | id, err := uuid.Parse(string(message)) |
| 211 | if err != nil { |
| 212 | return |
| 213 | } |
| 214 | // Don't process updates for ourself! |
| 215 | if id == m.id { |
| 216 | return |
| 217 | } |
| 218 | if updating { |
| 219 | needsUpdate = true |
| 220 | return |
| 221 | } |
| 222 | updating = true |
| 223 | go update() |
| 224 | }) |
| 225 | if err != nil { |
| 226 | return err |
| 227 | } |
| 228 | go func() { |
| 229 | <-ctx.Done() |
| 230 | cancelFunc() |
| 231 | }() |
| 232 | return nil |
| 233 | } |
| 234 | |
| 235 | func (m *Manager) syncReplicas(ctx context.Context) error { |
| 236 | m.closeMutex.Lock() |