james/yarr
4
0
Fork 0
mirror of https://github.com/nkanaev/yarr.git synced 2025-05-24 00:33:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix refresh sync

Browse Source
This commit is contained in:
Nazar Kanaev 2021-05-27 11:52:16 +01:00
parent eb9bfc57e2
commit 214c7aacfc
2 changed files with 22 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
src/assets/javascripts/app.js
View File

@ -633,10 +633,7 @@ var vm = new Vue({
fetchAllFeeds: function() { fetchAllFeeds: function() {
if (this.loading.feeds) return if (this.loading.feeds) return
api.feeds.refresh().then(function() { api.feeds.refresh().then(function() {
// TODO: this is hacky. come up with something decent vm.refreshStats()
setTimeout(function() {
vm.refreshStats()
}, 1000)
}) })
}, },
computeStats: function() { computeStats: function() {

38
src/worker/worker.go
View File

@ -9,6 +9,8 @@ import (
"github.com/nkanaev/yarr/src/storage" "github.com/nkanaev/yarr/src/storage"
) )
const NUM_WORKERS = 4
type Worker struct { type Worker struct {
db *storage.Storage db *storage.Storage
pending *int32 pending *int32
@ -89,30 +91,34 @@ func (w *Worker) SetRefreshRate(minute int64) {
} }
func (w *Worker) RefreshFeeds() { func (w *Worker) RefreshFeeds() {
log.Print("Refreshing feeds")
go w.refresher()
}
func (w *Worker) refresher() {
w.reflock.Lock() w.reflock.Lock()
defer w.reflock.Unlock()
w.db.ResetFeedErrors() if *w.pending > 0 {
log.Print("Refreshing already in progress")
feeds := w.db.ListFeeds()
if len(feeds) == 0 {
return return
} }
feeds := w.db.ListFeeds()
if len(feeds) == 0 {
log.Print("Nothing to refresh")
return
}
log.Print("Refreshing feeds")
atomic.StoreInt32(w.pending, int32(len(feeds))) atomic.StoreInt32(w.pending, int32(len(feeds)))
go w.refresher(feeds)
}
func (w *Worker) refresher(feeds []storage.Feed) {
w.db.ResetFeedErrors()
srcqueue := make(chan storage.Feed, len(feeds)) srcqueue := make(chan storage.Feed, len(feeds))
dstqueue := make(chan []storage.Item) dstqueue := make(chan []storage.Item)
// hardcoded to 4 workers ;) for i := 0; i < NUM_WORKERS; i++ {
go w.worker(srcqueue, dstqueue) go w.worker(srcqueue, dstqueue)
go w.worker(srcqueue, dstqueue) }
go w.worker(srcqueue, dstqueue)
go w.worker(srcqueue, dstqueue)
for _, feed := range feeds { for _, feed := range feeds {
srcqueue <- feed srcqueue <- feed
@ -120,14 +126,12 @@ func (w *Worker) refresher() {
for i := 0; i < len(feeds); i++ { for i := 0; i < len(feeds); i++ {
w.db.CreateItems(<-dstqueue) w.db.CreateItems(<-dstqueue)
atomic.AddInt32(w.pending, -1) atomic.AddInt32(w.pending, -1)
w.db.SyncSearch()
} }
close(srcqueue) close(srcqueue)
close(dstqueue) close(dstqueue)
w.db.SyncSearch()
log.Printf("Finished refreshing %d feeds", len(feeds)) log.Printf("Finished refreshing %d feeds", len(feeds))
w.reflock.Unlock()
} }
func (w *Worker) worker(srcqueue <-chan storage.Feed, dstqueue chan<- []storage.Item) { func (w *Worker) worker(srcqueue <-chan storage.Feed, dstqueue chan<- []storage.Item) {