this repo has no description
remove unused MemPersister
-81
-81
cmd/relay/events/persist.go
-81
cmd/relay/events/persist.go
···
2
2
3
3
import (
4
4
"context"
5
-
"fmt"
6
-
"sync"
7
5
8
6
"github.com/bluesky-social/indigo/cmd/relay/models"
9
7
)
···
18
16
19
17
SetEventBroadcaster(func(*XRPCStreamEvent))
20
18
}
21
-
22
-
// MemPersister is the most naive implementation of event persistence
23
-
// This EventPersistence option works fine with all event types
24
-
// ill do better later
25
-
type MemPersister struct {
26
-
buf []*XRPCStreamEvent
27
-
lk sync.Mutex
28
-
seq int64
29
-
30
-
broadcast func(*XRPCStreamEvent)
31
-
}
32
-
33
-
func NewMemPersister() *MemPersister {
34
-
return &MemPersister{}
35
-
}
36
-
37
-
func (mp *MemPersister) Persist(ctx context.Context, e *XRPCStreamEvent) error {
38
-
mp.lk.Lock()
39
-
defer mp.lk.Unlock()
40
-
mp.seq++
41
-
switch {
42
-
case e.RepoCommit != nil:
43
-
e.RepoCommit.Seq = mp.seq
44
-
case e.RepoHandle != nil:
45
-
e.RepoHandle.Seq = mp.seq
46
-
case e.RepoIdentity != nil:
47
-
e.RepoIdentity.Seq = mp.seq
48
-
case e.RepoAccount != nil:
49
-
e.RepoAccount.Seq = mp.seq
50
-
case e.RepoMigrate != nil:
51
-
e.RepoMigrate.Seq = mp.seq
52
-
case e.RepoTombstone != nil:
53
-
e.RepoTombstone.Seq = mp.seq
54
-
case e.LabelLabels != nil:
55
-
e.LabelLabels.Seq = mp.seq
56
-
default:
57
-
panic("no event in persist call")
58
-
}
59
-
mp.buf = append(mp.buf, e)
60
-
61
-
mp.broadcast(e)
62
-
63
-
return nil
64
-
}
65
-
66
-
func (mp *MemPersister) Playback(ctx context.Context, since int64, cb func(*XRPCStreamEvent) error) error {
67
-
mp.lk.Lock()
68
-
l := len(mp.buf)
69
-
mp.lk.Unlock()
70
-
71
-
if since >= int64(l) {
72
-
return nil
73
-
}
74
-
75
-
// TODO: abusing the fact that buf[0].seq is currently always 1
76
-
for _, e := range mp.buf[since:l] {
77
-
if err := cb(e); err != nil {
78
-
return err
79
-
}
80
-
}
81
-
82
-
return nil
83
-
}
84
-
85
-
func (mp *MemPersister) TakeDownRepo(ctx context.Context, uid models.Uid) error {
86
-
return fmt.Errorf("repo takedowns not currently supported by memory persister, test usage only")
87
-
}
88
-
89
-
func (mp *MemPersister) Flush(ctx context.Context) error {
90
-
return nil
91
-
}
92
-
93
-
func (mp *MemPersister) SetEventBroadcaster(brc func(*XRPCStreamEvent)) {
94
-
mp.broadcast = brc
95
-
}
96
-
97
-
func (mp *MemPersister) Shutdown(context.Context) error {
98
-
return nil
99
-
}