From c955615fb1b896412eac47bfac272342c62547c2 Mon Sep 17 00:00:00 2001 From: nquidox Date: Sat, 28 Feb 2026 10:53:02 +0300 Subject: [PATCH] context fix --- cmd/main.go | 9 ++++- internal/parsers/interface.go | 3 +- internal/parsers/mandarake/handleTasks.go | 41 +++++++++------------- internal/parsers/mandarake/handler.go | 15 ++++---- internal/parsers/mandarake/service.go | 9 +---- internal/parsers/mandarake/service_test.go | 1 - internal/parsers/surugaya.go | 8 ++--- internal/processor/service.go | 5 +-- 8 files changed, 39 insertions(+), 52 deletions(-) diff --git a/cmd/main.go b/cmd/main.go index 1fd8de8..92ae164 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -1,17 +1,24 @@ package main import ( + "context" + "os" + "os/signal" + "syscall" "task-processor/config" "task-processor/internal/app" "task-processor/internal/logging" ) func main() { + ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer cancel() + c := config.NewConfig() logging.LogSetup(c.LogLevel) appl := app.New(c) - appl.Run() + appl.Run(ctx) } diff --git a/internal/parsers/interface.go b/internal/parsers/interface.go index 453d619..1a4c6fe 100644 --- a/internal/parsers/interface.go +++ b/internal/parsers/interface.go @@ -1,10 +1,11 @@ package parsers import ( + "context" "task-processor/internal/appState" "task-processor/internal/shared" ) type TaskHandler interface { - HandleTasks(tasks []shared.Task, sender chan shared.TaskResult, state *appState.State) + HandleTasks(ctx context.Context, tasks []shared.Task, sender chan shared.TaskResult, state *appState.State) } diff --git a/internal/parsers/mandarake/handleTasks.go b/internal/parsers/mandarake/handleTasks.go index 1920554..1b566e7 100644 --- a/internal/parsers/mandarake/handleTasks.go +++ b/internal/parsers/mandarake/handleTasks.go @@ -4,15 +4,20 @@ import ( "context" "github.com/chromedp/chromedp" log "github.com/sirupsen/logrus" + "runtime" "sync" "task-processor/internal/appState" "task-processor/internal/shared" ) -func (s *Parser) HandleTasks(tasks []shared.Task, sender chan shared.TaskResult, state *appState.State) { - log.Debug(logHeader + logWorker + "handling tasks") +func (s *Parser) HandleTasks(ctx context.Context, tasks []shared.Task, sender chan shared.TaskResult, state *appState.State) { + log.Infof("%v %v handling tasks", logHeader, logWorker) - allocCtx, allocCancel := chromedp.NewRemoteAllocator(s.baseCtx, s.externalBrowser) + allocCtx, allocCancel := chromedp.NewRemoteAllocator(ctx, s.externalBrowser) + defer allocCancel() + + sessionCtx, sessionCancel := chromedp.NewContext(allocCtx /* chromedp.WithLogf(log.Printf) */, chromedp.WithLogf(func(string, ...any) {})) + defer sessionCancel() receiver := make(chan shared.Task, len(tasks)) for _, task := range tasks { @@ -20,41 +25,27 @@ func (s *Parser) HandleTasks(tasks []shared.Task, sender chan shared.TaskResult, } close(receiver) + log.Debugf("%v gorutines before wait group: %v", logHeader, runtime.NumGoroutine()) wg := sync.WaitGroup{} for i := 0; i < s.goroutinesNumber; i++ { wg.Add(1) go func() { defer wg.Done() - s.worker(allocCtx, receiver, sender, state) + s.worker(sessionCtx, receiver, sender) }() } wg.Wait() - allocCancel() - log.Debug(logHeader + logWorker + "finished handling tasks") + + log.Debugf("%v gorutines after wait group: %v", logHeader, runtime.NumGoroutine()) + log.Infof(logHeader + logWorker + "finished handling tasks") } -func (s *Parser) worker(ctx context.Context, receiver chan shared.Task, sender chan shared.TaskResult, state *appState.State) { +func (s *Parser) worker(ctx context.Context, receiver chan shared.Task, sender chan shared.TaskResult) { for task := range receiver { - log.WithField("task_uuid", task.MerchUuid).Debug(logHeader + logWorker + "processing task") - - //pageCtx, pageCancel := chromedp.NewContext(ctx, chromedp.WithLogf(func(string, ...any) {})) - // - //price, err := s.getPrice(pageCtx, task) - //pageCancel() - - //price, err := s.getMinimalPrice(task) - //if err != nil { - // log.WithField("task_uuid", task.MerchUuid).Warn(logHeader + logWorker + logTaskWarning + "failed to process, zero price") - // sender <- shared.TaskResult{ - // MerchUuid: task.MerchUuid, - // Origin: task.Origin, - // Price: zeroPrice, - // } - // continue - //} + log.WithField("task_uuid", task.MerchUuid).Infof("%v %v processing task", logHeader, logWorker) //price will be zeroPrice value in case of any error or if price not found - price := s.getMinimalPrice(task) + price := s.getMinimalPrice(ctx, task) sender <- shared.TaskResult{ MerchUuid: task.MerchUuid, Origin: task.Origin, diff --git a/internal/parsers/mandarake/handler.go b/internal/parsers/mandarake/handler.go index 92e24dd..cd6b272 100644 --- a/internal/parsers/mandarake/handler.go +++ b/internal/parsers/mandarake/handler.go @@ -1,39 +1,36 @@ package mandarake import ( - "context" log "github.com/sirupsen/logrus" ) const ( zeroPrice int32 = 0 taxMultiplier float64 = 1.1 - logHeader = "Mandarake parser | " - logWorker = "worker: " - logTaskWarning = "task warning: " - logGetPrice = "get price: " + logHeader = "Mandarake parser |" + logWorker = "worker:" + logTaskWarning = "task warning:" + logGetPrice = "get price:" ) type Parser struct { - baseCtx context.Context externalBrowser string goroutinesNumber int } -type ParserDeps struct { +type Deps struct { Enabled bool ExternalBrowser string GoroutinesNumber int } -func NewParser(deps ParserDeps) *Parser { +func NewParser(deps Deps) *Parser { if !deps.Enabled { log.Info(logHeader + "disabled") return nil } return &Parser{ - baseCtx: context.Background(), externalBrowser: deps.ExternalBrowser, goroutinesNumber: deps.GoroutinesNumber, } diff --git a/internal/parsers/mandarake/service.go b/internal/parsers/mandarake/service.go index 3147fa1..1e47a3c 100644 --- a/internal/parsers/mandarake/service.go +++ b/internal/parsers/mandarake/service.go @@ -41,14 +41,7 @@ func (s *Parser) getPrice(ctx context.Context, task shared.Task) (int32, error) return minimal, nil } -func (s *Parser) getMinimalPrice(task shared.Task) int32 { - ctx := context.Background() - allocCtx, allocCancel := chromedp.NewRemoteAllocator(ctx, s.externalBrowser) - defer allocCancel() - - sessionCtx, sessionCancel := chromedp.NewContext(allocCtx, chromedp.WithLogf(log.Printf)) - defer sessionCancel() - +func (s *Parser) getMinimalPrice(sessionCtx context.Context, task shared.Task) int32 { var ( singlePrice string rangedPrice string diff --git a/internal/parsers/mandarake/service_test.go b/internal/parsers/mandarake/service_test.go index c0f5a79..736331a 100644 --- a/internal/parsers/mandarake/service_test.go +++ b/internal/parsers/mandarake/service_test.go @@ -65,7 +65,6 @@ func TestParser_processPrices(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { s := &Parser{ - baseCtx: tt.fields.baseCtx, externalBrowser: tt.fields.externalBrowser, goroutinesNumber: tt.fields.goroutinesNumber, } diff --git a/internal/parsers/surugaya.go b/internal/parsers/surugaya.go index 9dc3c5f..1fecccf 100644 --- a/internal/parsers/surugaya.go +++ b/internal/parsers/surugaya.go @@ -11,21 +11,19 @@ import ( type SurugayaParser struct { scrapper sc.SurugayaScrapperClient - ctx context.Context } -func NewSurugayaParser(ctx context.Context, scrapper sc.SurugayaScrapperClient) *SurugayaParser { +func NewSurugayaParser(scrapper sc.SurugayaScrapperClient) *SurugayaParser { log.Debug("Surugaya parser init") return &SurugayaParser{ scrapper: scrapper, - ctx: ctx, } } -func (s *SurugayaParser) HandleTasks(tasks []shared.Task, sender chan shared.TaskResult, state *appState.State) { +func (s *SurugayaParser) HandleTasks(ctx context.Context, tasks []shared.Task, sender chan shared.TaskResult, state *appState.State) { log.WithField("count", len(tasks)).Debug("Handling Surugaya Tasks") - stream, err := s.scrapper.ProcessTasks(s.ctx) + stream, err := s.scrapper.ProcessTasks(ctx) if err != nil { log.WithField("err", err).Error("Error creating stream") return diff --git a/internal/processor/service.go b/internal/processor/service.go index 4748566..61e7fe0 100644 --- a/internal/processor/service.go +++ b/internal/processor/service.go @@ -1,13 +1,14 @@ package processor import ( + "context" log "github.com/sirupsen/logrus" "sync" "task-processor/internal/appState" "task-processor/internal/shared" ) -func (p *Processor) StartWork(receivedTasks []shared.TaskResponse) { +func (p *Processor) StartWork(ctx context.Context, receivedTasks []shared.TaskResponse) { log.Info("Starting work...") p.state.ResetCounters() @@ -27,7 +28,7 @@ func (p *Processor) StartWork(receivedTasks []shared.TaskResponse) { go func(origin string, tasks []shared.Task) { defer wg.Done() log.Info("Running task handler for origin: ", origin) - p.handlers[origin].HandleTasks(tasks, p.out, p.state) + p.handlers[origin].HandleTasks(ctx, tasks, p.out, p.state) }(origin, tasks) } wg.Wait()