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

Use named NATS durable consumers

Browse source
This commit is contained in:
Neil Alexander 2022-01-07 16:18:01 +00:00
parent a422321435
commit c73a0d9a07
No known key found for this signature in database
GPG key ID: A02A2019A2BB0944
12 changed files with 42 additions and 22 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
appservice/consumers/roomserver.go
View file

@ -34,6 +34,7 @@ import (
type OutputRoomEventConsumer struct {
ctx context.Context
jetstream nats.JetStreamContext
durable nats.SubOpt
topic string
asDB storage.Database
rsAPI api.RoomserverInternalAPI
@ -54,7 +55,8 @@ func NewOutputRoomEventConsumer(
return &OutputRoomEventConsumer{
ctx: process.Context(),
jetstream: js,
topic: cfg.Global.JetStream.TopicFor(jetstream.OutputRoomEvent),
durable: nats.Durable(cfg.Global.JetStream.Namespaced("AppserviceRoomserverConsumer")),
topic: cfg.Global.JetStream.Namespaced(jetstream.OutputRoomEvent),
asDB: appserviceDB,
rsAPI: rsAPI,
serverName: string(cfg.Global.ServerName),
@ -64,7 +66,7 @@ func NewOutputRoomEventConsumer(
// Start consuming from room servers
func (s *OutputRoomEventConsumer) Start() error {
_, err := s.jetstream.Subscribe(s.topic, s.onMessage)
_, err := s.jetstream.Subscribe(s.topic, s.onMessage, s.durable)
return err
}

14
federationapi/consumers/eduserver.go
View file

@ -34,6 +34,7 @@ import (
type OutputEDUConsumer struct {
ctx context.Context
jetstream nats.JetStreamContext
durable nats.SubOpt
db storage.Database
queues *queue.OutgoingQueues
ServerName gomatrixserverlib.ServerName
@ -56,21 +57,22 @@ func NewOutputEDUConsumer(
queues: queues,
db: store,
ServerName: cfg.Matrix.ServerName,
typingTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputTypingEvent),
sendToDeviceTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputSendToDeviceEvent),
receiptTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputReceiptEvent),
durable: nats.Durable(cfg.Matrix.JetStream.Namespaced("FederationAPIEDUServerConsumer")),
typingTopic: cfg.Matrix.JetStream.Namespaced(jetstream.OutputTypingEvent),
sendToDeviceTopic: cfg.Matrix.JetStream.Namespaced(jetstream.OutputSendToDeviceEvent),
receiptTopic: cfg.Matrix.JetStream.Namespaced(jetstream.OutputReceiptEvent),
}
}
// Start consuming from EDU servers
func (t *OutputEDUConsumer) Start() error {
if _, err := t.jetstream.Subscribe(t.typingTopic, t.onTypingEvent); err != nil {
if _, err := t.jetstream.Subscribe(t.typingTopic, t.onTypingEvent, t.durable); err != nil {
return err
}
if _, err := t.jetstream.Subscribe(t.sendToDeviceTopic, t.onSendToDeviceEvent); err != nil {
if _, err := t.jetstream.Subscribe(t.sendToDeviceTopic, t.onSendToDeviceEvent, t.durable); err != nil {
return err
}
if _, err := t.jetstream.Subscribe(t.receiptTopic, t.onReceiptEvent); err != nil {
if _, err := t.jetstream.Subscribe(t.receiptTopic, t.onReceiptEvent, t.durable); err != nil {
return err
}
return nil

2
federationapi/consumers/keychange.go
View file

@ -57,7 +57,7 @@ func NewKeyChangeConsumer(
consumer: &internal.ContinualConsumer{
Process: process,
ComponentName: "federationapi/keychange",
Topic: string(cfg.Matrix.JetStream.TopicFor(jetstream.OutputKeyChangeEvent)),
Topic: string(cfg.Matrix.JetStream.Namespaced(jetstream.OutputKeyChangeEvent)),
Consumer: kafkaConsumer,
PartitionStore: store,
},

6
federationapi/consumers/roomserver.go
View file

@ -37,6 +37,7 @@ type OutputRoomEventConsumer struct {
cfg *config.FederationAPI
rsAPI api.RoomserverInternalAPI
jetstream nats.JetStreamContext
durable nats.SubOpt
db storage.Database
queues *queue.OutgoingQueues
topic string
@ -58,13 +59,14 @@ func NewOutputRoomEventConsumer(
db: store,
queues: queues,
rsAPI: rsAPI,
topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputRoomEvent),
durable: nats.Durable(cfg.Matrix.JetStream.Namespaced("FederationAPIRoomServerConsumer")),
topic: cfg.Matrix.JetStream.Namespaced(jetstream.OutputRoomEvent),
}
}
// Start consuming from room servers
func (s *OutputRoomEventConsumer) Start() error {
_, err := s.jetstream.Subscribe(s.topic, s.onMessage)
_, err := s.jetstream.Subscribe(s.topic, s.onMessage, s.durable)
return err
}

1
roomserver/internal/api.go
View file

@ -67,6 +67,7 @@ func NewRoomserverAPI(
InputRoomEventTopic: inputRoomEventTopic,
OutputRoomEventTopic: outputRoomEventTopic,
JetStream: consumer,
Durable: cfg.Matrix.JetStream.Namespaced("RoomserverInputConsumer"),
ServerName: cfg.Matrix.ServerName,
ACLs: serverACLs,
},

3
roomserver/internal/input/input.go
View file

@ -43,6 +43,7 @@ var keyContentFields = map[string]string{
type Inputer struct {
DB storage.Database
JetStream nats.JetStreamContext
Durable nats.SubOpt
ServerName gomatrixserverlib.ServerName
ACLs *acls.ServerACLs
InputRoomEventTopic string
@ -85,6 +86,8 @@ func (r *Inputer) Start() error {
// or nak them within a certain amount of time. This stops that from
// happening, so we don't end up doing a lot of unnecessary duplicate work.
nats.MaxDeliver(0),
// Use a durable named consumer.
r.Durable,
)
return err
}

2
setup/config/config_jetstream.go
View file

@ -19,7 +19,7 @@ type JetStream struct {
InMemory bool `yaml:"in_memory"`
}
func (c *JetStream) TopicFor(name string) string {
func (c *JetStream) Namespaced(name string) string {
return fmt.Sprintf("%s%s", c.TopicPrefix, name)
}

6
syncapi/consumers/clientapi.go
View file

@ -34,6 +34,7 @@ import (
type OutputClientDataConsumer struct {
ctx context.Context
jetstream nats.JetStreamContext
durable nats.SubOpt
topic string
db storage.Database
stream types.StreamProvider
@ -52,7 +53,8 @@ func NewOutputClientDataConsumer(
return &OutputClientDataConsumer{
ctx: process.Context(),
jetstream: js,
topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputClientData),
topic: cfg.Matrix.JetStream.Namespaced(jetstream.OutputClientData),
durable: nats.Durable(cfg.Matrix.JetStream.Namespaced("SyncAPIClientAPIConsumer")),
db: store,
notifier: notifier,
stream: stream,
@ -61,7 +63,7 @@ func NewOutputClientDataConsumer(
// Start consuming from room servers
func (s *OutputClientDataConsumer) Start() error {
_, err := s.jetstream.Subscribe(s.topic, s.onMessage)
_, err := s.jetstream.Subscribe(s.topic, s.onMessage, s.durable)
return err
}

6
syncapi/consumers/eduserver_receipts.go
View file

@ -34,6 +34,7 @@ import (
type OutputReceiptEventConsumer struct {
ctx context.Context
jetstream nats.JetStreamContext
durable nats.SubOpt
topic string
db storage.Database
stream types.StreamProvider
@ -53,7 +54,8 @@ func NewOutputReceiptEventConsumer(
return &OutputReceiptEventConsumer{
ctx: process.Context(),
jetstream: js,
topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputReceiptEvent),
topic: cfg.Matrix.JetStream.Namespaced(jetstream.OutputReceiptEvent),
durable: nats.Durable(cfg.Matrix.JetStream.Namespaced("SyncAPIEDUServerReceiptConsumer")),
db: store,
notifier: notifier,
stream: stream,
@ -62,7 +64,7 @@ func NewOutputReceiptEventConsumer(
// Start consuming from EDU api
func (s *OutputReceiptEventConsumer) Start() error {
_, err := s.jetstream.Subscribe(s.topic, s.onMessage)
_, err := s.jetstream.Subscribe(s.topic, s.onMessage, s.durable)
return err
}

6
syncapi/consumers/eduserver_sendtodevice.go
View file

@ -36,6 +36,7 @@ import (
type OutputSendToDeviceEventConsumer struct {
ctx context.Context
jetstream nats.JetStreamContext
durable nats.SubOpt
topic string
db storage.Database
serverName gomatrixserverlib.ServerName // our server name
@ -56,7 +57,8 @@ func NewOutputSendToDeviceEventConsumer(
return &OutputSendToDeviceEventConsumer{
ctx: process.Context(),
jetstream: js,
topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputSendToDeviceEvent),
topic: cfg.Matrix.JetStream.Namespaced(jetstream.OutputSendToDeviceEvent),
durable: nats.Durable(cfg.Matrix.JetStream.Namespaced("SyncAPIEDUServerSendToDeviceConsumer")),
db: store,
serverName: cfg.Matrix.ServerName,
notifier: notifier,
@ -66,7 +68,7 @@ func NewOutputSendToDeviceEventConsumer(
// Start consuming from EDU api
func (s *OutputSendToDeviceEventConsumer) Start() error {
_, err := s.jetstream.Subscribe(s.topic, s.onMessage)
_, err := s.jetstream.Subscribe(s.topic, s.onMessage, s.durable)
return err
}

6
syncapi/consumers/eduserver_typing.go
View file

@ -35,6 +35,7 @@ import (
type OutputTypingEventConsumer struct {
ctx context.Context
jetstream nats.JetStreamContext
durable nats.SubOpt
topic string
eduCache *cache.EDUCache
stream types.StreamProvider
@ -55,7 +56,8 @@ func NewOutputTypingEventConsumer(
return &OutputTypingEventConsumer{
ctx: process.Context(),
jetstream: js,
topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputTypingEvent),
topic: cfg.Matrix.JetStream.Namespaced(jetstream.OutputTypingEvent),
durable: nats.Durable(cfg.Matrix.JetStream.Namespaced("SyncAPIEDUServerTypingConsumer")),
eduCache: eduCache,
notifier: notifier,
stream: stream,
@ -64,7 +66,7 @@ func NewOutputTypingEventConsumer(
// Start consuming from EDU api
func (s *OutputTypingEventConsumer) Start() error {
_, err := s.jetstream.Subscribe(s.topic, s.onMessage)
_, err := s.jetstream.Subscribe(s.topic, s.onMessage, s.durable)
return err
}

6
syncapi/consumers/roomserver.go
View file

@ -38,6 +38,7 @@ type OutputRoomEventConsumer struct {
cfg *config.SyncAPI
rsAPI api.RoomserverInternalAPI
jetstream nats.JetStreamContext
durable nats.SubOpt
topic string
db storage.Database
pduStream types.StreamProvider
@ -60,7 +61,8 @@ func NewOutputRoomEventConsumer(
ctx: process.Context(),
cfg: cfg,
jetstream: js,
topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputRoomEvent),
topic: cfg.Matrix.JetStream.Namespaced(jetstream.OutputRoomEvent),
durable: nats.Durable(cfg.Matrix.JetStream.Namespaced("SyncAPIRoomServerConsumer")),
db: store,
notifier: notifier,
pduStream: pduStream,
@ -71,7 +73,7 @@ func NewOutputRoomEventConsumer(
// Start consuming from room servers
func (s *OutputRoomEventConsumer) Start() error {
_, err := s.jetstream.Subscribe(s.topic, s.onMessage)
_, err := s.jetstream.Subscribe(s.topic, s.onMessage, s.durable)
return err
}