Function NewQueueCleaner

internal/maintenance/queue_cleaner.go:87–100

(archetype *baseservice.Archetype, config *QueueCleanerConfig, exec riverdriver.Executor)

}

func NewQueueCleaner(archetype *baseservice.Archetype, config *QueueCleanerConfig, exec riverdriver.Executor) *QueueCleaner {
	batchSizes := config.WithDefaults()

	return baseservice.Init(archetype, &QueueCleaner{
		Config: (&QueueCleanerConfig{
			BatchSizes:      batchSizes,
			Interval:        cmp.Or(config.Interval, queueCleanerIntervalDefault),
			RetentionPeriod: cmp.Or(config.RetentionPeriod, QueueRetentionPeriodDefault),
			Schema:          config.Schema,
		}).mustValidate(),
		exec:                    exec,
		reducedBatchSizeBreaker: riversharedmaintenance.ReducedBatchSizeBreaker(batchSizes),
	})
}

func (s *QueueCleaner) Start(ctx context.Context) error {
	ctx, shouldStart, started, stopped := s.StartInit(ctx)