Method Start

subscription_manager.go:47–95

(ctx context.Context)

}

func (sm *subscriptionManager) Start(ctx context.Context) error {
	ctx, shouldStart, started, stopped := sm.StartInit(ctx)
	if !shouldStart {
		return nil
	}

	go func() {
		started()
		defer stopped() // this defer should come first so it's last out

		sm.Logger.DebugContext(ctx, sm.Name+": Run loop started")
		defer sm.Logger.DebugContext(ctx, sm.Name+": Run loop stopped")

		// On shutdown, close and remove all active subscriptions.
		defer func() {
			sm.mu.Lock()
			defer sm.mu.Unlock()

			for subID, sub := range sm.subscriptions {
				close(sub.Chan)
				delete(sm.subscriptions, subID)
			}
		}()

		for {
			select {
			case <-ctx.Done():
				// Distribute remaining subscriptions until the channel is
				// closed. This does make the subscription manager a little
				// problematic in that it requires the subscription channel to
				// be closed before it will fully stop. This always happens in
				// the case of a real client by virtue of the completer always
				// stopping at the same time as the subscription manager, but
				// one has to be careful in tests.
				sm.Logger.DebugContext(ctx, sm.Name+": Stopping; distributing subscriptions until channel is closed")
				for updates := range sm.subscribeCh {
					sm.distributeJobUpdates(ctx, updates)
				}

				return

			case updates := <-sm.subscribeCh:
				sm.distributeJobUpdates(ctx, updates)
			}
		}
	}()

	return nil
}

func (sm *subscriptionManager) logStats(ctx context.Context, svcName string) {
	sm.statsMu.Lock()