Method sendEDU

federationapi/queue/destinationqueue.go:117–157

sendEDU adds the EDU 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.EDU, receipt *shared.Receipt)

// If the queue is empty then it starts a background goroutine to
// start sending events to that destination.
func (oq *destinationQueue) sendEDU(event *gomatrixserverlib.EDU, receipt *shared.Receipt) {
	if event == nil {
		logrus.Errorf("attempt to send nil EDU with destination %q", oq.destination)
		return
	}
	// Create a database entry that associates the given PDU NID with
	// this destination queue. We'll then be able to retrieve the PDU
	// later.
	if err := oq.db.AssociateEDUWithDestination(
		oq.process.Context(),
		oq.destination, // the destination server name
		receipt,        // NIDs from federationapi_queue_json table
		event.Type,
		nil, // this will use the default expireEDUTypes map
	); err != nil {
		logrus.WithError(err).Errorf("failed to associate EDU with destination %q", oq.destination)
		return
	}
	// Check if the destination is blacklisted. If it isn't then wake
	// up the queue.
	if !oq.statistics.Blacklisted() {
		// If there's room in memory to hold the event then add it to the
		// list.
		oq.pendingMutex.Lock()
		if len(oq.pendingEDUs) < maxEDUsInMemory {
			oq.pendingEDUs = append(oq.pendingEDUs, &queuedEDU{
				edu:     event,
				receipt: receipt,
			})
		} else {
			oq.overflowed.Store(true)
		}
		oq.pendingMutex.Unlock()
		// Wake up the queue if it's asleep.
		oq.wakeQueueIfNeeded()
		select {
		case oq.notify <- struct{}{}:
		default:
		}
	}
}

// wakeQueueIfNeeded will wake up the destination queue if it is
// not already running. If it is running but it is backing off