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

Wait for streams to process the event

Browse source
This commit is contained in:
Till Faelligen 2022-12-22 07:00:09 +01:00
parent 23d1002142
commit 1e85ec6441
No known key found for this signature in database
GPG key ID: ACCDC9606D472758
1 changed files with 15 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
roomserver/roomserver_test.go
View file

@ -9,6 +9,7 @@ import (
"github.com/gorilla/mux" "github.com/gorilla/mux"
"github.com/matrix-org/dendrite/federationapi" "github.com/matrix-org/dendrite/federationapi"
"github.com/matrix-org/dendrite/keyserver" "github.com/matrix-org/dendrite/keyserver"
"github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/syncapi" "github.com/matrix-org/dendrite/syncapi"
"github.com/matrix-org/dendrite/userapi" "github.com/matrix-org/dendrite/userapi"
"github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/gomatrixserverlib"
@ -164,6 +165,9 @@ func TestPurgeRoom(t *testing.T) {
base, db, close := mustCreateDatabase(t, dbType) base, db, close := mustCreateDatabase(t, dbType)
defer close() defer close()
jsCtx, _ := base.NATS.Prepare(base.ProcessContext, &base.Cfg.Global.JetStream)
defer jetstream.DeleteAllStreams(jsCtx, &base.Cfg.Global.JetStream)
fedClient := base.CreateFederationClient() fedClient := base.CreateFederationClient()
rsAPI := roomserver.NewInternalAPI(base) rsAPI := roomserver.NewInternalAPI(base)
keyAPI := keyserver.NewInternalAPI(base, &base.Cfg.KeyServer, fedClient, rsAPI) keyAPI := keyserver.NewInternalAPI(base, &base.Cfg.KeyServer, fedClient, rsAPI)
@ -248,9 +252,17 @@ func TestPurgeRoom(t *testing.T) {
if err = rsAPI.PerformAdminPurgeRoom(ctx, &api.PerformAdminPurgeRoomRequest{RoomID: room.ID}, purgeResp); err != nil { if err = rsAPI.PerformAdminPurgeRoom(ctx, &api.PerformAdminPurgeRoomRequest{RoomID: room.ID}, purgeResp); err != nil {
t.Fatal(err) t.Fatal(err)
} }
// TODO: Find a better solution, e.g. "hook" into the JetStream stream.
// Gives the SyncAPI and FederationAPI some time to delete their entries // wait for all consumers to process the purge event
time.Sleep(time.Millisecond * 100) var sum = 1
for sum > 0 {
sum = 0
consumerCh := jsCtx.Consumers(base.Cfg.Global.JetStream.Prefixed(jetstream.OutputRoomEvent))
for x := range consumerCh {
sum += x.NumAckPending
}
time.Sleep(time.Millisecond)
}
roomInfo, err = db.RoomInfo(ctx, room.ID) roomInfo, err = db.RoomInfo(ctx, room.ID)
if err != nil { if err != nil {