concurrent download

TODO.md

@@ -1,4 +1,4 @@
-## MVP
+## MVP DONE
 
 * ✓ Write TODO
 * ✓ Cron
@@ -14,7 +14,7 @@
 * ✓ Wrap content inside item ⚠️  atom calls it entry
 * ✓ Implement atom
 * ✓ Cleanup
-* Show Feed name instead of Link
+* ✓ Show Feed name instead of Link
 * ✓ Make Rhash optional
 
 ## Issues

app/app.go

@@ -3,6 +3,7 @@ package app
 import (
 	"io"
 	"net/http"
+	"sync"
 	"time"
 
 	"go.balki.me/tss/log"
@@ -36,10 +37,16 @@ func Run(configPath string) {
 
 	tgram := telegram.NewTelegramSender(tgramProxy, cfg.TelegramAuthToken)
 
+	wg := sync.WaitGroup{}
 	for _, feed := range cfg.Feeds {
 		log.Info("processing feed", "feed", feed.Name)
-		ProcessFeed(feed, scheduler, cfg.DbDir, tgram)
+		wg.Add(1)
+		go func() {
+			ProcessFeed(feed, scheduler, cfg.DbDir, tgram)
+			wg.Done()
+		}()
 	}
+	wg.Wait()
 }
 
 func ProcessFeed(feed FeedCfg, scheduler Scheduler, dbDir string, tgram telegram.TelegramSender) {

app/schedule.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"sync"
 	"time"
 
 	"github.com/robfig/cron/v3"
@@ -19,6 +20,7 @@ type Scheduler interface {
 
 type scheduler struct {
 	filePath        string
+	mutex           sync.Mutex
 	lastSuccessTime map[string]time.Time
 }
 
@@ -45,6 +47,8 @@ func NewScheduler(filePath string) (Scheduler, error) {
 }
 
 func (s *scheduler) Save() error {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
 	data, err := yaml.Marshal(&s.lastSuccessTime)
 	if err != nil {
 		return err
@@ -53,10 +57,14 @@ func (s *scheduler) Save() error {
 }
 
 func (s *scheduler) Good(name string) {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
 	s.lastSuccessTime[name] = time.Now()
 }
 
 func (s *scheduler) ShouldDownload(name string, scheduleSpec string) (bool, error) {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
 	lastSuccess, ok := s.lastSuccessTime[name]
 	if !ok {
 		return true, nil