From b49160f1f1457512ab9c7905055ec14907e303f7 Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Tue, 22 Mar 2022 13:56:02 +0000 Subject: [PATCH] Rename `TopicFor` to `Prefixed` --- appservice/consumers/roomserver.go | 2 +- clientapi/clientapi.go | 2 +- eduserver/eduserver.go | 6 ++-- federationapi/consumers/eduserver.go | 6 ++-- federationapi/consumers/keychange.go | 4 +-- federationapi/consumers/roomserver.go | 2 +- keyserver/keyserver.go | 2 +- roomserver/internal/input/input.go | 31 +++++++++------------ roomserver/roomserver.go | 4 +-- setup/config/config_jetstream.go | 4 +-- setup/jetstream/nats.go | 2 +- syncapi/consumers/clientapi.go | 2 +- syncapi/consumers/eduserver_receipts.go | 2 +- syncapi/consumers/eduserver_sendtodevice.go | 2 +- syncapi/consumers/eduserver_typing.go | 2 +- syncapi/consumers/roomserver.go | 2 +- syncapi/consumers/userapi.go | 2 +- syncapi/syncapi.go | 6 ++-- userapi/consumers/syncapi_readupdate.go | 2 +- userapi/consumers/syncapi_streamevent.go | 2 +- userapi/userapi.go | 4 +-- 21 files changed, 43 insertions(+), 48 deletions(-) diff --git a/appservice/consumers/roomserver.go b/appservice/consumers/roomserver.go index 9d723bed1..01790722a 100644 --- a/appservice/consumers/roomserver.go +++ b/appservice/consumers/roomserver.go @@ -56,7 +56,7 @@ func NewOutputRoomEventConsumer( ctx: process.Context(), jetstream: js, durable: cfg.Global.JetStream.Durable("AppserviceRoomserverConsumer"), - topic: cfg.Global.JetStream.TopicFor(jetstream.OutputRoomEvent), + topic: cfg.Global.JetStream.Prefixed(jetstream.OutputRoomEvent), asDB: appserviceDB, rsAPI: rsAPI, serverName: string(cfg.Global.ServerName), diff --git a/clientapi/clientapi.go b/clientapi/clientapi.go index 75184d3b7..4550343c7 100644 --- a/clientapi/clientapi.go +++ b/clientapi/clientapi.go @@ -55,7 +55,7 @@ func AddPublicRoutes( syncProducer := &producers.SyncAPIProducer{ JetStream: js, - Topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputClientData), + Topic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputClientData), } routing.Setup( diff --git a/eduserver/eduserver.go b/eduserver/eduserver.go index 9fe8704cf..91208a400 100644 --- a/eduserver/eduserver.go +++ b/eduserver/eduserver.go @@ -48,9 +48,9 @@ func NewInternalAPI( Cache: eduCache, UserAPI: userAPI, JetStream: js, - OutputTypingEventTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputTypingEvent), - OutputSendToDeviceEventTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputSendToDeviceEvent), - OutputReceiptEventTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputReceiptEvent), + OutputTypingEventTopic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputTypingEvent), + OutputSendToDeviceEventTopic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputSendToDeviceEvent), + OutputReceiptEventTopic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputReceiptEvent), ServerName: cfg.Matrix.ServerName, } } diff --git a/federationapi/consumers/eduserver.go b/federationapi/consumers/eduserver.go index 1f81fa258..e14e60f47 100644 --- a/federationapi/consumers/eduserver.go +++ b/federationapi/consumers/eduserver.go @@ -58,9 +58,9 @@ func NewOutputEDUConsumer( db: store, ServerName: cfg.Matrix.ServerName, durable: cfg.Matrix.JetStream.Durable("FederationAPIEDUServerConsumer"), - typingTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputTypingEvent), - sendToDeviceTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputSendToDeviceEvent), - receiptTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputReceiptEvent), + typingTopic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputTypingEvent), + sendToDeviceTopic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputSendToDeviceEvent), + receiptTopic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputReceiptEvent), } } diff --git a/federationapi/consumers/keychange.go b/federationapi/consumers/keychange.go index 33d716d25..94e454359 100644 --- a/federationapi/consumers/keychange.go +++ b/federationapi/consumers/keychange.go @@ -55,8 +55,8 @@ func NewKeyChangeConsumer( return &KeyChangeConsumer{ ctx: process.Context(), jetstream: js, - durable: cfg.Matrix.JetStream.TopicFor("FederationAPIKeyChangeConsumer"), - topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputKeyChangeEvent), + durable: cfg.Matrix.JetStream.Prefixed("FederationAPIKeyChangeConsumer"), + topic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputKeyChangeEvent), queues: queues, db: store, serverName: cfg.Matrix.ServerName, diff --git a/federationapi/consumers/roomserver.go b/federationapi/consumers/roomserver.go index 989f7cf49..ff2c8e5d4 100644 --- a/federationapi/consumers/roomserver.go +++ b/federationapi/consumers/roomserver.go @@ -61,7 +61,7 @@ func NewOutputRoomEventConsumer( queues: queues, rsAPI: rsAPI, durable: cfg.Matrix.JetStream.Durable("FederationAPIRoomServerConsumer"), - topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputRoomEvent), + topic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputRoomEvent), } } diff --git a/keyserver/keyserver.go b/keyserver/keyserver.go index cf66bd387..c557dfbaa 100644 --- a/keyserver/keyserver.go +++ b/keyserver/keyserver.go @@ -46,7 +46,7 @@ func NewInternalAPI( logrus.WithError(err).Panicf("failed to connect to key server database") } keyChangeProducer := &producers.KeyChange{ - Topic: string(cfg.Matrix.JetStream.TopicFor(jetstream.OutputKeyChangeEvent)), + Topic: string(cfg.Matrix.JetStream.Prefixed(jetstream.OutputKeyChangeEvent)), JetStream: js, DB: db, } diff --git a/roomserver/internal/input/input.go b/roomserver/internal/input/input.go index d267599ec..beae0f80d 100644 --- a/roomserver/internal/input/input.go +++ b/roomserver/internal/input/input.go @@ -78,26 +78,21 @@ func (r *Inputer) startWorkerForRoom(roomID string) { }) w := v.(*worker) if !loaded { - consumer := "DendriteRoomInputConsumerPull" + jetstream.Tokenise(w.roomID) + consumer := r.Cfg.Matrix.JetStream.Prefixed("RoomInput" + jetstream.Tokenise(w.roomID)) subject := jetstream.InputRoomEventSubj(w.roomID) - if info, err := w.r.JetStream.ConsumerInfo( - jetstream.InputRoomEvent, - consumer, - ); err != nil || info == nil { - if _, err := w.r.JetStream.AddConsumer( - r.Cfg.Matrix.JetStream.TopicFor(jetstream.InputRoomEvent), - &nats.ConsumerConfig{ - Durable: consumer, - AckPolicy: nats.AckExplicitPolicy, - DeliverPolicy: nats.DeliverAllPolicy, - FilterSubject: subject, - AckWait: MaximumMissingProcessingTime + (time.Second * 10), - }, - ); err != nil { - logrus.WithError(err).Errorf("Failed to create consumer for room %q", w.roomID) - return - } + if _, err := w.r.JetStream.AddConsumer( + r.Cfg.Matrix.JetStream.Prefixed(jetstream.InputRoomEvent), + &nats.ConsumerConfig{ + Durable: consumer, + AckPolicy: nats.AckAllPolicy, + DeliverPolicy: nats.DeliverAllPolicy, + FilterSubject: subject, + AckWait: MaximumMissingProcessingTime + (time.Second * 10), + }, + ); err != nil { + logrus.WithError(err).Errorf("Failed to create consumer for room %q", w.roomID) + return } sub, err := w.r.JetStream.PullSubscribe( diff --git a/roomserver/roomserver.go b/roomserver/roomserver.go index 896773bab..36e3c5269 100644 --- a/roomserver/roomserver.go +++ b/roomserver/roomserver.go @@ -54,8 +54,8 @@ func NewInternalAPI( return internal.NewRoomserverAPI( base.ProcessContext, cfg, roomserverDB, js, nc, - cfg.Matrix.JetStream.TopicFor(jetstream.InputRoomEvent), - cfg.Matrix.JetStream.TopicFor(jetstream.OutputRoomEvent), + cfg.Matrix.JetStream.Prefixed(jetstream.InputRoomEvent), + cfg.Matrix.JetStream.Prefixed(jetstream.OutputRoomEvent), base.Caches, perspectiveServerNames, ) } diff --git a/setup/config/config_jetstream.go b/setup/config/config_jetstream.go index 9271cd8b4..b6a93d398 100644 --- a/setup/config/config_jetstream.go +++ b/setup/config/config_jetstream.go @@ -19,12 +19,12 @@ type JetStream struct { InMemory bool `yaml:"in_memory"` } -func (c *JetStream) TopicFor(name string) string { +func (c *JetStream) Prefixed(name string) string { return fmt.Sprintf("%s%s", c.TopicPrefix, name) } func (c *JetStream) Durable(name string) string { - return c.TopicFor(name) + return c.Prefixed(name) } func (c *JetStream) Defaults(generate bool) { diff --git a/setup/jetstream/nats.go b/setup/jetstream/nats.go index f42055ec7..9afdf490e 100644 --- a/setup/jetstream/nats.go +++ b/setup/jetstream/nats.go @@ -75,7 +75,7 @@ func setupNATS(cfg *config.JetStream, nc *natsclient.Conn) (natsclient.JetStream } for _, stream := range streams { // streams are defined in streams.go - name := cfg.TopicFor(stream.Name) + name := cfg.Prefixed(stream.Name) info, err := s.StreamInfo(name) if err != nil && err != natsclient.ErrStreamNotFound { logrus.WithError(err).Fatal("Unable to get stream info") diff --git a/syncapi/consumers/clientapi.go b/syncapi/consumers/clientapi.go index fcb7b5b1c..40c1cd3d6 100644 --- a/syncapi/consumers/clientapi.go +++ b/syncapi/consumers/clientapi.go @@ -61,7 +61,7 @@ func NewOutputClientDataConsumer( return &OutputClientDataConsumer{ ctx: process.Context(), jetstream: js, - topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputClientData), + topic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputClientData), durable: cfg.Matrix.JetStream.Durable("SyncAPIClientAPIConsumer"), db: store, notifier: notifier, diff --git a/syncapi/consumers/eduserver_receipts.go b/syncapi/consumers/eduserver_receipts.go index 4e4c61c67..ab79998ea 100644 --- a/syncapi/consumers/eduserver_receipts.go +++ b/syncapi/consumers/eduserver_receipts.go @@ -62,7 +62,7 @@ func NewOutputReceiptEventConsumer( return &OutputReceiptEventConsumer{ ctx: process.Context(), jetstream: js, - topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputReceiptEvent), + topic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputReceiptEvent), durable: cfg.Matrix.JetStream.Durable("SyncAPIEDUServerReceiptConsumer"), db: store, notifier: notifier, diff --git a/syncapi/consumers/eduserver_sendtodevice.go b/syncapi/consumers/eduserver_sendtodevice.go index b0beef063..bdbe77352 100644 --- a/syncapi/consumers/eduserver_sendtodevice.go +++ b/syncapi/consumers/eduserver_sendtodevice.go @@ -57,7 +57,7 @@ func NewOutputSendToDeviceEventConsumer( return &OutputSendToDeviceEventConsumer{ ctx: process.Context(), jetstream: js, - topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputSendToDeviceEvent), + topic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputSendToDeviceEvent), durable: cfg.Matrix.JetStream.Durable("SyncAPIEDUServerSendToDeviceConsumer"), db: store, serverName: cfg.Matrix.ServerName, diff --git a/syncapi/consumers/eduserver_typing.go b/syncapi/consumers/eduserver_typing.go index cae5df8a8..c2828c7fc 100644 --- a/syncapi/consumers/eduserver_typing.go +++ b/syncapi/consumers/eduserver_typing.go @@ -56,7 +56,7 @@ func NewOutputTypingEventConsumer( return &OutputTypingEventConsumer{ ctx: process.Context(), jetstream: js, - topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputTypingEvent), + topic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputTypingEvent), durable: cfg.Matrix.JetStream.Durable("SyncAPIEDUServerTypingConsumer"), eduCache: eduCache, notifier: notifier, diff --git a/syncapi/consumers/roomserver.go b/syncapi/consumers/roomserver.go index 640c505c2..5bdc0fad7 100644 --- a/syncapi/consumers/roomserver.go +++ b/syncapi/consumers/roomserver.go @@ -65,7 +65,7 @@ func NewOutputRoomEventConsumer( ctx: process.Context(), cfg: cfg, jetstream: js, - topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputRoomEvent), + topic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputRoomEvent), durable: cfg.Matrix.JetStream.Durable("SyncAPIRoomServerConsumer"), db: store, notifier: notifier, diff --git a/syncapi/consumers/userapi.go b/syncapi/consumers/userapi.go index a3b2dd53d..010fa7c8e 100644 --- a/syncapi/consumers/userapi.go +++ b/syncapi/consumers/userapi.go @@ -56,7 +56,7 @@ func NewOutputNotificationDataConsumer( ctx: process.Context(), jetstream: js, durable: cfg.Matrix.JetStream.Durable("SyncAPINotificationDataConsumer"), - topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputNotificationData), + topic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputNotificationData), db: store, notifier: notifier, stream: stream, diff --git a/syncapi/syncapi.go b/syncapi/syncapi.go index f1f827221..ed8118bfc 100644 --- a/syncapi/syncapi.go +++ b/syncapi/syncapi.go @@ -67,18 +67,18 @@ func AddPublicRoutes( userAPIStreamEventProducer := &producers.UserAPIStreamEventProducer{ JetStream: js, - Topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputStreamEvent), + Topic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputStreamEvent), } userAPIReadUpdateProducer := &producers.UserAPIReadProducer{ JetStream: js, - Topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputReadUpdate), + Topic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputReadUpdate), } _ = userAPIReadUpdateProducer keyChangeConsumer := consumers.NewOutputKeyChangeEventConsumer( - process, cfg, cfg.Matrix.JetStream.TopicFor(jetstream.OutputKeyChangeEvent), + process, cfg, cfg.Matrix.JetStream.Prefixed(jetstream.OutputKeyChangeEvent), js, keyAPI, rsAPI, syncDB, notifier, streams.DeviceListStreamProvider, ) diff --git a/userapi/consumers/syncapi_readupdate.go b/userapi/consumers/syncapi_readupdate.go index 2e58020b4..067f93330 100644 --- a/userapi/consumers/syncapi_readupdate.go +++ b/userapi/consumers/syncapi_readupdate.go @@ -47,7 +47,7 @@ func NewOutputReadUpdateConsumer( db: store, ServerName: cfg.Matrix.ServerName, durable: cfg.Matrix.JetStream.Durable("UserAPISyncAPIReadUpdateConsumer"), - topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputReadUpdate), + topic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputReadUpdate), pgClient: pgClient, userAPI: userAPI, syncProducer: syncProducer, diff --git a/userapi/consumers/syncapi_streamevent.go b/userapi/consumers/syncapi_streamevent.go index 110813274..da3cd3937 100644 --- a/userapi/consumers/syncapi_streamevent.go +++ b/userapi/consumers/syncapi_streamevent.go @@ -54,7 +54,7 @@ func NewOutputStreamEventConsumer( jetstream: js, db: store, durable: cfg.Matrix.JetStream.Durable("UserAPISyncAPIStreamEventConsumer"), - topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputStreamEvent), + topic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputStreamEvent), pgClient: pgClient, userAPI: userAPI, rsAPI: rsAPI, diff --git a/userapi/userapi.go b/userapi/userapi.go index 97bdf7b2d..e91ce3a7a 100644 --- a/userapi/userapi.go +++ b/userapi/userapi.go @@ -54,8 +54,8 @@ func NewInternalAPI( // it's handled by clientapi, and hence uses its topic. When user // API handles it for all account data, we can remove it from // here. - cfg.Matrix.JetStream.TopicFor(jetstream.OutputClientData), - cfg.Matrix.JetStream.TopicFor(jetstream.OutputNotificationData), + cfg.Matrix.JetStream.Prefixed(jetstream.OutputClientData), + cfg.Matrix.JetStream.Prefixed(jetstream.OutputNotificationData), ) userAPI := &internal.UserInternalAPI{