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
|
return count
|
||||||
}
|
}
|
||||||
func NewWorker(numWorkers int, connManager *wsync.ConnectionManager) *Worker {
|
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())
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
|
||||||
w := &Worker{
|
w := &Worker{
|
||||||
jobQueue: make(chan *Job, 100),
|
jobQueue: make(chan *Job, queueCap),
|
||||||
results: make(map[string]*JobResult),
|
results: make(map[string]*JobResult),
|
||||||
ctx: ctx,
|
ctx: ctx,
|
||||||
cancel: cancel,
|
cancel: cancel,
|
||||||
@@ -995,3 +1002,4 @@ func (w *Worker) Shutdown() {
|
|||||||
close(w.jobQueue)
|
close(w.jobQueue)
|
||||||
w.wg.Wait()
|
w.wg.Wait()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -74,11 +74,18 @@ type SyncQueueItem struct {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func NewSyncQueueProcessor(db *database.Queries) *SyncQueueProcessor {
|
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{
|
return &SyncQueueProcessor{
|
||||||
db: db,
|
db: db,
|
||||||
progressChan: make(chan *ProgressUpdate, 100),
|
progressChan: make(chan *ProgressUpdate, 100),
|
||||||
interval: 5 * time.Second,
|
interval: interval,
|
||||||
batchSize: 50,
|
batchSize: batchSize,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user