Send event adds the event to the pending queue for the destination. If the queue is empty then it starts a background goroutine to start sending events to that destination.
(event *gomatrixserverlib.HeaderedEvent, receipt *shared.Receipt)
| 71 | // If the queue is empty then it starts a background goroutine to |
| 72 | // start sending events to that destination. |
| 73 | func (oq *destinationQueue) sendEvent(event *gomatrixserverlib.HeaderedEvent, receipt *shared.Receipt) { |
| 74 | if event == nil { |
| 75 | logrus.Errorf("attempt to send nil PDU with destination %q", oq.destination) |
| 76 | return |
| 77 | } |
| 78 | // Create a database entry that associates the given PDU NID with |
| 79 | // this destination queue. We'll then be able to retrieve the PDU |
| 80 | // later. |
| 81 | if err := oq.db.AssociatePDUWithDestination( |
| 82 | oq.process.Context(), |
| 83 | "", // TODO: remove this, as we don't need to persist the transaction ID |
| 84 | oq.destination, // the destination server name |
| 85 | receipt, // NIDs from federationapi_queue_json table |
| 86 | ); err != nil { |
| 87 | logrus.WithError(err).Errorf("failed to associate PDU %q with destination %q", event.EventID(), oq.destination) |
| 88 | return |
| 89 | } |
| 90 | // Check if the destination is blacklisted. If it isn't then wake |
| 91 | // up the queue. |
| 92 | if !oq.statistics.Blacklisted() { |
| 93 | // If there's room in memory to hold the event then add it to the |
| 94 | // list. |
| 95 | oq.pendingMutex.Lock() |
| 96 | if len(oq.pendingPDUs) < maxPDUsInMemory { |
| 97 | oq.pendingPDUs = append(oq.pendingPDUs, &queuedPDU{ |
| 98 | pdu: event, |
| 99 | receipt: receipt, |
| 100 | }) |
| 101 | } else { |
| 102 | oq.overflowed.Store(true) |
| 103 | } |
| 104 | oq.pendingMutex.Unlock() |
| 105 | // Wake up the queue if it's asleep. |
| 106 | oq.wakeQueueIfNeeded() |
| 107 | select { |
| 108 | case oq.notify <- struct{}{}: |
| 109 | default: |
| 110 | } |
| 111 | } |
| 112 | } |
| 113 | |
| 114 | // sendEDU adds the EDU event to the pending queue for the destination. |
| 115 | // If the queue is empty then it starts a background goroutine to |
no test coverage detected