refactor scheduler

service/scheduler/app.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"github.com/dagu-dev/dagu/internal/config"
 	"github.com/dagu-dev/dagu/internal/logger"
-	"github.com/dagu-dev/dagu/service/scheduler/entry"
+	"github.com/dagu-dev/dagu/service/scheduler/entry_reader"
 	"github.com/dagu-dev/dagu/service/scheduler/scheduler"
 	"go.uber.org/fx"
 )
@@ -23,11 +23,11 @@ type Params struct {
 	EntryReader scheduler.EntryReader
 }
 
-func EntryReaderProvider(cfg *config.Config, jf entry.JobFactory, logger logger.Logger) scheduler.EntryReader {
-	return entry.NewEntryReader(cfg.DAGs, jf, logger)
+func EntryReaderProvider(cfg *config.Config, jf entry_reader.JobFactory, logger logger.Logger) scheduler.EntryReader {
+	return entry_reader.NewEntryReader(cfg.DAGs, jf, logger)
 }
 
-func JobFactoryProvider(cfg *config.Config) entry.JobFactory {
+func JobFactoryProvider(cfg *config.Config) entry_reader.JobFactory {
 	return &jobFactory{
 		Command: cfg.Command,
 		WorkDir: cfg.WorkDir,

service/scheduler/entry/entry.go → service/scheduler/entry_reader/entry_reader.go

@@ -1,4 +1,4 @@
-package entry
+package entry_reader
 
 import (
 	"github.com/dagu-dev/dagu/internal/logger"
@@ -35,7 +35,7 @@ func NewEntryReader(dagsDir string, jf JobFactory, logger logger.Logger) *EntryR
 		logger:   logger,
 	}
 	if err := er.initDags(); err != nil {
-		er.logger.Error("failed to init entry dags", tag.Error(err))
+		er.logger.Error("failed to init entry_reader dags", tag.Error(err))
 	}
 	go er.watchDags()
 	return er
@@ -131,19 +131,19 @@ func (er *EntryReader) watchDags() {
 					er.logger.Error("failed to read workflow cfg", tag.Error(err))
 				} else {
 					er.dags[filepath.Base(event.Name)] = workflow
-					er.logger.Info("reload workflow entry", "file", event.Name)
+					er.logger.Info("reload workflow entry_reader", "file", event.Name)
 				}
 			}
 			if event.Op == fsnotify.Rename || event.Op == fsnotify.Remove {
 				delete(er.dags, filepath.Base(event.Name))
-				er.logger.Info("remove workflow entry", "file", event.Name)
+				er.logger.Info("remove workflow entry_reader", "file", event.Name)
 			}
 			er.dagsLock.Unlock()
 		case err, ok := <-watcher.Errors():
 			if !ok {
 				return
 			}
-			er.logger.Error("watch entry dags error", tag.Error(err))
+			er.logger.Error("watch entry_reader dags error", tag.Error(err))
 		}
 	}
 

service/scheduler/entry/entry_test.go → service/scheduler/entry_reader/entry_reader_test.go

@@ -1,4 +1,4 @@
-package entry
+package entry_reader
 
 import (
 	"github.com/dagu-dev/dagu/internal/dag"

service/scheduler/scheduler/scheduler.go

@@ -141,7 +141,7 @@ func (s *Scheduler) run(now time.Time) {
 		go func(e *Entry) {
 			err := e.Invoke()
 			if err != nil {
-				s.logger.Error("failed to invoke entry", "entry", e.Job, "error", err)
+				s.logger.Error("failed to invoke entry_reader", "entry_reader", e.Job, "error", err)
 			}
 		}(e)
 	}