Add RoomEventType nats.Header to avoid unneeded unmarshalling (#2765)

appservice/consumers/roomserver.go

@@ -101,6 +101,11 @@ func (s *OutputRoomEventConsumer) onMessage(
 	log.WithField("appservice", state.ID).Tracef("Appservice worker received %d message(s) from roomserver", len(msgs))
 	events := make([]*gomatrixserverlib.HeaderedEvent, 0, len(msgs))
 	for _, msg := range msgs {
+		// Only handle events we care about
+		receivedType := api.OutputType(msg.Header.Get(jetstream.RoomEventType))
+		if receivedType != api.OutputTypeNewRoomEvent && receivedType != api.OutputTypeNewInviteEvent {
+			continue
+		}
 		// Parse out the event JSON
 		var output api.OutputEvent
 		if err := json.Unmarshal(msg.Data, &output); err != nil {

federationapi/consumers/roomserver.go

@@ -79,6 +79,13 @@ func (s *OutputRoomEventConsumer) Start() error {
 // realises that it cannot update the room state using the deltas.
 func (s *OutputRoomEventConsumer) onMessage(ctx context.Context, msgs []*nats.Msg) bool {
 	msg := msgs[0] // Guaranteed to exist if onMessage is called
+	receivedType := api.OutputType(msg.Header.Get(jetstream.RoomEventType))
+
+	// Only handle events we care about
+	if receivedType != api.OutputTypeNewRoomEvent && receivedType != api.OutputTypeNewInboundPeek {
+		return true
+	}
+
 	// Parse out the event JSON
 	var output api.OutputEvent
 	if err := json.Unmarshal(msg.Data, &output); err != nil {

roomserver/producers/roomevent.go

@@ -17,12 +17,13 @@ package producers
 import (
 	"encoding/json"
 
-	"github.com/matrix-org/dendrite/roomserver/acls"
-	"github.com/matrix-org/dendrite/roomserver/api"
-	"github.com/matrix-org/dendrite/setup/jetstream"
 	"github.com/nats-io/nats.go"
 	log "github.com/sirupsen/logrus"
 	"github.com/tidwall/gjson"
+
+	"github.com/matrix-org/dendrite/roomserver/acls"
+	"github.com/matrix-org/dendrite/roomserver/api"
+	"github.com/matrix-org/dendrite/setup/jetstream"
 )
 
 var keyContentFields = map[string]string{
@@ -40,10 +41,8 @@ type RoomEventProducer struct {
 func (r *RoomEventProducer) ProduceRoomEvents(roomID string, updates []api.OutputEvent) error {
 	var err error
 	for _, update := range updates {
-		msg := &nats.Msg{
+		msg := nats.NewMsg(r.Topic)
-			Subject: r.Topic,
+		msg.Header.Set(jetstream.RoomEventType, string(update.Type))
-			Header:  nats.Header{},
-		}
 		msg.Header.Set(jetstream.RoomID, roomID)
 		msg.Data, err = json.Marshal(update)
 		if err != nil {

setup/jetstream/streams.go

@@ -9,9 +9,10 @@ import (
 )
 
 const (
-	UserID  = "user_id"
+	UserID        = "user_id"
-	RoomID  = "room_id"
+	RoomID        = "room_id"
-	EventID = "event_id"
+	EventID       = "event_id"
+	RoomEventType = "output_room_event_type"
 )
 
 var (

test/testrig/jetstream.go

@@ -4,10 +4,11 @@ import (
 	"encoding/json"
 	"testing"
 
+	"github.com/nats-io/nats.go"
+
 	"github.com/matrix-org/dendrite/roomserver/api"
 	"github.com/matrix-org/dendrite/setup/base"
 	"github.com/matrix-org/dendrite/setup/jetstream"
-	"github.com/nats-io/nats.go"
 )
 
 func MustPublishMsgs(t *testing.T, jsctx nats.JetStreamContext, msgs ...*nats.Msg) {
@@ -21,10 +22,8 @@ func MustPublishMsgs(t *testing.T, jsctx nats.JetStreamContext, msgs ...*nats.Ms
 
 func NewOutputEventMsg(t *testing.T, base *base.BaseDendrite, roomID string, update api.OutputEvent) *nats.Msg {
 	t.Helper()
-	msg := &nats.Msg{
+	msg := nats.NewMsg(base.Cfg.Global.JetStream.Prefixed(jetstream.OutputRoomEvent))
-		Subject: base.Cfg.Global.JetStream.Prefixed(jetstream.OutputRoomEvent),
+	msg.Header.Set(jetstream.RoomEventType, string(update.Type))
-		Header:  nats.Header{},
-	}
 	msg.Header.Set(jetstream.RoomID, roomID)
 	var err error
 	msg.Data, err = json.Marshal(update)

userapi/consumers/roomserver.go

@@ -72,15 +72,16 @@ func (s *OutputRoomEventConsumer) Start() error {
 
 func (s *OutputRoomEventConsumer) onMessage(ctx context.Context, msgs []*nats.Msg) bool {
 	msg := msgs[0] // Guaranteed to exist if onMessage is called
+	// Only handle events we care about
+	if rsapi.OutputType(msg.Header.Get(jetstream.RoomEventType)) != rsapi.OutputTypeNewRoomEvent {
+		return true
+	}
 	var output rsapi.OutputEvent
 	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
 		log.WithError(err).Errorf("roomserver output log: message parse failure")
 		return true
 	}
-	if output.Type != rsapi.OutputTypeNewRoomEvent {
-		return true
-	}
 	event := output.NewRoomEvent.Event
 	if event == nil {
 		log.Errorf("userapi consumer: expected event")