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

Pull consumers

Browse source
This commit is contained in:
Neil Alexander 2022-02-02 11:36:53 +00:00
parent 9f969498ce
commit d95c55577b
No known key found for this signature in database
GPG key ID: A02A2019A2BB0944
11 changed files with 40 additions and 36 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
appservice/consumers/roomserver.go
View file

@ -34,7 +34,7 @@ import (
type OutputRoomEventConsumer struct { type OutputRoomEventConsumer struct {
ctx context.Context ctx context.Context
jetstream nats.JetStreamContext jetstream nats.JetStreamContext
durable nats.SubOpt durable string
topic string topic string
asDB storage.Database asDB storage.Database
rsAPI api.RoomserverInternalAPI rsAPI api.RoomserverInternalAPI
@ -67,8 +67,8 @@ func NewOutputRoomEventConsumer(
// Start consuming from room servers // Start consuming from room servers
func (s *OutputRoomEventConsumer) Start() error { func (s *OutputRoomEventConsumer) Start() error {
return jetstream.JetStreamConsumer( return jetstream.JetStreamConsumer(
s.ctx, s.jetstream, s.topic, s.onMessage, s.ctx, s.jetstream, s.topic, s.durable, s.onMessage,
s.durable, nats.DeliverAll(), nats.ManualAck(), nats.DeliverAll(), nats.ManualAck(),
) )
} }

14
federationapi/consumers/eduserver.go
View file

@ -34,7 +34,7 @@ import (
type OutputEDUConsumer struct { type OutputEDUConsumer struct {
ctx context.Context ctx context.Context
jetstream nats.JetStreamContext jetstream nats.JetStreamContext
durable nats.SubOpt durable string
db storage.Database db storage.Database
queues *queue.OutgoingQueues queues *queue.OutgoingQueues
ServerName gomatrixserverlib.ServerName ServerName gomatrixserverlib.ServerName
@ -67,20 +67,20 @@ func NewOutputEDUConsumer(
// Start consuming from EDU servers // Start consuming from EDU servers
func (t *OutputEDUConsumer) Start() error { func (t *OutputEDUConsumer) Start() error {
if err := jetstream.JetStreamConsumer( if err := jetstream.JetStreamConsumer(
t.ctx, t.jetstream, t.typingTopic, t.onTypingEvent, t.ctx, t.jetstream, t.typingTopic, t.durable, t.onTypingEvent,
t.durable, nats.DeliverAll(), nats.ManualAck(), nats.DeliverAll(), nats.ManualAck(),
); err != nil { ); err != nil {
return err return err
} }
if err := jetstream.JetStreamConsumer( if err := jetstream.JetStreamConsumer(
t.ctx, t.jetstream, t.sendToDeviceTopic, t.onSendToDeviceEvent, t.ctx, t.jetstream, t.sendToDeviceTopic, t.durable, t.onSendToDeviceEvent,
t.durable, nats.DeliverAll(), nats.ManualAck(), nats.DeliverAll(), nats.ManualAck(),
); err != nil { ); err != nil {
return err return err
} }
if err := jetstream.JetStreamConsumer( if err := jetstream.JetStreamConsumer(
t.ctx, t.jetstream, t.receiptTopic, t.onReceiptEvent, t.ctx, t.jetstream, t.receiptTopic, t.durable, t.onReceiptEvent,
t.durable, nats.DeliverAll(), nats.ManualAck(), nats.DeliverAll(), nats.ManualAck(),
); err != nil { ); err != nil {
return err return err
} }

6
federationapi/consumers/roomserver.go
View file

@ -37,7 +37,7 @@ type OutputRoomEventConsumer struct {
cfg *config.FederationAPI cfg *config.FederationAPI
rsAPI api.RoomserverInternalAPI rsAPI api.RoomserverInternalAPI
jetstream nats.JetStreamContext jetstream nats.JetStreamContext
durable nats.SubOpt durable string
db storage.Database db storage.Database
queues *queue.OutgoingQueues queues *queue.OutgoingQueues
topic string topic string
@ -67,8 +67,8 @@ func NewOutputRoomEventConsumer(
// Start consuming from room servers // Start consuming from room servers
func (s *OutputRoomEventConsumer) Start() error { func (s *OutputRoomEventConsumer) Start() error {
return jetstream.JetStreamConsumer( return jetstream.JetStreamConsumer(
s.ctx, s.jetstream, s.topic, s.onMessage, s.ctx, s.jetstream, s.topic, s.durable, s.onMessage,
s.durable, nats.DeliverAll(), nats.ManualAck(), nats.DeliverAll(), nats.ManualAck(),
) )
} }

4
roomserver/internal/api.go
View file

@ -41,7 +41,7 @@ type RoomserverInternalAPI struct {
fsAPI fsAPI.FederationInternalAPI fsAPI fsAPI.FederationInternalAPI
asAPI asAPI.AppServiceQueryAPI asAPI asAPI.AppServiceQueryAPI
JetStream nats.JetStreamContext JetStream nats.JetStreamContext
Durable nats.SubOpt Durable string
InputRoomEventTopic string // JetStream topic for new input room events InputRoomEventTopic string // JetStream topic for new input room events
OutputRoomEventTopic string // JetStream topic for new output room events OutputRoomEventTopic string // JetStream topic for new output room events
PerspectiveServerNames []gomatrixserverlib.ServerName PerspectiveServerNames []gomatrixserverlib.ServerName
@ -87,7 +87,7 @@ func (r *RoomserverInternalAPI) SetFederationAPI(fsAPI fsAPI.FederationInternalA
InputRoomEventTopic: r.InputRoomEventTopic, InputRoomEventTopic: r.InputRoomEventTopic,
OutputRoomEventTopic: r.OutputRoomEventTopic, OutputRoomEventTopic: r.OutputRoomEventTopic,
JetStream: r.JetStream, JetStream: r.JetStream,
Durable: r.Durable, Durable: nats.Durable(r.Durable),
ServerName: r.Cfg.Matrix.ServerName, ServerName: r.Cfg.Matrix.ServerName,
FSAPI: fsAPI, FSAPI: fsAPI,
KeyRing: keyRing, KeyRing: keyRing,

6
setup/config/config_jetstream.go
View file

@ -2,8 +2,6 @@ package config
import ( import (
"fmt" "fmt"
"github.com/nats-io/nats.go"
) )
type JetStream struct { type JetStream struct {
@ -25,8 +23,8 @@ func (c *JetStream) TopicFor(name string) string {
return fmt.Sprintf("%s%s", c.TopicPrefix, name) return fmt.Sprintf("%s%s", c.TopicPrefix, name)
} }
func (c *JetStream) Durable(name string) nats.SubOpt { func (c *JetStream) Durable(name string) string {
return nats.Durable(c.TopicFor(name)) return c.TopicFor(name)
} }
func (c *JetStream) Defaults(generate bool) { func (c *JetStream) Defaults(generate bool) {

10
setup/jetstream/helpers.go
View file

@ -9,11 +9,17 @@ import (
) )
func JetStreamConsumer( func JetStreamConsumer(
ctx context.Context, nats nats.JetStreamContext, subj string, ctx context.Context, nats nats.JetStreamContext, subj, durable string,
f func(ctx context.Context, msg *nats.Msg) bool, f func(ctx context.Context, msg *nats.Msg) bool,
opts ...nats.SubOpt, opts ...nats.SubOpt,
) error { ) error {
sub, err := nats.SubscribeSync(subj, opts...) if _, err := nats.ConsumerInfo(subj, durable); err == nil {
if err := nats.DeleteConsumer(subj, durable); err != nil {
return fmt.Errorf("nats.DeleteConsumer: %w", err)
}
}
sub, err := nats.PullSubscribe(subj, durable, opts...)
if err != nil { if err != nil {
return fmt.Errorf("nats.SubscribeSync: %w", err) return fmt.Errorf("nats.SubscribeSync: %w", err)
} }

6
syncapi/consumers/clientapi.go
View file

@ -34,7 +34,7 @@ import (
type OutputClientDataConsumer struct { type OutputClientDataConsumer struct {
ctx context.Context ctx context.Context
jetstream nats.JetStreamContext jetstream nats.JetStreamContext
durable nats.SubOpt durable string
topic string topic string
db storage.Database db storage.Database
stream types.StreamProvider stream types.StreamProvider
@ -64,8 +64,8 @@ func NewOutputClientDataConsumer(
// Start consuming from room servers // Start consuming from room servers
func (s *OutputClientDataConsumer) Start() error { func (s *OutputClientDataConsumer) Start() error {
return jetstream.JetStreamConsumer( return jetstream.JetStreamConsumer(
s.ctx, s.jetstream, s.topic, s.onMessage, s.ctx, s.jetstream, s.topic, s.durable, s.onMessage,
s.durable, nats.DeliverAll(), nats.ManualAck(), nats.DeliverAll(), nats.ManualAck(),
) )
} }

6
syncapi/consumers/eduserver_receipts.go
View file

@ -34,7 +34,7 @@ import (
type OutputReceiptEventConsumer struct { type OutputReceiptEventConsumer struct {
ctx context.Context ctx context.Context
jetstream nats.JetStreamContext jetstream nats.JetStreamContext
durable nats.SubOpt durable string
topic string topic string
db storage.Database db storage.Database
stream types.StreamProvider stream types.StreamProvider
@ -65,8 +65,8 @@ func NewOutputReceiptEventConsumer(
// Start consuming from EDU api // Start consuming from EDU api
func (s *OutputReceiptEventConsumer) Start() error { func (s *OutputReceiptEventConsumer) Start() error {
return jetstream.JetStreamConsumer( return jetstream.JetStreamConsumer(
s.ctx, s.jetstream, s.topic, s.onMessage, s.ctx, s.jetstream, s.topic, s.durable, s.onMessage,
s.durable, nats.DeliverAll(), nats.ManualAck(), nats.DeliverAll(), nats.ManualAck(),
) )
} }

6
syncapi/consumers/eduserver_sendtodevice.go
View file

@ -36,7 +36,7 @@ import (
type OutputSendToDeviceEventConsumer struct { type OutputSendToDeviceEventConsumer struct {
ctx context.Context ctx context.Context
jetstream nats.JetStreamContext jetstream nats.JetStreamContext
durable nats.SubOpt durable string
topic string topic string
db storage.Database db storage.Database
serverName gomatrixserverlib.ServerName // our server name serverName gomatrixserverlib.ServerName // our server name
@ -69,8 +69,8 @@ func NewOutputSendToDeviceEventConsumer(
// Start consuming from EDU api // Start consuming from EDU api
func (s *OutputSendToDeviceEventConsumer) Start() error { func (s *OutputSendToDeviceEventConsumer) Start() error {
return jetstream.JetStreamConsumer( return jetstream.JetStreamConsumer(
s.ctx, s.jetstream, s.topic, s.onMessage, s.ctx, s.jetstream, s.topic, s.durable, s.onMessage,
s.durable, nats.DeliverAll(), nats.ManualAck(), nats.DeliverAll(), nats.ManualAck(),
) )
} }

6
syncapi/consumers/eduserver_typing.go
View file

@ -35,7 +35,7 @@ import (
type OutputTypingEventConsumer struct { type OutputTypingEventConsumer struct {
ctx context.Context ctx context.Context
jetstream nats.JetStreamContext jetstream nats.JetStreamContext
durable nats.SubOpt durable string
topic string topic string
eduCache *cache.EDUCache eduCache *cache.EDUCache
stream types.StreamProvider stream types.StreamProvider
@ -67,8 +67,8 @@ func NewOutputTypingEventConsumer(
// Start consuming from EDU api // Start consuming from EDU api
func (s *OutputTypingEventConsumer) Start() error { func (s *OutputTypingEventConsumer) Start() error {
return jetstream.JetStreamConsumer( return jetstream.JetStreamConsumer(
s.ctx, s.jetstream, s.topic, s.onMessage, s.ctx, s.jetstream, s.topic, s.durable, s.onMessage,
s.durable, nats.DeliverAll(), nats.ManualAck(), nats.DeliverAll(), nats.ManualAck(),
) )
} }

6
syncapi/consumers/roomserver.go
View file

@ -38,7 +38,7 @@ type OutputRoomEventConsumer struct {
cfg *config.SyncAPI cfg *config.SyncAPI
rsAPI api.RoomserverInternalAPI rsAPI api.RoomserverInternalAPI
jetstream nats.JetStreamContext jetstream nats.JetStreamContext
durable nats.SubOpt durable string
topic string topic string
db storage.Database db storage.Database
pduStream types.StreamProvider pduStream types.StreamProvider
@ -74,8 +74,8 @@ func NewOutputRoomEventConsumer(
// Start consuming from room servers // Start consuming from room servers
func (s *OutputRoomEventConsumer) Start() error { func (s *OutputRoomEventConsumer) Start() error {
return jetstream.JetStreamConsumer( return jetstream.JetStreamConsumer(
s.ctx, s.jetstream, s.topic, s.onMessage, s.ctx, s.jetstream, s.topic, s.durable, s.onMessage,
s.durable, nats.DeliverAll(), nats.ManualAck(), nats.DeliverAll(), nats.ManualAck(),
) )
} }