update golang test task

golang/main.go

@@ -1,7 +1,10 @@
 package main
 
 import (
+	"bytes"
+	"errors"
 	"fmt"
+	"sync"
 	"time"
 )
 
@@ -15,91 +18,176 @@ import (
 // Обновленный код отправить через pull-request в github
 // Как видите, никаких привязок к внешним сервисам нет - полный карт-бланш на модификацию кода.
 
-// A Ttype represents a meaninglessness of our life
-type Ttype struct {
+var (
+	errTask = errors.New("some error occured")
+)
+
+const (
+	resultOK           = "task has been successed"
+	resultErr          = "something went wrong"
+	workInterval       = 150 * time.Millisecond
+	generationInterval = 10 * time.Second
+	printInterval      = 3 * time.Second
+)
+
+type task struct {
 	id         int
-	cT         string // время создания
-	fT         string // время выполнения
-	taskRESULT []byte
+	createdAt  time.Time
+	finishedAt time.Time
+	result     string
+	err        error
+}
+
+func (t *task) error() string {
+	return fmt.Sprintf("Id: %d created: %s, error: %s, result: %s\n",
+		t.id, t.createdAt.Format(time.RFC3339), t.err, t.result)
+}
+
+func (t *task) success() string {
+	return fmt.Sprintf("id: %d, created: %s, finished: %s, result: %s\n",
+		t.id, t.createdAt.Format(time.RFC3339), t.finishedAt.Format(time.RFC3339), t.result)
 }
 
 func main() {
-	taskCreturer := func(a chan Ttype) {
-		go func() {
-			for {
-				ft := time.Now().Format(time.RFC3339)
-				if time.Now().Nanosecond()%2 > 0 { // вот такое условие появления ошибочных тасков
-					ft = "Some error occured"
+	done := make(chan struct{})
+	tasks := tasksPipe(done)
+
+	w := newWorker(printInterval, tasks)
+	go w.work()
+	go w.sort()
+	go w.print()
+
+	time.Sleep(generationInterval)
+	close(done)
+	w.stop()
+}
+func tasksPipe(
+	done <-chan struct{},
+) <-chan task {
+	tasks := make(chan task)
+
+	go func() {
+		var i int
+		for {
+			select {
+			case <-done:
+				close(tasks)
+				return
+			default:
+				i++
+				task := task{id: i, createdAt: time.Now()}
+
+				if task.createdAt.UnixMicro()%2 > 0 {
+					task.err = errTask
 				}
-				a <- Ttype{cT: ft, id: int(time.Now().Unix())} // передаем таск на выполнение
+
+				tasks <- task
 			}
-		}()
-	}
+		}
+	}()
 
-	superChan := make(chan Ttype, 10)
+	return tasks
+}
 
-	go taskCreturer(superChan)
+type worker struct {
+	printInterval time.Duration
 
-	task_worker := func(a Ttype) Ttype {
-		tt, _ := time.Parse(time.RFC3339, a.cT)
-		if tt.After(time.Now().Add(-20 * time.Second)) {
-			a.taskRESULT = []byte("task has been successed")
-		} else {
-			a.taskRESULT = []byte("something went wrong")
-		}
-		a.fT = time.Now().Format(time.RFC3339Nano)
+	tasks    <-chan task
+	results  chan task
+	shutdown chan struct{}
+	wg       sync.WaitGroup
 
-		time.Sleep(time.Millisecond * 150)
+	errors    *bytes.Buffer
+	successes *bytes.Buffer
+}
 
-		return a
+func newWorker(
+	printInterval time.Duration,
+	tasks <-chan task,
+) *worker {
+	w := worker{
+		printInterval: printInterval,
+		tasks:         tasks,
+		results:       make(chan task),
+		shutdown:      make(chan struct{}),
+		errors:        new(bytes.Buffer),
+		successes:     new(bytes.Buffer),
+		wg:            sync.WaitGroup{},
 	}
 
-	doneTasks := make(chan Ttype)
-	undoneTasks := make(chan error)
+	return &w
+}
+
+func (w *worker) work() {
+	w.wg.Add(1)
+	for {
+		select {
+		case <-w.shutdown:
+			close(w.results)
+			w.wg.Done()
+			return
+		case task := <-w.tasks:
+
+			if task.err != nil {
+				task.result = resultErr
+			} else {
+				task.result = resultOK
+				task.finishedAt = time.Now()
+			}
 
-	tasksorter := func(t Ttype) {
-		if string(t.taskRESULT[14:]) == "successed" {
-			doneTasks <- t
-		} else {
-			undoneTasks <- fmt.Errorf("Task id %d time %s, error %s", t.id, t.cT, t.taskRESULT)
+			w.results <- task
+			time.Sleep(workInterval)
 		}
+
 	}
+}
 
-	go func() {
-		// получение тасков
-		for t := range superChan {
-			t = task_worker(t)
-			go tasksorter(t)
+func (w *worker) sort() {
+	w.wg.Add(1)
+	for {
+		select {
+		case <-w.shutdown:
+			w.wg.Done()
+			return
+		case task := <-w.results:
+			if task.err != nil {
+				w.errors.WriteString(task.error())
+			} else {
+				w.successes.WriteString(task.success())
+			}
 		}
-		close(superChan)
-	}()
+	}
+}
 
-	result := map[int]Ttype{}
-	err := []error{}
-	go func() {
-		for r := range doneTasks {
-			go func() {
-				result[r.id] = r
-			}()
-		}
-		for r := range undoneTasks {
-			go func() {
-				err = append(err, r)
-			}()
+func (w *worker) print() {
+	w.wg.Add(1)
+	for {
+		select {
+		case <-w.shutdown:
+			w.printResults()
+			w.wg.Done()
+			return
+		case <-time.Tick(w.printInterval):
+			w.printResults()
 		}
-		close(doneTasks)
-		close(undoneTasks)
-	}()
-
-	time.Sleep(time.Second * 3)
-
-	println("Errors:")
-	for r := range err {
-		println(r)
 	}
+}
 
-	println("Done tasks:")
-	for r := range result {
-		println(r)
+func (w *worker) printResults() {
+	if w.successes.Len() > 0 {
+		fmt.Println("Successes:")
+		fmt.Println(w.successes.String())
+	}
+	if w.errors.Len() > 0 {
+		fmt.Println("Errors:")
+		fmt.Println(w.errors.String())
 	}
+
+	w.errors.Reset()
+	w.successes.Reset()
+}
+
+func (w *worker) stop() {
+	close(w.shutdown)
+	w.wg.Wait()
 }