feat: Merge optimized for N = 0 and N = 1

merge.go

@@ -4,6 +4,14 @@ import "sync"
 
 // Merge fans multiple channels in to a single channel
 func Merge(ins ...<-chan interface{}) <-chan interface{} {
+	// Don't merge anything if we don't have to
+	if l := len(ins); l == 0 {
+		out := make(chan interface{})
+		close(out)
+		return out
+	} else if l == 1 {
+		return ins[0]
+	}
 	out := make(chan interface{})
 	// Create a WaitGroup that waits for all of the ins to close
 	var wg sync.WaitGroup
@@ -22,7 +30,7 @@ func Merge(ins ...<-chan interface{}) <-chan interface{} {
 					out <- i
 				}
 			}
-			// Tell the WaitGroup that one of the errChans is closed
+			// Tell the WaitGroup that one of the channels is closed
 			wg.Done()
 		}(ins[i])
 	}

merge_test.go

@@ -96,6 +96,23 @@ func TestMerge(t *testing.T) {
 			waitFor:    500 * time.Millisecond,
 			errorCount: 1,
 		}},
+	}, {
+		description: "Single channel passes through",
+		expectedErrors: []error{
+			errors.New("[task a] error 0"),
+			errors.New("[task a] error 1"),
+			errors.New("[task a] error 2"),
+		},
+		tasks: []task{{
+			id:      "a",
+			waitFor: 0,
+			// We shoud expect to 'never' receive this error, because it will emit after the maxTestDuration
+			errorCount: 3,
+		}},
+	}, {
+		description:    "Closed channel returned",
+		expectedErrors: []error{},
+		tasks:          []task{},
 	}} {
 		t.Run(test.description, func(t *testing.T) {
 			// Start doing all of the tasks