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

package mandarake

import (
	"context"
	"github.com/chromedp/chromedp"
	log "github.com/sirupsen/logrus"
	"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")

	allocCtx, allocCancel := chromedp.NewRemoteAllocator(s.baseCtx, s.externalBrowser)

	receiver := make(chan shared.Task, len(tasks))
	for _, task := range tasks {
		receiver <- task
	}
	close(receiver)

	wg := sync.WaitGroup{}
	for i := 0; i < s.goroutinesNumber; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			s.worker(allocCtx, receiver, sender, state)
		}()
	}
	wg.Wait()
	allocCancel()
	log.Debug(logHeader + logWorker + "finished handling tasks")
}

func (s *Parser) worker(ctx context.Context, receiver chan shared.Task, sender chan shared.TaskResult, state *appState.State) {
	for task := range receiver {
		log.WithField("task_uuid", task.MerchUuid).Debug(logHeader + logWorker + "processing task")

		pageCtx, pageCancel := chromedp.NewContext(ctx, chromedp.WithLogf(log.Printf))

		price, err := s.getPrice(pageCtx, task)
		pageCancel()
		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
		}

		sender <- shared.TaskResult{
			MerchUuid: task.MerchUuid,
			Origin:    task.Origin,
			Price:     price,
		}
	}
}