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

We don't really need Ch here do we

Browse source
This commit is contained in:
Neil Alexander 2020-07-03 16:24:46 +01:00
parent 82d11484bc
commit 8bc9de2fe0
2 changed files with 14 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
federationsender/queue/destinationqueue.go
View file

@ -56,7 +56,7 @@ type destinationQueue struct {
pendingEDUs []*gomatrixserverlib.EDU // owned by backgroundSend pendingEDUs []*gomatrixserverlib.EDU // owned by backgroundSend
pendingInvites []*gomatrixserverlib.InviteV2Request // owned by backgroundSend pendingInvites []*gomatrixserverlib.InviteV2Request // owned by backgroundSend
notifyPDUs chan bool // interrupts idle wait for PDUs notifyPDUs chan bool // interrupts idle wait for PDUs
interruptBackoffCh chan bool // interrupts backoff interruptBackoff chan bool // interrupts backoff
} }
// Send event adds the event to the pending queue for the destination. // Send event adds the event to the pending queue for the destination.
@ -133,7 +133,7 @@ func (oq *destinationQueue) sendInvite(ev *gomatrixserverlib.InviteV2Request) {
func (oq *destinationQueue) wakeQueueIfNeeded() { func (oq *destinationQueue) wakeQueueIfNeeded() {
// If we are backing off then interrupt the backoff. // If we are backing off then interrupt the backoff.
if oq.backingOff.CAS(true, false) { if oq.backingOff.CAS(true, false) {
oq.interruptBackoffCh <- true oq.interruptBackoff <- true
} }
// If we aren't running then wake up the queue. // If we aren't running then wake up the queue.
if !oq.running.Load() { if !oq.running.Load() {
@ -217,7 +217,7 @@ func (oq *destinationQueue) backgroundSend() {
oq.backingOff.Store(true) oq.backingOff.Store(true)
select { select {
case <-time.After(duration): case <-time.After(duration):
case <-oq.interruptBackoffCh: case <-oq.interruptBackoff:
} }
oq.backingOff.Store(false) oq.backingOff.Store(false)
} }

22
federationsender/queue/queue.go
View file

@ -85,17 +85,17 @@ func (oqs *OutgoingQueues) getQueue(destination gomatrixserverlib.ServerName) *d
oq := oqs.queues[destination] oq := oqs.queues[destination]
if oq == nil { if oq == nil {
oq = &destinationQueue{ oq = &destinationQueue{
db: oqs.db, db: oqs.db,
rsAPI: oqs.rsAPI, rsAPI: oqs.rsAPI,
origin: oqs.origin, origin: oqs.origin,
destination: destination, destination: destination,
client: oqs.client, client: oqs.client,
statistics: oqs.statistics.ForServer(destination), statistics: oqs.statistics.ForServer(destination),
incomingEDUs: make(chan *gomatrixserverlib.EDU, 128), incomingEDUs: make(chan *gomatrixserverlib.EDU, 128),
incomingInvites: make(chan *gomatrixserverlib.InviteV2Request, 128), incomingInvites: make(chan *gomatrixserverlib.InviteV2Request, 128),
notifyPDUs: make(chan bool, 128), notifyPDUs: make(chan bool, 128),
interruptBackoffCh: make(chan bool), interruptBackoff: make(chan bool),
signing: oqs.signing, signing: oqs.signing,
} }
oqs.queues[destination] = oq oqs.queues[destination] = oq
} }