mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2025-12-31 02:33:09 -06:00
Code Issues Packages Projects Releases Wiki Activity

Remove dependency on saramajetstream & sarama

Browse source 
Signed-off-by: Till Faelligen <tfaelligen@gmail.com>
This commit is contained in:
Till Faelligen 2022-02-01 19:04:58 +01:00
parent 893aa3b141
commit 8310cb83c9
1 changed files with 12 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
setup/jetstream/nats.go
View file

@ -5,20 +5,25 @@ import (
"sync" "sync"
"time" "time"
"github.com/Shopify/sarama"
"github.com/matrix-org/dendrite/setup/config" "github.com/matrix-org/dendrite/setup/config"
"github.com/sirupsen/logrus" "github.com/sirupsen/logrus"
saramajs "github.com/S7evinK/saramajetstream"
natsserver "github.com/nats-io/nats-server/v2/server" natsserver "github.com/nats-io/nats-server/v2/server"
"github.com/nats-io/nats.go" "github.com/nats-io/nats.go"
natsclient "github.com/nats-io/nats.go" natsclient "github.com/nats-io/nats.go"
) )
const (
// OffsetNewest tells e.g. the database to get the most current data
OffsetNewest int64 = -1
// OffsetOldest tells e.g. the database to get the oldest data
OffsetOldest int64 = -2
)
var natsServer *natsserver.Server var natsServer *natsserver.Server
var natsServerMutex sync.Mutex var natsServerMutex sync.Mutex
func Prepare(cfg *config.JetStream) (nats.JetStreamContext, sarama.Consumer, sarama.SyncProducer) { func Prepare(cfg *config.JetStream) natsclient.JetStreamContext {
// check if we need an in-process NATS Server // check if we need an in-process NATS Server
if len(cfg.Addresses) != 0 { if len(cfg.Addresses) != 0 {
return setupNATS(cfg, nil) return setupNATS(cfg, nil)
@ -52,20 +57,20 @@ func Prepare(cfg *config.JetStream) (nats.JetStreamContext, sarama.Consumer, sar
return setupNATS(cfg, nc) return setupNATS(cfg, nc)
} }
func setupNATS(cfg *config.JetStream, nc *natsclient.Conn) (nats.JetStreamContext, sarama.Consumer, sarama.SyncProducer) { func setupNATS(cfg *config.JetStream, nc *natsclient.Conn) natsclient.JetStreamContext {
if nc == nil { if nc == nil {
var err error var err error
nc, err = nats.Connect(strings.Join(cfg.Addresses, ",")) nc, err = nats.Connect(strings.Join(cfg.Addresses, ","))
if err != nil { if err != nil {
logrus.WithError(err).Panic("Unable to connect to NATS") logrus.WithError(err).Panic("Unable to connect to NATS")
return nil, nil, nil return nil
} }
} }
s, err := nc.JetStream() s, err := nc.JetStream()
if err != nil { if err != nil {
logrus.WithError(err).Panic("Unable to get JetStream context") logrus.WithError(err).Panic("Unable to get JetStream context")
return nil, nil, nil return nil
} }
for _, stream := range streams { // streams are defined in streams.go for _, stream := range streams { // streams are defined in streams.go
@ -93,7 +98,5 @@ func setupNATS(cfg *config.JetStream, nc *natsclient.Conn) (nats.JetStreamContex
} }
} }
consumer := saramajs.NewJetStreamConsumer(nc, s, "") return s
producer := saramajs.NewJetStreamProducer(nc, s, "")
return s, consumer, producer
} }