Function TestMergeNamed_SourceEOF

pkg/stream/stream_test.go:196–242

(t *testing.T)

}

func TestMergeNamed_SourceEOF(t *testing.T) {
	r1, w1 := Pipe[string](2)
	r2, w2 := Pipe[string](2)

	go func() {
		w1.Send("a", nil)
		w1.Close()
	}()
	go func() {
		w2.Send("b", nil)
		w2.Send("c", nil)
		w2.Close()
	}()

	merged := MergeNamed(map[string]*Reader[string]{
		"stream1": r1,
		"stream2": r2,
	})
	defer merged.Close()

	var values []string
	var eofSources []string

	for {
		v, err := merged.Recv()
		if errors.Is(err, io.EOF) {
			break
		}
		if name, ok := GetSourceName(err); ok {
			eofSources = append(eofSources, name)
			continue
		}
		if err != nil {
			t.Fatalf("unexpected error: %v", err)
		}
		values = append(values, v)
	}

	if len(values) != 3 {
		t.Errorf("expected 3 values, got %d: %v", len(values), values)
	}

	// 至少应该报告一个源 EOF
	if len(eofSources) == 0 {
		t.Error("expected at least one source EOF")
	}
}

func TestTransform_Basic(t *testing.T) {
	reader := FromSlice([]int{1, 2, 3, 4, 5})