diff options
author | Frédéric Guillot <fred@miniflux.net> | 2017-11-19 21:10:04 -0800 |
---|---|---|
committer | Frédéric Guillot <fred@miniflux.net> | 2017-11-19 22:01:46 -0800 |
commit | 8ffb773f43c8dc54801ca1d111854e7e881c93c9 (patch) | |
tree | 38133a2fc612597a75fed1d13e5b4042f58a2b7e /scheduler |
First commit
Diffstat (limited to 'scheduler')
-rw-r--r-- | scheduler/scheduler.go | 24 | ||||
-rw-r--r-- | scheduler/worker.go | 35 | ||||
-rw-r--r-- | scheduler/worker_pool.go | 34 |
3 files changed, 93 insertions, 0 deletions
diff --git a/scheduler/scheduler.go b/scheduler/scheduler.go new file mode 100644 index 0000000..ab87e99 --- /dev/null +++ b/scheduler/scheduler.go @@ -0,0 +1,24 @@ +// Copyright 2017 Frédéric Guillot. All rights reserved. +// Use of this source code is governed by the Apache 2.0 +// license that can be found in the LICENSE file. + +package scheduler + +import ( + "github.com/miniflux/miniflux2/storage" + "log" + "time" +) + +// NewScheduler starts a new scheduler to push jobs to a pool of workers. +func NewScheduler(store *storage.Storage, workerPool *WorkerPool, frequency, batchSize int) { + c := time.Tick(time.Duration(frequency) * time.Minute) + for now := range c { + jobs := store.GetJobs(batchSize) + log.Printf("[Scheduler:%v] => Pushing %d jobs\n", now, len(jobs)) + + for _, job := range jobs { + workerPool.Push(job) + } + } +} diff --git a/scheduler/worker.go b/scheduler/worker.go new file mode 100644 index 0000000..39382a3 --- /dev/null +++ b/scheduler/worker.go @@ -0,0 +1,35 @@ +// Copyright 2017 Frédéric Guillot. All rights reserved. +// Use of this source code is governed by the Apache 2.0 +// license that can be found in the LICENSE file. + +package scheduler + +import ( + "github.com/miniflux/miniflux2/model" + "github.com/miniflux/miniflux2/reader/feed" + "log" + "time" +) + +// A Worker refresh a feed in the background. +type Worker struct { + id int + feedHandler *feed.Handler +} + +// Run wait for a job and refresh the given feed. +func (w *Worker) Run(c chan model.Job) { + log.Printf("[Worker] #%d started\n", w.id) + + for { + job := <-c + log.Printf("[Worker #%d] got userID=%d, feedID=%d\n", w.id, job.UserID, job.FeedID) + + err := w.feedHandler.RefreshFeed(job.UserID, job.FeedID) + if err != nil { + log.Println("Worker:", err) + } + + time.Sleep(time.Millisecond * 1000) + } +} diff --git a/scheduler/worker_pool.go b/scheduler/worker_pool.go new file mode 100644 index 0000000..b753f89 --- /dev/null +++ b/scheduler/worker_pool.go @@ -0,0 +1,34 @@ +// Copyright 2017 Frédéric Guillot. All rights reserved. +// Use of this source code is governed by the Apache 2.0 +// license that can be found in the LICENSE file. + +package scheduler + +import ( + "github.com/miniflux/miniflux2/model" + "github.com/miniflux/miniflux2/reader/feed" +) + +// WorkerPool handle a pool of workers. +type WorkerPool struct { + queue chan model.Job +} + +// Push send a job on the queue. +func (w *WorkerPool) Push(job model.Job) { + w.queue <- job +} + +// NewWorkerPool creates a pool of background workers. +func NewWorkerPool(feedHandler *feed.Handler, nbWorkers int) *WorkerPool { + workerPool := &WorkerPool{ + queue: make(chan model.Job), + } + + for i := 0; i < nbWorkers; i++ { + worker := &Worker{id: i, feedHandler: feedHandler} + go worker.Run(workerPool.queue) + } + + return workerPool +} |