HW6 is completed

hw06_pipeline_execution/pipeline.go

@@ -9,6 +9,30 @@ type (
 type Stage func(in In) (out Out)
 
 func ExecutePipeline(in In, done In, stages ...Stage) Out {
-	// Place your code here.
-	return nil
+	if in == nil {
+		return nil
+	}
+
+	for _, st := range stages {
+		tmpChan := make(chan interface{})
+		go func(ch In) {
+			defer close(tmpChan)
+
+			for {
+				select {
+				case item, ok := <-ch:
+					if !ok {
+						return
+					}
+					tmpChan <- item
+				case <-done:
+					return
+				}
+			}
+		}(in)
+
+		in = st(tmpChan)
+	}
+
+	return in
 }

hw06_pipeline_execution/pipeline_test.go

@@ -5,7 +5,7 @@
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/require"
 )
 
 const (
@@ -36,6 +36,35 @@
 		g("Stringifier", func(v interface{}) interface{} { return strconv.Itoa(v.(int)) }),
 	}
 
+	t.Run("empty stages list", func(t *testing.T) {
+		in := make(Bi)
+		data := []string{"a", "b", "c", "d", "e"}
+
+		go func() {
+			for _, v := range data {
+				in <- v
+			}
+			close(in)
+		}()
+
+		result := make([]string, 0, len(data))
+		start := time.Now()
+		for s := range ExecutePipeline(in, nil, []Stage{}...) {
+			result = append(result, s.(string))
+		}
+		elapsed := time.Since(start)
+
+		require.Equal(t, data, result)
+		require.Less(t,
+			int64(elapsed),
+			int64(sleepPerStage)+int64(fault))
+	})
+
+	t.Run("nil check", func(t *testing.T) {
+		res := ExecutePipeline(nil, nil, stages...)
+		require.Nil(t, res)
+	})
+
 	t.Run("simple case", func(t *testing.T) {
 		in := make(Bi)
 		data := []int{1, 2, 3, 4, 5}