Function NewBatcher

coderd/agentapi/metadatabatcher/metadata_batcher.go:134–179

NewBatcher creates a new Batcher and starts it. Here ctx controls the lifetime of the batcher, canceling it will result in the Batcher exiting it's processing routine (run).

(ctx context.Context, reg prometheus.Registerer, store database.Store, ps pubsub.Pubsub, opts ...Option)

// NewBatcher creates a new Batcher and starts it. Here ctx controls the lifetime of the batcher, canceling it will
// result in the Batcher exiting it's processing routine (run).
func NewBatcher(ctx context.Context, reg prometheus.Registerer, store database.Store, ps pubsub.Pubsub, opts ...Option) (*Batcher, error) {
	b := &Batcher{
		store:   store,
		ps:      ps,
		Metrics: NewMetrics(),
		done:    make(chan struct{}),
		log:     slog.Logger{},
		clock:   quartz.NewReal(),
	}

	for _, opt := range opts {
		opt(b)
	}

	b.Metrics.register(reg)

	if b.interval == 0 {
		b.interval = defaultMetadataFlushInterval
	}

	if b.maxBatchSize == 0 {
		b.maxBatchSize = defaultMetadataBatchSize
	}

	// Create warn ticker after options are applied so it uses the correct clock.
	b.warnTicker = b.clock.NewTicker(10 * time.Second)

	if b.timer == nil {
		b.timer = b.clock.NewTimer(b.interval)
	}

	// Create buffered channel with 5x batch size capacity
	channelSize := b.maxBatchSize * defaultChannelBufferMultiplier
	b.updateCh = make(chan update, channelSize)

	// Initialize batch map
	b.batch = make(map[compositeKey]value)

	b.ctx, b.cancel = context.WithCancel(ctx)
	go func() {
		b.run(b.ctx)
		close(b.done)
	}()

	return b, nil
}

func (b *Batcher) Close() {
	b.cancel()