merch-tracker-v2/task-processor
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

context fix

Browse source
This commit is contained in:
nquidox 2026-02-28 10:53:02 +03:00
parent 4665c90c2c
commit c955615fb1
8 changed files with 39 additions and 52 deletions
Download patch file Download diff file Expand all files Collapse all files

9
cmd/main.go
View file

@ -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)
}

3
internal/parsers/interface.go
View file

@ -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)
}

41
internal/parsers/mandarake/handleTasks.go
View file

@ -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,

15
internal/parsers/mandarake/handler.go
View file

@ -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,
}

9
internal/parsers/mandarake/service.go
View file

@ -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

1
internal/parsers/mandarake/service_test.go
View file

@ -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,
}

8
internal/parsers/surugaya.go
View file

@ -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

5
internal/processor/service.go
View file

@ -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()