Method distributeJobUpdates

subscription_manager.go:122–147

Receives updates from the completer and prompts the client to update statistics and distribute jobs into any listening subscriber channels. (Subscriber channels are non-blocking so this should be quite fast.)

(ctx context.Context, updates []jobcompleter.CompleterJobUpdated)

// statistics and distribute jobs into any listening subscriber channels.
// (Subscriber channels are non-blocking so this should be quite fast.)
func (sm *subscriptionManager) distributeJobUpdates(ctx context.Context, updates []jobcompleter.CompleterJobUpdated) {
	func() {
		sm.statsMu.Lock()
		defer sm.statsMu.Unlock()

		for _, update := range updates {
			stats := update.JobStats
			sm.statsAggregate.CompleteDuration += stats.CompleteDuration
			sm.statsAggregate.QueueWaitDuration += stats.QueueWaitDuration
			sm.statsAggregate.RunDuration += stats.RunDuration
			sm.statsNumJobs++
		}
	}()

	sm.mu.Lock()
	defer sm.mu.Unlock()

	// Quick path so we don't need to allocate anything if no one is listening.
	if len(sm.subscriptions) < 1 {
		return
	}

	for _, update := range updates {
		sm.distributeJobEvent(ctx, update.Job, jobStatisticsFromInternal(update.JobStats), update.Snoozed)
	}
}

// Distribute a single event into any listening subscriber channels.
//