diff options
| -rw-r--r-- | cmd/server/main.go | 18 | ||||
| -rw-r--r-- | pkg/worker/list_processor.go | 7 |
2 files changed, 20 insertions, 5 deletions
diff --git a/cmd/server/main.go b/cmd/server/main.go index 8b1cc00..54a7ba0 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -109,17 +109,27 @@ func main() { v.SetMyselfIn(extRouter) } + // processors + var ( + fileScanner = worker.NewFileScanner(*root, mediaRepository) + exifScanner = worker.NewEXIFScanner(mediaRepository) + ) + // worker var ( serverWorker = worker.NewServerWorker(&fasthttp.Server{Handler: r.Handler}) - fileScanner = worker.NewFileScanner(*root, mediaRepository) - exifScanner = worker.NewEXIFScanner(mediaRepository) + fileWorker = worker.NewWorkerFromChanProcessor[string](fileScanner, scheduler) + exifWorker = worker.NewWorkerFromListProcessor[*media.Media]( + exifScanner, + scheduler, + logrus.WithField("context", "exif scanner"), + ) ) pool := worker.NewWorkerPool() pool.AddWorker("http server", serverWorker) - pool.AddWorker("exif scanner", worker.NewWorkerFromListProcessor[*media.Media](exifScanner, scheduler)) - pool.AddWorker("file scanner", worker.NewWorkerFromChanProcessor[string](fileScanner, scheduler)) + pool.AddWorker("exif scanner", exifWorker) + pool.AddWorker("file scanner", fileWorker) ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt) defer stop() diff --git a/pkg/worker/list_processor.go b/pkg/worker/list_processor.go index 8169e4e..c9c20a9 100644 --- a/pkg/worker/list_processor.go +++ b/pkg/worker/list_processor.go @@ -4,6 +4,8 @@ import ( "context" "errors" "sync" + + "github.com/sirupsen/logrus" ) type ( @@ -26,6 +28,7 @@ type ( listProcessorWorker[T any] struct { listProcessor ListProcessor[T] + logrus *logrus.Entry scheduler *Scheduler } ) @@ -33,10 +36,12 @@ type ( func NewWorkerFromListProcessor[T any]( listProcessor ListProcessor[T], scheduler *Scheduler, + logrus *logrus.Entry, ) Worker { return &listProcessorWorker[T]{ listProcessor: listProcessor, scheduler: scheduler, + logrus: logrus, } } @@ -75,7 +80,7 @@ func (l *listProcessorWorker[T]) Start(ctx context.Context) error { defer l.scheduler.Return() defer wg.Done() if err := l.listProcessor.Process(ctx, v); err != nil && !errors.Is(err, context.Canceled) { - println("Err", err.Error()) + l.logrus.WithError(err).Error("Error processing batch") } }(v) } |