this repo has no description
0
fork

Configure Feed

Select the types of activity you want to include in your feed.

connect rainbow to pebble persister

+37 -2
+20 -1
cmd/rainbow/main.go
··· 53 53 Value: "bsky.network", 54 54 }, 55 55 &cli.StringFlag{ 56 + Name: "persist-db", 57 + Value: "", 58 + Usage: "path to persistence db", 59 + }, 60 + &cli.StringFlag{ 56 61 Name: "api-listen", 57 62 Value: ":2480", 58 63 }, ··· 110 115 otel.SetTracerProvider(tp) 111 116 } 112 117 113 - spl := splitter.NewSplitter(cctx.String("splitter-host")) 118 + persistPath := cctx.String("persist-db") 119 + upstreamHost := cctx.String("splitter-host") 120 + var spl *splitter.Splitter 121 + var err error 122 + if persistPath != "" { 123 + log.Infof("building splitter with storage at: %s", persistPath) 124 + spl, err = splitter.NewDiskSplitter(upstreamHost, persistPath) 125 + if err != nil { 126 + log.Fatalw("failed to create splitter", "path", persistPath, "error", err) 127 + return err 128 + } 129 + } else { 130 + log.Info("building in-memory splitter") 131 + spl = splitter.NewMemSplitter(upstreamHost) 132 + } 114 133 115 134 // set up metrics endpoint 116 135 go func() {
+17 -1
splitter/splitter.go
··· 30 30 type Splitter struct { 31 31 Host string 32 32 erb *EventRingBuffer 33 + pp *events.PebblePersist 33 34 events *events.EventManager 34 35 35 36 // cursor storage ··· 41 42 consumers map[uint64]*SocketConsumer 42 43 } 43 44 44 - func NewSplitter(host string) *Splitter { 45 + func NewMemSplitter(host string) *Splitter { 45 46 erb := NewEventRingBuffer(20_000, 10_000) 46 47 47 48 em := events.NewEventManager(erb) ··· 52 53 events: em, 53 54 consumers: make(map[uint64]*SocketConsumer), 54 55 } 56 + } 57 + func NewDiskSplitter(host, path string) (*Splitter, error) { 58 + pp, err := events.NewPebblePersistance(path) 59 + if err != nil { 60 + return nil, err 61 + } 62 + 63 + em := events.NewEventManager(pp) 64 + return &Splitter{ 65 + cursorFile: "cursor-file", 66 + Host: host, 67 + pp: pp, 68 + events: em, 69 + consumers: make(map[uint64]*SocketConsumer), 70 + }, nil 55 71 } 56 72 57 73 func (s *Splitter) Start(addr string) error {