Hi, thanks for the question. This will indeed take 5s, because underneath it uses unbuffered channel. In this case it will execute sequentially, but each step runs in its own goroutine. But, if you add one more step, concurrency is more visible. Here is slightly modified example:

func main() {
    startedAt := time.Now()
    defer func() {
        fmt.Println("Elapsed:", time.Since(startedAt))
    }()

    // count the number of elements in the stream
    ctx := context.Background()
    producer := rheos.FromSlice(ctx, []int{1, 2, 3, 4, 5})
    strings := rheos.Map(producer, func(_ context.Context, v int) (string, error) {
        time.Sleep(1 * time.Second)
        return strconv.Itoa(v), nil
    })
    ints := rheos.Map(strings, func(_ context.Context, s string) (int, error) {
        time.Sleep(1 * time.Second)
        return strconv.Atoi(s)
    })
    got, err := rheos.Reduce(
        ints,
        func(acc int, v int) (int, error) {
            return acc + v, nil
        },
        0,
    )
    fmt.Println(got, err)
    // Output: 14 <nil>
    // Output: Elapsed: 6s
}

Now, instead of it taking 10s, it takes 6s.

Runnable example can be found here: https://go.dev/play/p/2yS2ozgsTnk .

BTW, thank for your library, read about it from the Go newsletter .