this repo has no description
only one Write call per event
+11
-3
+11
-3
events/diskpersist.go
+11
-3
events/diskpersist.go
···
226
226
evtKindHandle = 2
227
227
)
228
228
229
+
var emptyHeader = make([]byte, headerSize)
230
+
229
231
func (p *DiskPersistence) Persist(ctx context.Context, e *XRPCStreamEvent) error {
230
232
buffer := p.buffers.Get().(*bytes.Buffer)
231
233
defer p.buffers.Put(buffer)
232
234
233
235
buffer.Truncate(0)
236
+
237
+
buffer.Write(emptyHeader)
234
238
235
239
var evtKind uint32
236
240
switch {
···
249
253
// only those two get peristed right now
250
254
}
251
255
256
+
b := buffer.Bytes()
257
+
258
+
binary.LittleEndian.PutUint32(b, 0)
259
+
binary.LittleEndian.PutUint32(b[4:], evtKind)
260
+
binary.LittleEndian.PutUint32(b[8:], uint32(len(b)-headerSize))
261
+
252
262
p.seqLk.Lock()
253
263
defer p.seqLk.Unlock()
254
264
···
265
275
// only those two get peristed right now
266
276
}
267
277
268
-
if err := p.writeHeader(ctx, 0, evtKind, uint32(buffer.Len()), seq); err != nil {
269
-
return fmt.Errorf("failed to write header: %w", err)
270
-
}
278
+
binary.LittleEndian.PutUint64(b[12:], uint64(seq))
271
279
272
280
if _, err := io.Copy(p.logfi, buffer); err != nil {
273
281
return err