Method Enqueue

pkg/eventqueue/eventqueue.go:98–134

Enqueue pushes the given event onto the EventQueue. If the queue has been stopped, the Event will not be enqueued, and its cancel channel will be closed, indicating that the Event was not ran. This function may block if the queue is at its capacity for events. If a single Event has Enqueue called on

(ev *Event)

// if the Event has been previously enqueued, if the Event is nil, or the queue
// itself is not initialized properly.
func (q *EventQueue) Enqueue(ev *Event) (<-chan any, error) {
	if q.notSafeToAccess() || ev == nil {
		return nil, fmt.Errorf("unable to Enqueue event")
	}

	// Events can only be enqueued once.
	if !ev.enqueued.CompareAndSwap(false, true) {
		return nil, fmt.Errorf("unable to Enqueue event; event has already had Enqueue called on it")
	}

	// Multiple Enqueues can occur at the same time. Ensure that events channel
	// is not closed while we are enqueueing events.
	q.eventsMu.RLock()
	defer q.eventsMu.RUnlock()

	select {
	// The event should be drained from the queue (e.g., it should not be
	// processed).
	case <-q.drain:
		// Closed eventResults channel signifies cancellation.
		close(ev.cancelled)
		close(ev.eventResults)

		return ev.eventResults, nil
	default:
		// The events channel may be closed even if an event has been pushed
		// onto the events channel, as events are consumed off of the events
		// channel asynchronously! If the EventQueue is closed before this
		// event is processed, then it will be cancelled.

		ev.stats.waitEnqueue.Start()
		ev.stats.waitConsumeOffQueue.Start()
		q.events <- ev
		ev.stats.waitEnqueue.End(true)
		return ev.eventResults, nil
	}
}

// Event is an event that can be enqueued onto an EventQueue.
type Event struct {