Function newBinder

enterprise/tailnet/pgcoord.go:508–561

(ctx context.Context,
	logger slog.Logger,
	id uuid.UUID,
	store database.Store,
	ps pubsub.Pubsub,
	bindings <-chan binding,
	startWorkers <-chan struct{},
)

}

func newBinder(ctx context.Context,
	logger slog.Logger,
	id uuid.UUID,
	store database.Store,
	ps pubsub.Pubsub,
	bindings <-chan binding,
	startWorkers <-chan struct{},
) *binder {
	b := &binder{
		ctx:           ctx,
		logger:        logger,
		coordinatorID: id,
		store:         store,
		pubsub:        ps,
		bindings:      bindings,
		latest:        make(map[bKey]binding),
		workQ:         newWorkQ[bKey](ctx),
		close:         make(chan struct{}),
	}
	go b.handleBindings()
	// add to the waitgroup immediately to avoid any races waiting for it before
	// the workers start.
	b.workerWG.Add(numBinderWorkers)
	go func() {
		<-startWorkers
		for i := 0; i < numBinderWorkers; i++ {
			go b.worker()
		}
	}()

	go func() {
		defer close(b.close)
		<-b.ctx.Done()
		b.logger.Debug(b.ctx, "binder exiting, waiting for workers")

		b.workerWG.Wait()

		b.logger.Debug(b.ctx, "updating peers to lost")

		ctx, cancel := context.WithTimeout(dbauthz.As(context.Background(), pgCoordSubject), time.Second*15)
		defer cancel()
		peerIDs, err := b.store.UpdateTailnetPeerStatusByCoordinator(ctx, database.UpdateTailnetPeerStatusByCoordinatorParams{
			CoordinatorID: b.coordinatorID,
			Status:        database.TailnetStatusLost,
		})
		if err != nil {
			b.logger.Error(b.ctx, "update peer status to lost", slog.Error(err))
		}
		for _, peerID := range peerIDs {
			publishPeerUpdate(ctx, b.pubsub, b.logger, peerID)
		}
	}()
	return b
}

func (b *binder) handleBindings() {
	for {