feat: [main] Reimplemented with a worker pool as external library

golang/cmd/main.go

@@ -0,0 +1,58 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	meaninglessTask "test_task/internal/tasks/meaningless"
+	"test_task/pkg/workerPool"
+	"time"
+)
+
+const (
+	MAX_WORKER_COUNT      = 10
+	LOG_INTERVAL          = 3 * time.Second
+	TASK_CREATION_TIMEOUT = 10 * time.Second
+)
+
+func main() {
+	ctx, cancel := context.WithTimeout(context.Background(), TASK_CREATION_TIMEOUT)
+	defer cancel()
+
+	wp := workerPool.New(MAX_WORKER_COUNT, ctx)
+	wp.WithFactory(func() workerPool.Task {
+		return meaninglessTask.New()
+	})
+
+	wp.Wg.Add(1)
+
+	errChan := make(chan error, 1)
+	go wp.StartFactory(errChan)
+
+	go func() {
+		if err := <-errChan; err != nil {
+			fmt.Printf("Error starting task factory: %v\n", err)
+			os.Exit(1)
+		}
+	}()
+
+	for i := 0; i <= MAX_WORKER_COUNT-1; i++ {
+		wp.Wg.Add(1)
+		go wp.ProcessAndSortTask()
+	}
+
+	ticker := time.NewTicker(LOG_INTERVAL)
+	defer ticker.Stop()
+
+	stopChan := make(chan struct{})
+	go wp.PrintResults(ticker, stopChan)
+
+	wp.Wg.Wait()
+
+	fmt.Println("exiting...")
+	go func() {
+		stopChan <- struct{}{}
+		close(stopChan)
+		wp.Shutdown()
+	}()
+}

golang/go.mod

@@ -0,0 +1,3 @@
+module test_task
+
+go 1.22.2

golang/internal/tasks/meaningless/meaningless.go

@@ -0,0 +1,60 @@
+package meaninglessTask
+
+import (
+	"fmt"
+	"test_task/pkg/workerPool"
+	"time"
+)
+
+const (
+	TASK_SUCCESS_MESSAGE = "task has been successed"
+	TASK_ERROR_MESSAGE   = "something went wrong"
+)
+
+var _ workerPool.Task = (*MeaninglessTask)(nil)
+
+// A MeaninglessTask represents a meaninglessness of our life
+type MeaninglessTask struct {
+	ID           int
+	CreationTime string
+	FinishTime   string
+	result       string
+}
+
+func New() *MeaninglessTask {
+	currentTime := time.Now().Format(time.RFC3339)
+	if time.Now().Nanosecond()%2 > 0 {
+		currentTime = "Some error occured"
+	}
+
+	return &MeaninglessTask{
+		ID:           int(time.Now().Unix()),
+		CreationTime: currentTime,
+	}
+}
+
+func (mt *MeaninglessTask) Process() {
+	tt, _ := time.Parse(time.RFC3339, mt.CreationTime)
+	if tt.After(time.Now().Add(-20 * time.Second)) {
+		mt.result = TASK_SUCCESS_MESSAGE
+	} else {
+		mt.result = TASK_ERROR_MESSAGE
+	}
+	mt.FinishTime = time.Now().Format(time.RFC3339)
+}
+
+func (mt *MeaninglessTask) Result() string {
+	return mt.result
+}
+
+func (mt *MeaninglessTask) IsSuccess() bool {
+	return mt.result == TASK_SUCCESS_MESSAGE
+}
+
+func (mt *MeaninglessTask) Error() error {
+	return fmt.Errorf("ID: %d, Created at: %s, Error: %s", mt.ID, mt.CreationTime, mt.result)
+}
+
+func (mt *MeaninglessTask) Status() string {
+	return fmt.Sprintf("ID: %d, Created at: %s, Finished at: %s", mt.ID, mt.CreationTime, mt.FinishTime)
+}

golang/pkg/workerPool/workerPool.go

@@ -0,0 +1,104 @@
+package workerPool
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+)
+
+type Task interface {
+	Process()
+	Result() string
+	IsSuccess() bool
+	Status() string
+	Error() error
+}
+
+type WorkerPool struct {
+	Wg           sync.WaitGroup
+	workersCount int
+	ctx          context.Context
+	taskFactory  func() Task
+	Tasks        chan Task
+	successTasks chan Task
+	errorTasks   chan error
+}
+
+func New(workerCount int, ctx context.Context) *WorkerPool {
+	return &WorkerPool{
+		Wg:           sync.WaitGroup{},
+		workersCount: workerCount,
+		ctx:          ctx,
+		Tasks:        make(chan Task, 10),
+		successTasks: make(chan Task, 10),
+		errorTasks:   make(chan error, 10),
+	}
+}
+
+func (wp *WorkerPool) WithFactory(taskFactory func() Task) {
+	wp.taskFactory = taskFactory
+}
+
+func (wp *WorkerPool) StartFactory(errChan chan error) {
+	if wp.taskFactory == nil {
+		errChan <- fmt.Errorf("Task factory was not provided")
+		return
+	}
+
+	defer close(wp.Tasks)
+	defer wp.Wg.Done()
+	for {
+		select {
+		case <-wp.ctx.Done():
+			return
+		default:
+			wp.Tasks <- wp.taskFactory()
+		}
+	}
+}
+
+func (wp *WorkerPool) ProcessAndSortTask() {
+	defer wp.Wg.Done()
+	for task := range wp.Tasks {
+		task.Process()
+
+		wp.sortTask(task)
+
+		time.Sleep(time.Millisecond * 150)
+	}
+}
+
+func (wp *WorkerPool) sortTask(t Task) {
+	if t.IsSuccess() {
+		wp.successTasks <- t
+	} else {
+		wp.errorTasks <- t.Error()
+	}
+}
+
+func (wp *WorkerPool) PrintResults(ticker *time.Ticker, stopChan chan struct{}) {
+	for {
+		select {
+		case <-ticker.C:
+			fmt.Println("Success tasks:")
+			for i := 0; i <= len(wp.successTasks); i++ {
+				task := <-wp.successTasks
+				fmt.Println(task.Status())
+			}
+
+			fmt.Println("Errors:")
+			for i := 0; i <= len(wp.successTasks); i++ {
+				err := <-wp.errorTasks
+				fmt.Println(err)
+			}
+		case <-stopChan:
+			return
+		}
+	}
+}
+
+func (wp *WorkerPool) Shutdown() {
+	close(wp.successTasks)
+	close(wp.errorTasks)
+}