miniflux_v2/internal/cli/refresh_feeds.go

// SPDX-FileCopyrightText: Copyright The Miniflux Authors. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

package cli // import "miniflux.app/v2/internal/cli"

import (
	"sync"
	"time"

	"miniflux.app/v2/internal/config"
	"miniflux.app/v2/internal/logger"
	"miniflux.app/v2/internal/model"
	feedHandler "miniflux.app/v2/internal/reader/handler"
	"miniflux.app/v2/internal/storage"
)

func refreshFeeds(store *storage.Storage) {
	var wg sync.WaitGroup

	startTime := time.Now()
	jobs, err := store.NewBatch(config.Opts.BatchSize())
	if err != nil {
		logger.Error("[Cronjob] %v", err)
	}

	nbJobs := len(jobs)
	logger.Info("[Cronjob]] Created %d jobs from a batch size of %d", nbJobs, config.Opts.BatchSize())
	var jobQueue = make(chan model.Job, nbJobs)

	logger.Info("[Cronjob] Starting a pool of %d workers", config.Opts.WorkerPoolSize())
	for i := 0; i < config.Opts.WorkerPoolSize(); i++ {
		wg.Add(1)
		go func(workerID int) {
			defer wg.Done()
			for job := range jobQueue {
				logger.Info("[Cronjob] Refreshing feed #%d for user #%d in worker #%d", job.FeedID, job.UserID, workerID)
				if err := feedHandler.RefreshFeed(store, job.UserID, job.FeedID, false); err != nil {
					logger.Error("[Cronjob] Refreshing the feed #%d returned this error: %v", job.FeedID, err)
				}
			}
		}(i)
	}

	for _, job := range jobs {
		jobQueue <- job
	}
	close(jobQueue)

	wg.Wait()
	logger.Info("[Cronjob] Refreshed %d feed(s) in %s", nbJobs, time.Since(startTime))
}