hoernschen/dendrite
1
0
Fork 0
mirror of https://github.com/hoernschen/dendrite.git synced 2025-08-02 06:12:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Deduplicate FS database, EDU persistence table (#1207)

Browse source 
* Deduplicate FS database, add some EDU persistence groundwork

* Extend TransactionWriter to use optional existing transaction, use that for FS SQLite database writes

* Fix build due to bad keyserver import

* Working EDU persistence

* gocyclo, unsurprisingly

* Remove unused

* Update copyright notices
This commit is contained in:
Neil Alexander 2020-07-20 16:55:20 +01:00 • committed by GitHub
parent f3c482b078
commit 11a39fe3b5
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
22 changed files with 1334 additions and 698 deletions
Download patch file Download diff file Expand all files Collapse all files

33
federationsender/queue/queue.go
View file

@ -61,12 +61,23 @@ func NewOutgoingQueues(
queues: map[gomatrixserverlib.ServerName]*destinationQueue{},
}
// Look up which servers we have pending items for and then rehydrate those queues.
if serverNames, err := db.GetPendingServerNames(context.Background()); err == nil {
for _, serverName := range serverNames {
queues.getQueue(serverName).wakeQueueIfNeeded()
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 server names for destination queue hydration")
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{}{}
}
} else {
log.WithError(err).Error("Failed to get EDU server names for destination queue hydration")
}
for serverName := range serverNames {
queues.getQueue(serverName).wakeQueueIfNeeded()
}
return queues
}
@ -91,9 +102,9 @@ func (oqs *OutgoingQueues) getQueue(destination gomatrixserverlib.ServerName) *d
destination: destination,
client: oqs.client,
statistics: oqs.statistics.ForServer(destination),
incomingEDUs: make(chan *gomatrixserverlib.EDU, 128),
incomingInvites: make(chan *gomatrixserverlib.InviteV2Request, 128),
notifyPDUs: make(chan bool, 1),
notifyEDUs: make(chan bool, 1),
interruptBackoff: make(chan bool),
signing: oqs.signing,
}
@ -196,8 +207,18 @@ func (oqs *OutgoingQueues) SendEDU(
}).Info("Sending EDU event")
}
ephemeralJSON, err := json.Marshal(e)
if err != nil {
return fmt.Errorf("json.Marshal: %w", err)
}
nid, err := oqs.db.StoreJSON(context.TODO(), string(ephemeralJSON))
if err != nil {
return fmt.Errorf("sendevent: oqs.db.StoreJSON: %w", err)
}
for _, destination := range destinations {
oqs.getQueue(destination).sendEDU(e)
oqs.getQueue(destination).sendEDU(nid)
}
return nil