scheduler.go 1.5 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556
  1. // SPDX-FileCopyrightText: Copyright The Miniflux Authors. All rights reserved.
  2. // SPDX-License-Identifier: Apache-2.0
  3. package cli // import "miniflux.app/v2/internal/cli"
  4. import (
  5. "log/slog"
  6. "time"
  7. "miniflux.app/v2/internal/config"
  8. "miniflux.app/v2/internal/storage"
  9. "miniflux.app/v2/internal/worker"
  10. )
  11. func runScheduler(store *storage.Storage, pool *worker.Pool) {
  12. slog.Debug(`Starting background scheduler...`)
  13. go feedScheduler(
  14. store,
  15. pool,
  16. config.Opts.PollingFrequency(),
  17. config.Opts.BatchSize(),
  18. config.Opts.PollingParsingErrorLimit(),
  19. )
  20. go cleanupScheduler(
  21. store,
  22. config.Opts.CleanupFrequencyHours(),
  23. )
  24. }
  25. func feedScheduler(store *storage.Storage, pool *worker.Pool, frequency, batchSize, errorLimit int) {
  26. for range time.Tick(time.Duration(frequency) * time.Minute) {
  27. // Generate a batch of feeds for any user that has feeds to refresh.
  28. batchBuilder := store.NewBatchBuilder()
  29. batchBuilder.WithBatchSize(batchSize)
  30. batchBuilder.WithErrorLimit(errorLimit)
  31. batchBuilder.WithoutDisabledFeeds()
  32. batchBuilder.WithNextCheckExpired()
  33. if jobs, err := batchBuilder.FetchJobs(); err != nil {
  34. slog.Error("Unable to fetch jobs from database", slog.Any("error", err))
  35. } else if len(jobs) > 0 {
  36. slog.Info("Created a batch of feeds",
  37. slog.Int("nb_jobs", len(jobs)),
  38. )
  39. pool.Push(jobs)
  40. }
  41. }
  42. }
  43. func cleanupScheduler(store *storage.Storage, frequency int) {
  44. for range time.Tick(time.Duration(frequency) * time.Hour) {
  45. runCleanupTasks(store)
  46. }
  47. }