this repo has no description
no waits between loading pages
+9
-8
+9
-8
cmd/nexus/outbox.go
+9
-8
cmd/nexus/outbox.go
···
96
96
97
97
// continuously load events from DB into memory cache
98
98
func (o *Outbox) runCacheLoader(ctx context.Context) {
99
-
ticker := time.NewTicker(10 * time.Millisecond)
100
-
defer ticker.Stop()
101
-
102
99
for {
103
100
select {
104
101
case <-ctx.Done():
105
102
return
106
-
case <-ticker.C:
107
-
o.loadMoreEvents()
103
+
default:
104
+
loaded := o.loadMoreEvents()
105
+
if !loaded {
106
+
time.Sleep(500 * time.Millisecond)
107
+
}
108
108
}
109
109
}
110
110
}
111
111
112
-
func (o *Outbox) loadMoreEvents() {
112
+
func (o *Outbox) loadMoreEvents() bool {
113
113
o.cacheMu.RLock()
114
114
lastID := o.lastLoadedID
115
115
o.cacheMu.RUnlock()
···
122
122
Limit(batchSize).
123
123
Find(&events).Error; err != nil {
124
124
o.logger.Error("failed to load events into cache", "error", err, "lastID", lastID)
125
-
return
125
+
return false
126
126
}
127
127
128
128
if len(events) == 0 {
129
-
return
129
+
return false
130
130
}
131
131
132
132
o.cacheMu.Lock()
···
151
151
// back pressure if pendingIDs channel is full
152
152
o.pendingIDs <- dbEvt.ID
153
153
}
154
+
return true
154
155
}
155
156
156
157
// runDelivery continuously pulls from pendingIDs and delivers events