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

Fix up event filtering

Browse source
This commit is contained in:
Andrew Morgan 2018-06-24 20:54:04 +01:00
parent 79754a2ab7
commit 9f863de5d6
1 changed files with 13 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
src/github.com/matrix-org/dendrite/appservice/consumers/roomserver.go
View file

@ -113,12 +113,8 @@ func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error {
return err
}
// Check if any events need to passed on to external application services
if len(events) > 0 {
// Check if this was a membership event
return s.filterRoomserverEvents(ctx, events)
}
return s.filterRoomserverEvents(ctx, append(events, ev))
// Combine any state and non-state events and send them to the application service
return s.filterRoomserverEvents(ctx, ev)
}
// lookupStateEvents looks up the state events that are added by a new event.
@ -172,20 +168,18 @@ func (s *OutputRoomEventConsumer) lookupStateEvents(
// application service.
func (s *OutputRoomEventConsumer) filterRoomserverEvents(
ctx context.Context,
events []gomatrixserverlib.Event,
event gomatrixserverlib.Event,
) error {
for _, event := range events {
for _, ws := range s.workerStates {
// Check if this event is interesting to this application service
if s.appserviceIsInterestedInEvent(ctx, event, ws.AppService) {
// Queue this event to be sent off to the application service
if err := s.asDB.StoreEvent(ctx, ws.AppService.ID, &event); err != nil {
log.WithError(err).Warn("failed to insert incoming event into appservices database")
} else {
// Tell our worker to send out new messages by updating remaining message
// count and waking them up with a broadcast
ws.NotifyNewEvent()
}
for _, ws := range s.workerStates {
// Check if this event is interesting to this application service
if s.appserviceIsInterestedInEvent(ctx, event, ws.AppService) {
// Queue this event to be sent off to the application service
if err := s.asDB.StoreEvent(ctx, ws.AppService.ID, &event); err != nil {
log.WithError(err).Warn("failed to insert incoming event into appservices database")
} else {
// Tell our worker to send out new messages by updating remaining message
// count and waking them up with a broadcast
ws.NotifyNewEvent()
}
}
}