hoernschen/dendrite
1
0
Fork 0
mirror of https://github.com/hoernschen/dendrite.git synced 2024-12-27 07:28:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'master' into matthew/peeking

Browse source
This commit is contained in:
Matthew Hodgson 2020-09-04 00:06:58 +01:00 committed by GitHub
commit 4f5c8cac33
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 25 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

37
federationsender/queue/queue.go
View file

@ -20,6 +20,7 @@ import (
"encoding/json"
"fmt"
"sync"
"time"
stateapi "github.com/matrix-org/dendrite/currentstateserver/api"
"github.com/matrix-org/dendrite/federationsender/statistics"
@ -65,26 +66,28 @@ func NewOutgoingQueues(
queues: map[gomatrixserverlib.ServerName]*destinationQueue{},
}
// Look up which servers we have pending items for and then rehydrate those queues.
serverNames := map[gomatrixserverlib.ServerName]struct{}{}
if names, err := db.GetPendingPDUServerNames(context.Background()); err == nil {
for _, serverName := range names {
serverNames[serverName] = struct{}{}
time.AfterFunc(time.Second*5, func() {
serverNames := map[gomatrixserverlib.ServerName]struct{}{}
if names, err := db.GetPendingPDUServerNames(context.Background()); err == nil {
for _, serverName := range names {
serverNames[serverName] = struct{}{}
}
} else {
log.WithError(err).Error("Failed to get PDU server names for destination queue hydration")
}
} else {
log.WithError(err).Error("Failed to get PDU server names for destination queue hydration")
}
if names, err := db.GetPendingEDUServerNames(context.Background()); err == nil {
for _, serverName := range names {
serverNames[serverName] = struct{}{}
if names, err := db.GetPendingEDUServerNames(context.Background()); err == nil {
for _, serverName := range names {
serverNames[serverName] = struct{}{}
}
} else {
log.WithError(err).Error("Failed to get EDU server names for destination queue hydration")
}
} else {
log.WithError(err).Error("Failed to get EDU server names for destination queue hydration")
}
for serverName := range serverNames {
if !queues.getQueue(serverName).statistics.Blacklisted() {
queues.getQueue(serverName).wakeQueueIfNeeded()
for serverName := range serverNames {
if !queues.getQueue(serverName).statistics.Blacklisted() {
queues.getQueue(serverName).wakeQueueIfNeeded()
}
}
}
})
return queues
}

9
keyserver/keyserver.go
View file

@ -48,10 +48,11 @@ func NewInternalAPI(
DB: db,
}
updater := internal.NewDeviceListUpdater(db, keyChangeProducer, fedClient, 8) // 8 workers TODO: configurable
err = updater.Start()
if err != nil {
logrus.WithError(err).Panicf("failed to start device list updater")
}
go func() {
if err := updater.Start(); err != nil {
logrus.WithError(err).Panicf("failed to start device list updater")
}
}()
return &internal.KeyInternalAPI{
DB: db,
ThisServer: cfg.Matrix.ServerName,