0
0
Fork 0
mirror of https://github.com/matrix-org/dendrite synced 2024-12-16 15:43:42 +01:00
dendrite/test/testrig/jetstream.go

36 lines
909 B
Go
Raw Normal View History

package testrig
import (
"encoding/json"
"testing"
"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) {
t.Helper()
for _, msg := range msgs {
if _, err := jsctx.PublishMsg(msg); err != nil {
t.Fatalf("MustPublishMsgs: failed to publish message: %s", err)
}
}
}
func NewOutputEventMsg(t *testing.T, base *base.BaseDendrite, roomID string, update api.OutputEvent) *nats.Msg {
t.Helper()
msg := &nats.Msg{
Subject: base.Cfg.Global.JetStream.Prefixed(jetstream.OutputRoomEvent),
Header: nats.Header{},
}
msg.Header.Set(jetstream.RoomID, roomID)
var err error
msg.Data, err = json.Marshal(update)
if err != nil {
t.Fatalf("failed to marshal update: %s", err)
}
return msg
}