HW6 Pipeline

hw06_pipeline_execution/pipeline.go

@@ -9,6 +9,39 @@ type (
 type Stage func(in In) (out Out)
 
 func ExecutePipeline(in In, done In, stages ...Stage) Out {
-	// Place your code here.
-	return nil
+	if len(stages) == 0 {
+		return in
+	}
+
+	proxyChan := func(inChan In, doneChan In) Out {
+		out := make(Bi)
+
+		go func() {
+			defer func() {
+				close(out)
+				//nolint:revive
+				for range inChan {
+				}
+			}()
+			for {
+				select {
+				case <-doneChan:
+					return
+				case v, ok := <-inChan:
+					if !ok {
+						return
+					}
+					out <- v
+				}
+			}
+		}()
+		return out
+	}
+
+	outChan := in
+	for i := 0; i < len(stages); i++ {
+		outChan = stages[i](proxyChan(outChan, done))
+	}
+
+	return outChan
 }

hw06_pipeline_execution/pipeline_test.go

@@ -150,6 +150,51 @@ func TestAllStageStop(t *testing.T) {
 		wg.Wait()
 
 		require.Len(t, result, 0)
-
 	})
 }
+
+func TestEdgeInputs(t *testing.T) {
+	stages := []Stage{
+		func(in In) Out {
+			out := make(Bi)
+			go func() {
+				defer close(out)
+				for v := range in {
+					out <- v
+				}
+			}()
+			return out
+		},
+	}
+
+	// closed channel
+	in := make(Bi)
+	close(in)
+
+	result := make([]interface{}, 0)
+	for s := range ExecutePipeline(in, nil, stages...) {
+		result = append(result, s)
+	}
+
+	require.Len(t, result, 0)
+
+	// empty stages
+	in = make(Bi)
+	go func() {
+		for i := 0; i < 10; i++ {
+			in <- i
+		}
+		close(in)
+	}()
+
+	stages = []Stage{}
+	resultEmpty := make([]int, 0)
+	for s := range ExecutePipeline(in, nil, stages...) {
+		resultEmpty = append(resultEmpty, s.(int))
+	}
+
+	require.Len(t, resultEmpty, 10)
+	for i := 0; i < 10; i++ {
+		require.Equal(t, i, resultEmpty[i])
+	}
+}