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

Tweak that some more

Browse source
This commit is contained in:
Neil Alexander 2022-08-31 15:24:07 +01:00
parent ea115539ef
commit 10dd3a7c2a
No known key found for this signature in database
GPG key ID: A02A2019A2BB0944
1 changed files with 4 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
appservice/consumers/roomserver.go
View file

@ -104,17 +104,14 @@ func (s *OutputRoomEventConsumer) onMessage(
if err := json.Unmarshal(msg.Data, &output); err != nil { if err := json.Unmarshal(msg.Data, &output); err != nil {
// If the message was invalid, log it and move on to the next message in the stream // If the message was invalid, log it and move on to the next message in the stream
log.WithField("appservice", state.ID).WithError(err).Errorf("Appservice failed to parse message, ignoring") log.WithField("appservice", state.ID).WithError(err).Errorf("Appservice failed to parse message, ignoring")
_ = msg.Ack()
continue continue
} }
switch output.Type { switch output.Type {
case api.OutputTypeNewRoomEvent: case api.OutputTypeNewRoomEvent:
if output.NewRoomEvent == nil { if output.NewRoomEvent == nil {
_ = msg.Ack()
continue continue
} }
if !s.appserviceIsInterestedInEvent(ctx, output.NewRoomEvent.Event, state.ApplicationService) { if !s.appserviceIsInterestedInEvent(ctx, output.NewRoomEvent.Event, state.ApplicationService) {
_ = msg.Ack()
continue continue
} }
events = append(events, output.NewRoomEvent.Event) events = append(events, output.NewRoomEvent.Event)
@ -145,21 +142,20 @@ func (s *OutputRoomEventConsumer) onMessage(
events = append(events, output.NewInviteEvent.Event) events = append(events, output.NewInviteEvent.Event)
default: default:
_ = msg.Ack()
continue continue
} }
} }
// If there are no events selected for sending then we should // If there are no events selected for sending then we should
// negatively ack the pending events so we will retry them again // ack the events so that we don't get sent them again in the
// later. // future.
if len(events) == 0 { if len(events) == 0 {
return false return true
} }
// Send event to any relevant application services // Send event to any relevant application services
if err := s.sendEvents(ctx, state, events); err != nil { if err := s.sendEvents(ctx, state, events); err != nil {
log.WithField("appservice", state.ID).WithError(err).Errorf("Appservice failed to filter events") log.WithField("appservice", state.ID).WithError(err).Errorf("Appservice failed to send events")
return false return false
} }