this repo has no description
cleanup some unused functions
+2
-39
+2
-20
events/diskpersist.go
+2
-20
events/diskpersist.go
···
184
184
return nil
185
185
}
186
186
187
+
// swapLog swaps the current log file out for a new empty one
188
+
// must only be called while holding dp.lk
187
189
func (dp *DiskPersistence) swapLog(ctx context.Context) error {
188
190
if err := dp.logfi.Close(); err != nil {
189
191
return fmt.Errorf("failed to close current log file: %w", err)
···
258
260
)
259
261
260
262
var emptyHeader = make([]byte, headerSize)
261
-
262
-
func (p *DiskPersistence) addJobsToQueue(jobs []persistJob) error {
263
-
p.lk.Lock()
264
-
defer p.lk.Unlock()
265
-
266
-
for _, job := range jobs {
267
-
if err := p.doPersist(job); err != nil {
268
-
return err
269
-
}
270
-
271
-
// TODO: for some reason replacing this constant with p.writeBufferSize dramatically reduces perf...
272
-
if len(p.evtbuf) > 400 {
273
-
if err := p.flushLog(context.TODO()); err != nil {
274
-
return fmt.Errorf("failed to flush disk log: %w", err)
275
-
}
276
-
}
277
-
}
278
-
279
-
return nil
280
-
}
281
263
282
264
func (p *DiskPersistence) addJobToQueue(job persistJob) error {
283
265
p.lk.Lock()
-19
events/events.go
-19
events/events.go
···
78
78
}
79
79
}
80
80
81
-
type batchPersister interface {
82
-
PersistMany(ctx context.Context, evts []*XRPCStreamEvent) error
83
-
}
84
-
85
-
func (em *EventManager) persistAndSendEvents(ctx context.Context, evts []*XRPCStreamEvent) {
86
-
pm := em.persister.(batchPersister)
87
-
if err := pm.PersistMany(ctx, evts); err != nil {
88
-
log.Errorf("failed to persist outbound events: %s", err)
89
-
}
90
-
}
91
-
92
81
type Subscriber struct {
93
82
outgoing chan *XRPCStreamEvent
94
83
···
133
122
defer span.End()
134
123
135
124
em.persistAndSendEvent(ctx, ev)
136
-
return nil
137
-
}
138
-
139
-
func (em *EventManager) AddEventBatch(ctx context.Context, evs []*XRPCStreamEvent) error {
140
-
ctx, span := otel.Tracer("events").Start(ctx, "AddEventBatch")
141
-
defer span.End()
142
-
143
-
em.persistAndSendEvents(ctx, evs)
144
125
return nil
145
126
}
146
127