dendrite/setup/jetstream/nats.go
Neil Alexander 16035b9737
NATS JetStream tweaks (#2086)
* Use named NATS durable consumers

* Build fixes

* Remove dupe call to SetFederationAPI

* Use namespaced consumer name

* Fix namespacing

* Fix unit tests hopefully
2022-01-07 17:31:57 +00:00

99 lines
2.7 KiB
Go

package jetstream
import (
"strings"
"sync"
"time"
"github.com/Shopify/sarama"
"github.com/matrix-org/dendrite/setup/config"
"github.com/sirupsen/logrus"
saramajs "github.com/S7evinK/saramajetstream"
natsserver "github.com/nats-io/nats-server/v2/server"
"github.com/nats-io/nats.go"
natsclient "github.com/nats-io/nats.go"
)
var natsServer *natsserver.Server
var natsServerMutex sync.Mutex
func Prepare(cfg *config.JetStream) (nats.JetStreamContext, sarama.Consumer, sarama.SyncProducer) {
// check if we need an in-process NATS Server
if len(cfg.Addresses) != 0 {
return setupNATS(cfg, nil)
}
natsServerMutex.Lock()
if natsServer == nil {
var err error
natsServer, err = natsserver.NewServer(&natsserver.Options{
ServerName: "monolith",
DontListen: true,
JetStream: true,
StoreDir: string(cfg.StoragePath),
NoSystemAccount: true,
AllowNewAccounts: false,
})
if err != nil {
panic(err)
}
natsServer.ConfigureLogger()
go natsServer.Start()
}
natsServerMutex.Unlock()
if !natsServer.ReadyForConnections(time.Second * 10) {
logrus.Fatalln("NATS did not start in time")
}
nc, err := natsclient.Connect("", natsclient.InProcessServer(natsServer))
if err != nil {
logrus.Fatalln("Failed to create NATS client")
}
return setupNATS(cfg, nc)
}
func setupNATS(cfg *config.JetStream, nc *natsclient.Conn) (nats.JetStreamContext, sarama.Consumer, sarama.SyncProducer) {
if nc == nil {
var err error
nc, err = nats.Connect(strings.Join(cfg.Addresses, ","))
if err != nil {
logrus.WithError(err).Panic("Unable to connect to NATS")
return nil, nil, nil
}
}
s, err := nc.JetStream()
if err != nil {
logrus.WithError(err).Panic("Unable to get JetStream context")
return nil, nil, nil
}
for _, stream := range streams { // streams are defined in streams.go
name := cfg.TopicFor(stream.Name)
info, err := s.StreamInfo(name)
if err != nil && err != natsclient.ErrStreamNotFound {
logrus.WithError(err).Fatal("Unable to get stream info")
}
if info == nil {
stream.Subjects = []string{name}
// If we're trying to keep everything in memory (e.g. unit tests)
// then overwrite the storage policy.
if cfg.InMemory {
stream.Storage = nats.MemoryStorage
}
// Namespace the streams without modifying the original streams
// array, otherwise we end up with namespaces on namespaces.
namespaced := *stream
namespaced.Name = name
if _, err = s.AddStream(&namespaced); err != nil {
logrus.WithError(err).WithField("stream", name).Fatal("Unable to add stream")
}
}
}
consumer := saramajs.NewJetStreamConsumer(nc, s, "")
producer := saramajs.NewJetStreamProducer(nc, s, "")
return s, consumer, producer
}