alecodes/miniflux-archiver
2
0
Fork 0
generated from alecodes/base-template
Code Issues Pull requests Projects Releases Packages 1 Wiki Activity Actions

fead: expand archive options

Browse source 
add an option to select how to archive the entries
add an option for what entries to archive
change the concurrentcy limit to a semaphore instead of a ticker
This commit is contained in:
Alexander Navarro 2024-12-20 14:59:43 -03:00
parent 1788241cd5
commit b0506c3eab
9 changed files with 99 additions and 72 deletions
Download patch file Download diff file Expand all files Collapse all files

41
internal/app/archive.go
View file

@ -1,35 +1,17 @@
package app
import (
"sync"
"time"
"context"
"git.alecodes.page/alecodes/miniflux-archiver/internal/logger"
"git.alecodes.page/alecodes/miniflux-archiver/internal/miniflux"
"git.alecodes.page/alecodes/miniflux-archiver/internal/service"
"golang.org/x/sync/semaphore"
)
// NewRateLimiter creates a rate limiter that emits time events at a specified rate.
// request_per specifies the number of requests allowed per time_scale duration.
// time_scale specifies the duration over which the requests are allowed.
func newRateLimiter(request_per int, time_scale time.Duration) <-chan time.Time {
rate_limit := make(chan time.Time, request_per)
tickrate := time_scale / time.Duration(request_per)
for range request_per {
rate_limit <- time.Now()
}
go func() {
for t := range time.Tick(tickrate) {
rate_limit <- t
}
}()
return rate_limit
}
func Archive(minifluxConfig miniflux.MinifluxConfig, serviceConfig service.ServiceConfig) {
ctx := context.Background()
mf, err := miniflux.NewMiniflux(minifluxConfig)
if err != nil {
logger.Fatal("Could not connect to the miniflux server: %v", err)
@ -45,17 +27,16 @@ func Archive(minifluxConfig miniflux.MinifluxConfig, serviceConfig service.Servi
logger.Fatal("Could not retrieve entries from the miniflux feed: %v", err)
}
var wg sync.WaitGroup
ticker := newRateLimiter(5, 10*time.Second)
sem := semaphore.NewWeighted(int64(serviceConfig.MaxRequests))
for _, entry := range result.Entries {
wg.Add(1)
if err := sem.Acquire(ctx, 1); err != nil {
logger.Fatal("Failed to acquire semaphore: %v", err)
}
go func() {
defer wg.Done()
<-ticker
defer sem.Release(1)
// logger.Debug("Sending url \"%v\" to be marked as read", entry.URL)
err := externalService.Archive(entry.URL)
if err != nil {
logger.Fatal("Could not archive url \"%v\" from the service: %v", entry.URL, err)
@ -65,5 +46,7 @@ func Archive(minifluxConfig miniflux.MinifluxConfig, serviceConfig service.Servi
}()
}
wg.Wait()
if err := sem.Acquire(ctx, int64(serviceConfig.MaxRequests)); err != nil {
logger.Fatal("Failed to acquire semaphore: %v", err)
}
}