task-processor/internal/parsers/mandarake/handleTasks.go

package mandarake

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(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(ctx, s.externalBrowser)
	defer allocCancel()

	receiver := make(chan shared.Task, len(tasks))
	for _, task := range tasks {
		receiver <- task
	}
	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)
		}()
	}
	wg.Wait()

	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) {
	for task := range receiver {
		taskCtx, taskCancel := chromedp.NewContext(ctx /* chromedp.WithLogf(log.Printf) */, chromedp.WithLogf(func(string, ...any) {}))
		timeoutCtx, timeoutCancel := context.WithTimeout(taskCtx, s.taskTimeout)

		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(timeoutCtx, task)
		sender <- shared.TaskResult{
			MerchUuid: task.MerchUuid,
			Origin:    task.Origin,
			Price:     price,
		}

		timeoutCancel()
		taskCancel()
	}
}