Triggers addPendingLogs if it hasn't been run in CachePendingSeqMaxWait. Error returned to fulfil BackgroundTaskFunc signature.
(ctx context.Context)
| 268 | |
| 269 | // Triggers addPendingLogs if it hasn't been run in CachePendingSeqMaxWait. Error returned to fulfil BackgroundTaskFunc signature. |
| 270 | func (c *changeCache) InsertPendingEntries(ctx context.Context) error { |
| 271 | |
| 272 | lastAddPendingLogsTime := atomic.LoadInt64(&c.lastAddPendingTime) |
| 273 | if time.Since(time.Unix(0, lastAddPendingLogsTime)) < c.options.CachePendingSeqMaxWait { |
| 274 | return nil |
| 275 | } |
| 276 | |
| 277 | // Trigger _addPendingLogs to process any entries that have been pending too long: |
| 278 | c.lock.Lock() |
| 279 | changedChannels := c._addPendingLogs(ctx) |
| 280 | c.lock.Unlock() |
| 281 | |
| 282 | c.notifyChange(ctx, changedChannels) |
| 283 | |
| 284 | return nil |
| 285 | } |
| 286 | |
| 287 | // Cleanup function, invoked periodically. |
| 288 | // Removes skipped entries from skippedSeqs that have been waiting longer |