refactor(background): parameterize sync queue and worker pool constructors
Split each constructor into a default-args wrapper and a config-accepting variant so the sync queue interval/batch size and the worker pool size/ queue cap can be sourced from the settings registry at startup. These values are constructed once at boot, so they are tagged requires_restart in the admin UI. queue.go: - NewSyncQueueProcessorWithConfig(db, interval, batchSize) takes the flush interval and batch size as parameters; NewSyncQueueProcessor becomes a thin wrapper with the historical 5s / 50 defaults. worker.go: - NewWorkerWithConfig(numWorkers, queueCap, connManager) takes the queue capacity as a parameter; NewWorker becomes a thin wrapper with the historical cap of 100. No behavior change for existing callers; main.go will switch to the config-accepting variants in a follow-up wiring commit.
This commit is contained in:
@@ -176,10 +176,17 @@ func (w *Worker) GetActiveJobCount() int {
|
||||
return count
|
||||
}
|
||||
func NewWorker(numWorkers int, connManager *wsync.ConnectionManager) *Worker {
|
||||
return NewWorkerWithConfig(numWorkers, 100, connManager)
|
||||
}
|
||||
|
||||
// NewWorkerWithConfig constructs a worker pool with the given worker count and
|
||||
// job-queue capacity. Used at startup to source values from the settings
|
||||
// registry.
|
||||
func NewWorkerWithConfig(numWorkers, queueCap int, connManager *wsync.ConnectionManager) *Worker {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
w := &Worker{
|
||||
jobQueue: make(chan *Job, 100),
|
||||
jobQueue: make(chan *Job, queueCap),
|
||||
results: make(map[string]*JobResult),
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
@@ -995,3 +1002,4 @@ func (w *Worker) Shutdown() {
|
||||
close(w.jobQueue)
|
||||
w.wg.Wait()
|
||||
}
|
||||
|
||||
|
||||
@@ -74,11 +74,18 @@ type SyncQueueItem struct {
|
||||
}
|
||||
|
||||
func NewSyncQueueProcessor(db *database.Queries) *SyncQueueProcessor {
|
||||
return NewSyncQueueProcessorWithConfig(db, 5*time.Second, 50)
|
||||
}
|
||||
|
||||
// NewSyncQueueProcessorWithConfig constructs a processor with the given flush
|
||||
// interval and batch size. Used at startup to source values from the settings
|
||||
// registry.
|
||||
func NewSyncQueueProcessorWithConfig(db *database.Queries, interval time.Duration, batchSize int) *SyncQueueProcessor {
|
||||
return &SyncQueueProcessor{
|
||||
db: db,
|
||||
progressChan: make(chan *ProgressUpdate, 100),
|
||||
interval: 5 * time.Second,
|
||||
batchSize: 50,
|
||||
interval: interval,
|
||||
batchSize: batchSize,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user