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

Update consumers for the roomserver output stream

Browse source
This commit is contained in:
Neil Alexander 2022-01-25 17:07:37 +00:00
parent 0e8bb6b762
commit 98de7c2f77
No known key found for this signature in database
GPG key ID: A02A2019A2BB0944
3 changed files with 11 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
federationapi/consumers/roomserver.go
View file

@ -66,7 +66,11 @@ func NewOutputRoomEventConsumer(
// Start consuming from room servers // Start consuming from room servers
func (s *OutputRoomEventConsumer) Start() error { func (s *OutputRoomEventConsumer) Start() error {
_, err := s.jetstream.Subscribe(s.topic, s.onMessage, s.durable) _, err := s.jetstream.Subscribe(
s.topic, s.onMessage, s.durable,
nats.DeliverAll(),
nats.ManualAck(),
)
return err return err
} }

1
setup/jetstream/helpers.go
View file

@ -3,6 +3,7 @@ package jetstream
import "github.com/nats-io/nats.go" import "github.com/nats-io/nats.go"
func WithJetStreamMessage(msg *nats.Msg, f func(msg *nats.Msg) bool) { func WithJetStreamMessage(msg *nats.Msg, f func(msg *nats.Msg) bool) {
_ = msg.InProgress()
if f(msg) { if f(msg) {
_ = msg.Ack() _ = msg.Ack()
} else { } else {

6
syncapi/consumers/roomserver.go
View file

@ -73,7 +73,11 @@ func NewOutputRoomEventConsumer(
// Start consuming from room servers // Start consuming from room servers
func (s *OutputRoomEventConsumer) Start() error { func (s *OutputRoomEventConsumer) Start() error {
_, err := s.jetstream.Subscribe(s.topic, s.onMessage, s.durable) _, err := s.jetstream.Subscribe(
s.topic, s.onMessage, s.durable,
nats.DeliverAll(),
nats.ManualAck(),
)
return err return err
} }