(destination gomatrixserverlib.ServerName)
| 152 | } |
| 153 | |
| 154 | func (oqs *OutgoingQueues) getQueue(destination gomatrixserverlib.ServerName) *destinationQueue { |
| 155 | if oqs.statistics.ForServer(destination).Blacklisted() { |
| 156 | return nil |
| 157 | } |
| 158 | oqs.queuesMutex.Lock() |
| 159 | defer oqs.queuesMutex.Unlock() |
| 160 | oq, ok := oqs.queues[destination] |
| 161 | if !ok || oq == nil { |
| 162 | destinationQueueTotal.Inc() |
| 163 | oq = &destinationQueue{ |
| 164 | queues: oqs, |
| 165 | db: oqs.db, |
| 166 | process: oqs.process, |
| 167 | rsAPI: oqs.rsAPI, |
| 168 | origin: oqs.origin, |
| 169 | destination: destination, |
| 170 | client: oqs.client, |
| 171 | statistics: oqs.statistics.ForServer(destination), |
| 172 | notify: make(chan struct{}, 1), |
| 173 | interruptBackoff: make(chan bool), |
| 174 | signing: oqs.signing, |
| 175 | } |
| 176 | oqs.queues[destination] = oq |
| 177 | } |
| 178 | return oq |
| 179 | } |
| 180 | |
| 181 | func (oqs *OutgoingQueues) clearQueue(oq *destinationQueue) { |
| 182 | oqs.queuesMutex.Lock() |
no test coverage detected