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

Separate queueInputRoomEvents function

Browse source
This commit is contained in:
Neil Alexander 2022-03-23 09:47:28 +00:00
parent 01403f9d62
commit e62ee5cd8e
No known key found for this signature in database
GPG key ID: A02A2019A2BB0944
1 changed files with 42 additions and 28 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

70
roomserver/internal/input/input.go
View file

@ -278,6 +278,44 @@ func (w *worker) _next() {
} }
} }
// queueInputRoomEvents queues events into the roomserver input
// stream in NATS.
func (r *Inputer) queueInputRoomEvents(
ctx context.Context,
request *api.InputRoomEventsRequest,
response *api.InputRoomEventsResponse,
replyTo string, // empty string if not synchronous
) (err error) {
// For each event, marshal the input room event and then
// send it into the input queue.
for _, e := range request.InputRoomEvents {
roomID := e.Event.RoomID()
subj := r.Cfg.Matrix.JetStream.Prefixed(jetstream.InputRoomEventSubj(roomID))
msg := &nats.Msg{
Subject: subj,
Header: nats.Header{},
}
msg.Header.Set("room_id", roomID)
if replyTo != "" {
msg.Header.Set("sync", replyTo)
}
msg.Data, err = json.Marshal(e)
if err != nil {
response.ErrMsg = err.Error()
return
}
if _, err = r.JetStream.PublishMsg(msg, nats.Context(ctx)); err != nil {
logrus.WithError(err).WithFields(logrus.Fields{
"room_id": roomID,
"event_id": e.Event.EventID(),
"subj": subj,
}).Error("Roomserver failed to queue async event")
return
}
}
return
}
// InputRoomEvents implements api.RoomserverInternalAPI // InputRoomEvents implements api.RoomserverInternalAPI
func (r *Inputer) InputRoomEvents( func (r *Inputer) InputRoomEvents(
ctx context.Context, ctx context.Context,
@ -307,34 +345,10 @@ func (r *Inputer) InputRoomEvents(
} }
} }
// For each event, marshal the input room event and then // Queue up the event into the roomserver.
// send it into the input queue. if err := r.queueInputRoomEvents(ctx, request, response, replyTo); err != nil {
var err error response.ErrMsg = err.Error()
for _, e := range request.InputRoomEvents { return
roomID := e.Event.RoomID()
subj := r.Cfg.Matrix.JetStream.Prefixed(jetstream.InputRoomEventSubj(roomID))
msg := &nats.Msg{
Subject: subj,
Header: nats.Header{},
Reply: replyTo,
}
msg.Header.Set("room_id", roomID)
if replyTo != "" {
msg.Header.Set("sync", replyTo)
}
msg.Data, err = json.Marshal(e)
if err != nil {
response.ErrMsg = err.Error()
return
}
if _, err = r.JetStream.PublishMsg(msg); err != nil {
logrus.WithError(err).WithFields(logrus.Fields{
"room_id": roomID,
"event_id": e.Event.EventID(),
"subj": subj,
}).Error("Roomserver failed to queue async event")
return
}
} }
// If we aren't waiting for synchronous responses then we can // If we aren't waiting for synchronous responses then we can