huge refactor

internal/processor/handler.go

@@ -2,10 +2,9 @@ package processor
 
 import (
 	"context"
-	"parsing-service/internal/appState"
-	"parsing-service/internal/parsers"
-	"parsing-service/internal/shared"
-	pb "parsing-service/proto/taskProcessor"
+	"task-processor/internal/appState"
+	"task-processor/internal/parsers"
+	"task-processor/internal/shared"
 )
 
 type Processor struct {
@@ -13,7 +12,6 @@ type Processor struct {
 	out      chan shared.TaskResult
 	state    *appState.State
 	ctx      context.Context
-	client   pb.TaskProcessorClient
 	numCPUs  int
 }
 
@@ -22,7 +20,6 @@ type Deps struct {
 	Out      chan shared.TaskResult
 	State    *appState.State
 	Ctx      context.Context
-	Client   pb.TaskProcessorClient
 	NumCPUs  int
 }
 
@@ -32,7 +29,6 @@ func New(deps Deps) *Processor {
 		out:      deps.Out,
 		state:    deps.State,
 		ctx:      deps.Ctx,
-		client:   deps.Client,
 		numCPUs:  deps.NumCPUs,
 	}
 }

internal/processor/service.go

@@ -2,34 +2,33 @@ package processor
 
 import (
 	log "github.com/sirupsen/logrus"
-	"parsing-service/internal/appState"
-	"parsing-service/internal/shared"
 	"sync"
+	"task-processor/internal/appState"
+	"task-processor/internal/shared"
 )
 
 func (p *Processor) StartWork(receivedTasks []shared.TaskResponse) {
 	log.Info("Starting work...")
 	p.state.ResetCounters()
 
-	in := make(chan shared.Task, p.numCPUs*10)
+	if len(receivedTasks) == 0 {
+		p.state.SetStatus(appState.StatusIdle)
+		return
+	}
 
-	wg := &sync.WaitGroup{}
-	for i := 0; i < p.numCPUs*10; i++ {
+	p.state.SetStatus(appState.StatusWorkInProgress)
+	p.state.SetTasksReceived(len(receivedTasks))
+
+	sorted := p.sortTasks(receivedTasks)
+
+	var wg sync.WaitGroup
+	for origin, tasks := range sorted {
 		wg.Add(1)
-		go func() {
+		go func(origin string, tasks []shared.Task) {
 			defer wg.Done()
-			p.worker(in)
-		}()
+			p.handlers[origin].HandleTasks(tasks, p.out, p.state)
+		}(origin, tasks)
 	}
-
-	tasksNumber := len(receivedTasks)
-	if tasksNumber > 0 {
-		p.state.SetStatus(appState.StatusWorkInProgress)
-		p.state.SetTasksReceived(tasksNumber)
-		p.sortTasks(in, receivedTasks)
-	}
-
-	close(in)
 	wg.Wait()
 
 	log.Debug("All goroutines finished")
@@ -38,23 +37,24 @@ func (p *Processor) StartWork(receivedTasks []shared.TaskResponse) {
 	log.Debugf("State | %+v", p.state)
 }
 
-func (p *Processor) sortTasks(in chan<- shared.Task, receivedTasks []shared.TaskResponse) {
+func (p *Processor) sortTasks(receivedTasks []shared.TaskResponse) map[string][]shared.Task {
+	sorted := make(map[string][]shared.Task, len(receivedTasks))
 	for _, task := range receivedTasks {
 		switch {
 		case task.OriginSurugayaLink != "":
-			in <- shared.Task{
-				MerchUuid:  task.MerchUuid,
-				Origin:     shared.OriginSurugaya,
-				Link:       task.OriginSurugayaLink,
-				RetryCount: 3,
-			}
+			sorted[shared.OriginSurugaya] = append(sorted[shared.OriginSurugaya], shared.Task{
+				MerchUuid: task.MerchUuid,
+				Origin:    shared.OriginSurugaya,
+				Link:      task.OriginSurugayaLink,
+			})
 		case task.OriginMandarakeLink != "":
-			in <- shared.Task{
+			sorted[shared.OriginMandarake] = append(sorted[shared.OriginMandarake], shared.Task{
 				MerchUuid:  task.MerchUuid,
 				Origin:     shared.OriginMandarake,
 				Link:       task.OriginMandarakeLink,
 				RetryCount: 3,
-			}
+			})
 		}
 	}
+	return sorted
 }

internal/processor/worker.go

@@ -1,18 +0,0 @@
-package processor
-
-import (
-	log "github.com/sirupsen/logrus"
-	"parsing-service/internal/shared"
-)
-
-const zeroPrice = 0 //for debug purposes
-
-func (p *Processor) worker(in <-chan shared.Task) {
-	for task := range in {
-		err := p.handlers[task.Origin].HandleTask(task, p.out, p.state)
-		if err != nil {
-			log.WithField("err", err).Error("Worker | Handle task")
-			continue
-		}
-	}
-}