From e12d2a777bf3c935c4c2bbb4196b1dd043908c69 Mon Sep 17 00:00:00 2001
From: jpfourny <jpfourny@gmail.com>
Date: Wed, 10 Jan 2024 22:44:48 -0500
Subject: [PATCH] Move mapper and pred packages under stream package; refactor
 stream.Reduce

---
 examples/stream/map_test.go               |  12 +-
 pkg/stream/aggregate.go                   | 219 +++++++++-------------
 pkg/stream/aggregate_test.go              |  68 +++----
 pkg/stream/combine.go                     |   4 +-
 pkg/stream/iterate_test.go                |   4 +-
 pkg/{ => stream}/mapper/condition.go      |   0
 pkg/{ => stream}/mapper/condition_test.go |   2 +-
 pkg/{ => stream}/mapper/convert.go        |   0
 pkg/{ => stream}/mapper/convert_test.go   |   0
 pkg/{ => stream}/mapper/doc.go            |   0
 pkg/{ => stream}/mapper/format.go         |   0
 pkg/{ => stream}/mapper/format_test.go    |   0
 pkg/{ => stream}/mapper/misc.go           |   0
 pkg/{ => stream}/mapper/misc_test.go      |   0
 pkg/{ => stream}/mapper/parse.go          |   0
 pkg/{ => stream}/mapper/parse_test.go     |   0
 pkg/{ => stream}/mapper/pointer.go        |   0
 pkg/{ => stream}/mapper/pointer_test.go   |   0
 pkg/{ => stream}/mapper/slice.go          |   0
 pkg/{ => stream}/mapper/slice_test.go     |   0
 pkg/stream/match.go                       |  14 +-
 pkg/stream/misc.go                        |   3 +-
 pkg/{ => stream}/pred/compare.go          |   0
 pkg/{ => stream}/pred/compare_test.go     |   0
 pkg/{ => stream}/pred/doc.go              |   0
 pkg/{ => stream}/pred/in.go               |   0
 pkg/{ => stream}/pred/in_test.go          |   0
 pkg/{ => stream}/pred/logic.go            |   0
 pkg/{ => stream}/pred/logic_test.go       |   0
 pkg/{ => stream}/pred/string.go           |   0
 pkg/{ => stream}/pred/string_test.go      |   0
 pkg/{ => stream}/pred/zero.go             |   0
 pkg/{ => stream}/pred/zero_test.go        |   0
 pkg/stream/reducer/reducer.go             |  53 ++++++
 pkg/stream/transform_test.go              |   6 +-
 35 files changed, 196 insertions(+), 189 deletions(-)
 rename pkg/{ => stream}/mapper/condition.go (100%)
 rename pkg/{ => stream}/mapper/condition_test.go (97%)
 rename pkg/{ => stream}/mapper/convert.go (100%)
 rename pkg/{ => stream}/mapper/convert_test.go (100%)
 rename pkg/{ => stream}/mapper/doc.go (100%)
 rename pkg/{ => stream}/mapper/format.go (100%)
 rename pkg/{ => stream}/mapper/format_test.go (100%)
 rename pkg/{ => stream}/mapper/misc.go (100%)
 rename pkg/{ => stream}/mapper/misc_test.go (100%)
 rename pkg/{ => stream}/mapper/parse.go (100%)
 rename pkg/{ => stream}/mapper/parse_test.go (100%)
 rename pkg/{ => stream}/mapper/pointer.go (100%)
 rename pkg/{ => stream}/mapper/pointer_test.go (100%)
 rename pkg/{ => stream}/mapper/slice.go (100%)
 rename pkg/{ => stream}/mapper/slice_test.go (100%)
 rename pkg/{ => stream}/pred/compare.go (100%)
 rename pkg/{ => stream}/pred/compare_test.go (100%)
 rename pkg/{ => stream}/pred/doc.go (100%)
 rename pkg/{ => stream}/pred/in.go (100%)
 rename pkg/{ => stream}/pred/in_test.go (100%)
 rename pkg/{ => stream}/pred/logic.go (100%)
 rename pkg/{ => stream}/pred/logic_test.go (100%)
 rename pkg/{ => stream}/pred/string.go (100%)
 rename pkg/{ => stream}/pred/string_test.go (100%)
 rename pkg/{ => stream}/pred/zero.go (100%)
 rename pkg/{ => stream}/pred/zero_test.go (100%)
 create mode 100644 pkg/stream/reducer/reducer.go

diff --git a/examples/stream/map_test.go b/examples/stream/map_test.go
index f20fe11..a4fb4aa 100644
--- a/examples/stream/map_test.go
+++ b/examples/stream/map_test.go
@@ -2,9 +2,9 @@ package stream
 
 import (
 	"fmt"
+	mapper2 "github.com/jpfourny/papaya/pkg/stream/mapper"
 	"testing"
 
-	"github.com/jpfourny/papaya/pkg/mapper"
 	"github.com/jpfourny/papaya/pkg/pair"
 	"github.com/jpfourny/papaya/pkg/stream"
 )
@@ -13,7 +13,7 @@ func TestMapIntToString(t *testing.T) {
 	// Map stream of int to string.
 	s := stream.Map(
 		stream.Of(1, 2, 3),
-		mapper.Sprintf[int]("%d"),
+		mapper2.Sprintf[int]("%d"),
 	)
 	stream.ForEach(s, func(s string) {
 		fmt.Println(s)
@@ -28,7 +28,7 @@ func TestMapStringToInt(t *testing.T) {
 	// Map stream of string to int; default to 0 if parse fails.
 	s := stream.Map(
 		stream.Of("1", "2", "3", "foo"),
-		mapper.ParseIntOr[string, int](10, 64, -1),
+		mapper2.ParseIntOr[string, int](10, 64, -1),
 	)
 	stream.ForEach(s, func(i int) {
 		fmt.Println(i)
@@ -41,7 +41,7 @@ func TestMapStringToInt(t *testing.T) {
 
 	s = stream.MapOrDiscard(
 		stream.Of("1", "2", "3", "foo"),
-		mapper.TryParseInt[string, int](10, 64),
+		mapper2.TryParseInt[string, int](10, 64),
 	)
 	stream.ForEach(s, func(i int) {
 		fmt.Println(i)
@@ -107,7 +107,7 @@ func TestMapNumberToBool(t *testing.T) {
 	// Given stream of numbers, map each number to whether it is even.
 	s := stream.Map(
 		stream.Of(0, 2, 0),
-		mapper.NumberToBool[int, bool](),
+		mapper2.NumberToBool[int, bool](),
 	)
 	stream.ForEach(s, func(b bool) {
 		fmt.Println(b)
@@ -122,7 +122,7 @@ func TestMapBoolNumber(t *testing.T) {
 	// Given stream of bools, map each bool to 0 if false and 1 if true.
 	s := stream.Map(
 		stream.Of(false, true, false),
-		mapper.BoolToNumber[bool, int](),
+		mapper2.BoolToNumber[bool, int](),
 	)
 	stream.ForEach(s, func(i int) {
 		fmt.Println(i)
diff --git a/pkg/stream/aggregate.go b/pkg/stream/aggregate.go
index 625f31b..2b7abd5 100644
--- a/pkg/stream/aggregate.go
+++ b/pkg/stream/aggregate.go
@@ -4,21 +4,14 @@ import (
 	"github.com/jpfourny/papaya/pkg/cmp"
 	"github.com/jpfourny/papaya/pkg/constraint"
 	"github.com/jpfourny/papaya/pkg/optional"
+	"github.com/jpfourny/papaya/pkg/stream/mapper"
+	"github.com/jpfourny/papaya/pkg/stream/reducer"
 )
 
 // Reducer represents a function that takes two inputs of type E and returns an output of type E.
 // The Reducer is commonly used in the `Reduce` function to combine elements of a stream into a single result.
 type Reducer[E any] func(e1, e2 E) (result E)
 
-// Accumulator represents a function that takes an accumulated value of type A and an element of type E,
-// and returns the updated accumulated value of type A.
-// The Accumulator is commonly used in the `Aggregate` function to combine elements of a stream into a single result.
-type Accumulator[A, E any] func(a A, e E) (result A)
-
-// Finisher represents a function that takes an accumulated value of type A and returns the finished result of type F.
-// The Finisher is commonly used in the `Aggregate` function to compute the final result after all elements have been accumulated.
-type Finisher[A, F any] func(a A) (result F)
-
 // Reduce combines the elements of the stream into a single value using the given reducer function.
 // If the stream is empty, then an empty optional.Optional is returned.
 // The stream is fully consumed.
@@ -38,131 +31,48 @@ type Finisher[A, F any] func(a A) (result F)
 //	    return a + e
 //	  },
 //	) // None()
-func Reduce[E any](s Stream[E], reduce Reducer[E]) (result optional.Optional[E]) {
-	result = optional.Empty[E]()
+func Reduce[E any](s Stream[E], reduce Reducer[E]) optional.Optional[E] {
+	var accum E
+	var ok bool
 	s(func(e E) bool {
-		if result.Present() {
-			result = optional.Of(reduce(result.Get(), e))
+		if ok {
+			accum = reduce(accum, e)
 		} else {
-			result = optional.Of(e)
+			accum = e
+			ok = true
 		}
 		return true
 	})
-	return
-}
-
-// Aggregate combines the elements of the stream into a single value using the given identity value, accumulator function and finisher function.
-// The accumulated value is initialized to the identity value.
-// The accumulator function is used to combine each element with the accumulated value.
-// The finisher function is used to compute the final result after all elements have been accumulated.
-// The stream is fully consumed.
-//
-// Example usage:
-//
-//	s := stream.Aggregate(
-//	  stream.Of(1, 2, 3),
-//	  0,                 // Initial value
-//	  func(a, e int) int {
-//	    return a + e     // Accumulate with addition
-//	  },
-//	  func(a int) int {
-//	    return a * 2     // Finish with multiplication by 2
-//	  },
-//	) // (1+2+3) * 2 = 12
-func Aggregate[E, A, F any](s Stream[E], identity A, accumulate Accumulator[A, E], finish Finisher[A, F]) F {
-	a := identity
-	s(func(e E) bool {
-		a = accumulate(a, e)
-		return true
-	})
-	return finish(a)
+	return optional.Maybe(accum, ok)
 }
 
-// Sum computes the sum of all elements in the stream of any number type E and returns the result as type E.
-// The result of an empty stream is the zero value of type E.
+// Sum computes the sum of all elements in the stream of any real-number type E and returns the result as real-number type F.
+// The result of an empty stream is the zero value of type F.
 // The stream is fully consumed.
 //
 // Example usage:
 //
-//	n := stream.Sum(stream.Of(1, 2, 3)) // 6 (int)
-func Sum[E constraint.Numeric](s Stream[E]) E {
+//	n1 := stream.Sum[int](stream.Of(1, 2, 3)) // 6 (int)
+//	n2 := stream.Sum[float64](stream.Of(1, 2, 3)) // 6.0 (float64)
+func Sum[R, E constraint.RealNumber](s Stream[E]) R {
 	return Reduce(
-		s,
-		func(a E, e E) E { return a + e },
-	).OrElse(E(0))
+		Map(s, mapper.NumberToNumber[E, R]()),
+		reducer.Sum[R](),
+	).OrElseZero()
 }
 
-// SumInteger computes the sum of all elements in the stream of any signed-integer type E and returns the result as type int64.
-// The result of an empty stream is the zero value of type int64.
+// SumComplex computes the sum of all elements in the stream of any complex-number type E and returns the result as complex-number type F.
+// The result of an empty stream is the zero value of type F.
 // The stream is fully consumed.
 //
 // Example usage:
 //
-//	n := stream.SumInteger(stream.Of(1, 2, 3)) // 6 (int64)
-func SumInteger[E constraint.SignedInteger](s Stream[E]) int64 {
-	return Aggregate(
-		s,
-		int64(0),
-		func(a int64, e E) int64 { return a + int64(e) },
-		func(a int64) int64 { return a },
-	)
-}
-
-// SumUnsignedInteger computes the sum of all elements in the stream of any unsigned-integer type E and returns the result as type uint64.
-// The result of an empty stream is the zero value of type uint64.
-// The stream is fully consumed.
-//
-// Example usage:
-//
-//	n := stream.SumUnsignedInteger(stream.Of[uint](1, 2, 3)) // 6 (uint64)
-func SumUnsignedInteger[E constraint.UnsignedInteger](s Stream[E]) uint64 {
-	return Aggregate(
-		s,
-		uint64(0),
-		func(a uint64, e E) uint64 { return a + uint64(e) },
-		func(a uint64) uint64 { return a },
-	)
-}
-
-// SumFloat computes the sum of all elements in the stream of any floating-point type E and returns the result as type float64.
-// The result of an empty stream is the zero value of type float64.
-// The stream is fully consumed.
-//
-// Example usage:
-//
-//	n := stream.SumFloat(stream.Of(1.0, 2.0, 3.0)) // 6.0 (float64)
-func SumFloat[E constraint.RealNumber](s Stream[E]) float64 {
-	return Aggregate(
-		s,
-		float64(0),
-		func(a float64, e E) float64 { return a + float64(e) },
-		func(a float64) float64 { return a },
-	)
-}
-
-// Average computes the average of all elements in the stream of any number type E and returns the result as type float64.
-// The result of an empty stream is the zero value of type float64.
-// The stream is fully consumed.
-//
-// Example usage:
-//
-//	n := stream.Average(stream.Of(1, 2, 3)) // 2.0 (float64)
-func Average[E constraint.RealNumber](s Stream[E]) float64 {
-	var count uint64
-	return Aggregate(
-		s,
-		float64(0),
-		func(a float64, e E) float64 {
-			count++
-			return a + float64(e)
-		},
-		func(a float64) float64 {
-			if count == 0 {
-				return 0
-			}
-			return a / float64(count)
-		},
-	)
+//	n := stream.SumComplex[complex128](stream.Of(1+i, 2+i, 3+i)) // 6+3i (complex128)
+func SumComplex[R, E constraint.Complex](s Stream[E]) R {
+	return Reduce(
+		Map(s, mapper.ComplexToComplex[E, R]()),
+		reducer.Sum[R](),
+	).OrElseZero()
 }
 
 // Min returns the minimum element in the stream, or the zero value of the type parameter E if the stream is empty.
@@ -174,7 +84,7 @@ func Average[E constraint.RealNumber](s Stream[E]) float64 {
 //	min := stream.Min(stream.Of(3, 1, 2)) // Some(1)
 //	min = stream.Min(stream.Empty[int]()) // None()
 func Min[E constraint.Ordered](s Stream[E]) (min optional.Optional[E]) {
-	return MinBy(s, cmp.Natural[E]())
+	return Reduce(s, reducer.Min[E]())
 }
 
 // MinBy returns the minimum element in the stream.
@@ -186,15 +96,7 @@ func Min[E constraint.Ordered](s Stream[E]) (min optional.Optional[E]) {
 //	min := stream.MinBy(stream.Of(3, 1, 2), cmp.Natural[int]()) // Some(1)
 //	min = stream.MinBy(stream.Empty[int](), cmp.Natural[int]()) // None()
 func MinBy[E any](s Stream[E], compare cmp.Comparer[E]) (min optional.Optional[E]) {
-	return Reduce(
-		s,
-		func(a, e E) E {
-			if compare.LessThan(e, a) {
-				return e
-			}
-			return a
-		},
-	)
+	return Reduce(s, reducer.MinBy(compare))
 }
 
 // Max returns the maximum element in the stream.
@@ -206,7 +108,7 @@ func MinBy[E any](s Stream[E], compare cmp.Comparer[E]) (min optional.Optional[E
 //	max := stream.Max(stream.Of(3, 1, 2)) // Some(3)
 //	max = stream.Max(stream.Empty[int]()) // None()
 func Max[E constraint.Ordered](s Stream[E]) (max optional.Optional[E]) {
-	return MaxBy(s, cmp.Natural[E]())
+	return Reduce(s, reducer.Max[E]())
 }
 
 // MaxBy returns the maximum element in the stream, or the zero value of the type parameter E if the stream is empty.
@@ -218,13 +120,66 @@ func Max[E constraint.Ordered](s Stream[E]) (max optional.Optional[E]) {
 //	max := stream.MaxBy(stream.Of(3, 1, 2), cmp.Natural[int]()) // Some(3)
 //	max = stream.MaxBy(stream.Empty[int](), cmp.Natural[int]()) // None()
 func MaxBy[E any](s Stream[E], compare cmp.Comparer[E]) (max optional.Optional[E]) {
-	return Reduce(
+	return Reduce(s, reducer.MaxBy(compare))
+}
+
+// Accumulator represents a function that takes an accumulated value of type A and an element of type E,
+// and returns the updated accumulated value of type A.
+// The Accumulator is commonly used in the `Aggregate` function to combine elements of a stream into a single result.
+type Accumulator[A, E any] func(a A, e E) (result A)
+
+// Finisher represents a function that takes an accumulated value of type A and returns the finished result of type F.
+// The Finisher is commonly used in the `Aggregate` function to compute the final result after all elements have been accumulated.
+type Finisher[A, F any] func(a A) (result F)
+
+// Aggregate combines the elements of the stream into a single value using the given identity value, accumulator function and finisher function.
+// The accumulated value is initialized to the identity value.
+// The accumulator function is used to combine each element with the accumulated value.
+// The finisher function is used to compute the final result after all elements have been accumulated.
+// The stream is fully consumed.
+//
+// Example usage:
+//
+//	s := stream.Aggregate(
+//	  stream.Of(1, 2, 3),
+//	  0,                 // Initial value
+//	  func(a, e int) int {
+//	    return a + e     // Accumulate with addition
+//	  },
+//	  func(a int) int {
+//	    return a * 2     // Finish with multiplication by 2
+//	  },
+//	) // (1+2+3) * 2 = 12
+func Aggregate[E, A, F any](s Stream[E], identity A, accumulate Accumulator[A, E], finish Finisher[A, F]) F {
+	a := identity
+	s(func(e E) bool {
+		a = accumulate(a, e)
+		return true
+	})
+	return finish(a)
+}
+
+// Average computes the average of all elements in the stream of any number type E and returns the result as type float64.
+// The result of an empty stream is the zero value of type float64.
+// The stream is fully consumed.
+//
+// Example usage:
+//
+//	n := stream.Average(stream.Of(1, 2, 3)) // 2.0 (float64)
+func Average[E constraint.RealNumber](s Stream[E]) float64 {
+	var count uint64
+	return Aggregate(
 		s,
-		func(a, e E) E {
-			if compare.GreaterThan(e, a) {
-				return e
+		float64(0),
+		func(a float64, e E) float64 {
+			count++
+			return a + float64(e)
+		},
+		func(a float64) float64 {
+			if count == 0 {
+				return 0
 			}
-			return a
+			return a / float64(count)
 		},
 	)
 }
diff --git a/pkg/stream/aggregate_test.go b/pkg/stream/aggregate_test.go
index 793647d..8766ec8 100644
--- a/pkg/stream/aggregate_test.go
+++ b/pkg/stream/aggregate_test.go
@@ -1,6 +1,7 @@
 package stream
 
 import (
+	"github.com/jpfourny/papaya/pkg/cmp"
 	"testing"
 
 	"github.com/jpfourny/papaya/pkg/optional"
@@ -57,7 +58,7 @@ func TestAggregate(t *testing.T) {
 
 func TestSum(t *testing.T) {
 	t.Run("empty", func(t *testing.T) {
-		got := Sum(Empty[int]())
+		got := Sum[int](Empty[int]())
 		want := 0
 		if got != want {
 			t.Errorf("got %#v, want %#v", got, want)
@@ -65,90 +66,89 @@ func TestSum(t *testing.T) {
 	})
 
 	t.Run("non-empty", func(t *testing.T) {
-		got := Sum(Of(1, 2, 3))
+		got := Sum[int](Of(1, 2, 3))
 		want := 6
 		if got != want {
 			t.Errorf("got %#v, want %#v", got, want)
 		}
 	})
-
 }
 
-func TestSumInteger(t *testing.T) {
+func TestSumComplex(t *testing.T) {
 	t.Run("empty", func(t *testing.T) {
-		got := SumInteger(Empty[int]())
-		want := int64(0)
+		got := SumComplex[complex128](Empty[complex128]())
+		want := complex128(0)
 		if got != want {
 			t.Errorf("got %#v, want %#v", got, want)
 		}
 	})
 
 	t.Run("non-empty", func(t *testing.T) {
-		got := SumInteger(Of(1, 2, 3))
-		want := int64(6)
+		got := SumComplex[complex128](Of(complex(1, 2), complex(3, 4)))
+		want := complex(4, 6)
 		if got != want {
 			t.Errorf("got %#v, want %#v", got, want)
 		}
 	})
 }
 
-func TestSumUnsignedInteger(t *testing.T) {
+func TestAverage(t *testing.T) {
 	t.Run("empty", func(t *testing.T) {
-		got := SumUnsignedInteger(Empty[uint]())
-		want := uint64(0)
+		got := Average(Empty[int]())
+		want := 0.0
 		if got != want {
 			t.Errorf("got %#v, want %#v", got, want)
 		}
 	})
 
 	t.Run("non-empty", func(t *testing.T) {
-		got := SumUnsignedInteger(Of[uint](1, 2, 3))
-		want := uint64(6)
+		got := Average(Of(1, 2, 3))
+		want := 2.0
 		if got != want {
 			t.Errorf("got %#v, want %#v", got, want)
 		}
 	})
 }
 
-func TestSumFloat(t *testing.T) {
+func TestMin(t *testing.T) {
 	t.Run("empty", func(t *testing.T) {
-		got := SumFloat(Empty[float64]())
-		want := 0.0
+		got := Min(Empty[int]())
+		want := optional.Empty[int]()
 		if got != want {
-			t.Errorf("got %#v, want %#v", got, want)
+			t.Errorf("got %v, want %v", got, want)
 		}
 	})
 
 	t.Run("non-empty", func(t *testing.T) {
-		got := SumFloat(Of(1.0, 2.0, 3.0))
-		want := 6.0
+		got := Min(Of(3, 1, 2))
+		want := optional.Of(1)
 		if got != want {
-			t.Errorf("got %#v, want %#v", got, want)
+			t.Errorf("got %v, want %v", got, want)
 		}
 	})
 }
 
-func TestAverage(t *testing.T) {
+func TestMinBy(t *testing.T) {
 	t.Run("empty", func(t *testing.T) {
-		got := Average(Empty[int]())
-		want := 0.0
+		got := MinBy(Empty[int](), cmp.Natural[int]())
+		want := optional.Empty[int]()
 		if got != want {
-			t.Errorf("got %#v, want %#v", got, want)
+			t.Errorf("got %v, want %v", got, want)
 		}
 	})
 
 	t.Run("non-empty", func(t *testing.T) {
-		got := Average(Of(1, 2, 3))
-		want := 2.0
+		got := MinBy(Of(3, 1, 2), cmp.Natural[int]())
+		want := optional.Of(1)
 		if got != want {
-			t.Errorf("got %#v, want %#v", got, want)
+			t.Errorf("got %v, want %v", got, want)
 		}
 	})
 }
 
-func TestMin(t *testing.T) {
+func TestMax(t *testing.T) {
 	t.Run("empty", func(t *testing.T) {
-		got := Min(Empty[int]())
+		got := Max(Empty[int]())
 		want := optional.Empty[int]()
 		if got != want {
 			t.Errorf("got %v, want %v", got, want)
@@ -156,17 +156,17 @@ func TestMin(t *testing.T) {
 	})
 
 	t.Run("non-empty", func(t *testing.T) {
-		got := Min(Of(3, 1, 2))
-		want := optional.Of(1)
+		got := Max(Of(1, 3, 2))
+		want := optional.Of(3)
 		if got != want {
 			t.Errorf("got %v, want %v", got, want)
 		}
 	})
 }
 
-func TestMax(t *testing.T) {
+func TestMaxBy(t *testing.T) {
 	t.Run("empty", func(t *testing.T) {
-		got := Max(Empty[int]())
+		got := MaxBy(Empty[int](), cmp.Natural[int]())
 		want := optional.Empty[int]()
 		if got != want {
 			t.Errorf("got %v, want %v", got, want)
@@ -174,7 +174,7 @@ func TestMax(t *testing.T) {
 	})
 
 	t.Run("non-empty", func(t *testing.T) {
-		got := Max(Of(1, 3, 2))
+		got := MaxBy(Of(1, 3, 2), cmp.Natural[int]())
 		want := optional.Of(3)
 		if got != want {
 			t.Errorf("got %v, want %v", got, want)
diff --git a/pkg/stream/combine.go b/pkg/stream/combine.go
index 905a9d6..9a90633 100644
--- a/pkg/stream/combine.go
+++ b/pkg/stream/combine.go
@@ -2,10 +2,10 @@ package stream
 
 import (
 	"github.com/jpfourny/papaya/pkg/constraint"
-	"github.com/jpfourny/papaya/pkg/mapper"
 	"github.com/jpfourny/papaya/pkg/optional"
 	"github.com/jpfourny/papaya/pkg/pair"
-	"github.com/jpfourny/papaya/pkg/pred"
+	"github.com/jpfourny/papaya/pkg/stream/mapper"
+	"github.com/jpfourny/papaya/pkg/stream/pred"
 )
 
 // Combiner represents a function that combines two elements of type E1 and E2 into an element of type F.
diff --git a/pkg/stream/iterate_test.go b/pkg/stream/iterate_test.go
index 2ff1130..ea0cc12 100644
--- a/pkg/stream/iterate_test.go
+++ b/pkg/stream/iterate_test.go
@@ -1,12 +1,12 @@
 package stream
 
 import (
+	"github.com/jpfourny/papaya/pkg/stream/mapper"
+	"github.com/jpfourny/papaya/pkg/stream/pred"
 	"testing"
 
 	"github.com/jpfourny/papaya/internal/assert"
-	"github.com/jpfourny/papaya/pkg/mapper"
 	"github.com/jpfourny/papaya/pkg/optional"
-	"github.com/jpfourny/papaya/pkg/pred"
 )
 
 func TestIterate(t *testing.T) {
diff --git a/pkg/mapper/condition.go b/pkg/stream/mapper/condition.go
similarity index 100%
rename from pkg/mapper/condition.go
rename to pkg/stream/mapper/condition.go
diff --git a/pkg/mapper/condition_test.go b/pkg/stream/mapper/condition_test.go
similarity index 97%
rename from pkg/mapper/condition_test.go
rename to pkg/stream/mapper/condition_test.go
index 280af19..3c309f2 100644
--- a/pkg/mapper/condition_test.go
+++ b/pkg/stream/mapper/condition_test.go
@@ -1,10 +1,10 @@
 package mapper
 
 import (
+	"github.com/jpfourny/papaya/pkg/stream/pred"
 	"testing"
 
 	"github.com/jpfourny/papaya/pkg/optional"
-	"github.com/jpfourny/papaya/pkg/pred"
 )
 
 func TestIf(t *testing.T) {
diff --git a/pkg/mapper/convert.go b/pkg/stream/mapper/convert.go
similarity index 100%
rename from pkg/mapper/convert.go
rename to pkg/stream/mapper/convert.go
diff --git a/pkg/mapper/convert_test.go b/pkg/stream/mapper/convert_test.go
similarity index 100%
rename from pkg/mapper/convert_test.go
rename to pkg/stream/mapper/convert_test.go
diff --git a/pkg/mapper/doc.go b/pkg/stream/mapper/doc.go
similarity index 100%
rename from pkg/mapper/doc.go
rename to pkg/stream/mapper/doc.go
diff --git a/pkg/mapper/format.go b/pkg/stream/mapper/format.go
similarity index 100%
rename from pkg/mapper/format.go
rename to pkg/stream/mapper/format.go
diff --git a/pkg/mapper/format_test.go b/pkg/stream/mapper/format_test.go
similarity index 100%
rename from pkg/mapper/format_test.go
rename to pkg/stream/mapper/format_test.go
diff --git a/pkg/mapper/misc.go b/pkg/stream/mapper/misc.go
similarity index 100%
rename from pkg/mapper/misc.go
rename to pkg/stream/mapper/misc.go
diff --git a/pkg/mapper/misc_test.go b/pkg/stream/mapper/misc_test.go
similarity index 100%
rename from pkg/mapper/misc_test.go
rename to pkg/stream/mapper/misc_test.go
diff --git a/pkg/mapper/parse.go b/pkg/stream/mapper/parse.go
similarity index 100%
rename from pkg/mapper/parse.go
rename to pkg/stream/mapper/parse.go
diff --git a/pkg/mapper/parse_test.go b/pkg/stream/mapper/parse_test.go
similarity index 100%
rename from pkg/mapper/parse_test.go
rename to pkg/stream/mapper/parse_test.go
diff --git a/pkg/mapper/pointer.go b/pkg/stream/mapper/pointer.go
similarity index 100%
rename from pkg/mapper/pointer.go
rename to pkg/stream/mapper/pointer.go
diff --git a/pkg/mapper/pointer_test.go b/pkg/stream/mapper/pointer_test.go
similarity index 100%
rename from pkg/mapper/pointer_test.go
rename to pkg/stream/mapper/pointer_test.go
diff --git a/pkg/mapper/slice.go b/pkg/stream/mapper/slice.go
similarity index 100%
rename from pkg/mapper/slice.go
rename to pkg/stream/mapper/slice.go
diff --git a/pkg/mapper/slice_test.go b/pkg/stream/mapper/slice_test.go
similarity index 100%
rename from pkg/mapper/slice_test.go
rename to pkg/stream/mapper/slice_test.go
diff --git a/pkg/stream/match.go b/pkg/stream/match.go
index eac40c7..5aecd0d 100644
--- a/pkg/stream/match.go
+++ b/pkg/stream/match.go
@@ -2,7 +2,7 @@ package stream
 
 import (
 	"github.com/jpfourny/papaya/pkg/cmp"
-	"github.com/jpfourny/papaya/pkg/pred"
+	pred2 "github.com/jpfourny/papaya/pkg/stream/pred"
 )
 
 // AnyMatch returns true if any element in the stream matches the given Predicate.
@@ -64,7 +64,7 @@ func NoneMatch[E any](s Stream[E], p Predicate[E]) bool {
 //	out := stream.Contains(stream.Of(1, 2, 3), 2) // true
 //	out = stream.Contains(stream.Of(1, 2, 3), 4) // false
 func Contains[E comparable](s Stream[E], e E) bool {
-	return AnyMatch(s, pred.Equal(e))
+	return AnyMatch(s, pred2.Equal(e))
 }
 
 // ContainsBy returns true if the stream contains the given element; false otherwise.
@@ -75,7 +75,7 @@ func Contains[E comparable](s Stream[E], e E) bool {
 //	out := stream.ContainsBy(stream.Of(1, 2, 3), 2, cmp.Natural[int]()) // true
 //	out = stream.ContainsBy(stream.Of(1, 2, 3), 4, cmp.Natural[int]()) // false
 func ContainsBy[E any](s Stream[E], compare cmp.Comparer[E], e E) bool {
-	return AnyMatch(s, pred.EqualBy(e, compare))
+	return AnyMatch(s, pred2.EqualBy(e, compare))
 }
 
 // ContainsAny returns true if the stream contains any of the given elements; false otherwise.
@@ -86,7 +86,7 @@ func ContainsBy[E any](s Stream[E], compare cmp.Comparer[E], e E) bool {
 //	out := stream.ContainsAny(stream.Of(1, 2, 3), 2, 4) // true
 //	out = stream.ContainsAny(stream.Of(1, 2, 3), 4, 5) // false
 func ContainsAny[E comparable](s Stream[E], es ...E) bool {
-	return AnyMatch(s, pred.In(es...))
+	return AnyMatch(s, pred2.In(es...))
 }
 
 // ContainsAnyBy returns true if the stream contains any of the given elements; false otherwise.
@@ -97,7 +97,7 @@ func ContainsAny[E comparable](s Stream[E], es ...E) bool {
 //	out := stream.ContainsAnyBy(stream.Of(1, 2, 3), cmp.Natural[int](), 2, 4) // true
 //	out = stream.ContainsAnyBy(stream.Of(1, 2, 3), cmp.Natural[int](), 4, 5) // false
 func ContainsAnyBy[E any](s Stream[E], compare cmp.Comparer[E], es ...E) bool {
-	return AnyMatch(s, pred.InBy(compare, es...))
+	return AnyMatch(s, pred2.InBy(compare, es...))
 }
 
 // ContainsAll returns true if the stream contains all the given elements; false otherwise.
@@ -141,7 +141,7 @@ func ContainsAllBy[E any](s Stream[E], compare cmp.Comparer[E], es ...E) bool {
 //	out := stream.ContainsNone(stream.Of(1, 2, 3), 4, 5) // true
 //	out = stream.ContainsNone(stream.Of(1, 2, 3), 2, 4) // false
 func ContainsNone[E comparable](s Stream[E], es ...E) bool {
-	return NoneMatch(s, pred.In(es...))
+	return NoneMatch(s, pred2.In(es...))
 }
 
 // ContainsNoneBy returns true if the stream contains none of the given elements; false otherwise.
@@ -152,5 +152,5 @@ func ContainsNone[E comparable](s Stream[E], es ...E) bool {
 //	out := stream.ContainsNoneBy(stream.Of(1, 2, 3), cmp.Natural[int](), 4, 5) // true
 //	out = stream.ContainsNoneBy(stream.Of(1, 2, 3), cmp.Natural[int](), 2, 4) // false
 func ContainsNoneBy[E any](s Stream[E], compare cmp.Comparer[E], es ...E) bool {
-	return NoneMatch(s, pred.InBy(compare, es...))
+	return NoneMatch(s, pred2.InBy(compare, es...))
 }
diff --git a/pkg/stream/misc.go b/pkg/stream/misc.go
index 74f3347..97bc657 100644
--- a/pkg/stream/misc.go
+++ b/pkg/stream/misc.go
@@ -2,9 +2,8 @@ package stream
 
 import (
 	"github.com/jpfourny/papaya/pkg/optional"
+	"github.com/jpfourny/papaya/pkg/stream/mapper"
 	"strings"
-
-	"github.com/jpfourny/papaya/pkg/mapper"
 )
 
 // ForEach invokes the given consumer for each element in the stream.
diff --git a/pkg/pred/compare.go b/pkg/stream/pred/compare.go
similarity index 100%
rename from pkg/pred/compare.go
rename to pkg/stream/pred/compare.go
diff --git a/pkg/pred/compare_test.go b/pkg/stream/pred/compare_test.go
similarity index 100%
rename from pkg/pred/compare_test.go
rename to pkg/stream/pred/compare_test.go
diff --git a/pkg/pred/doc.go b/pkg/stream/pred/doc.go
similarity index 100%
rename from pkg/pred/doc.go
rename to pkg/stream/pred/doc.go
diff --git a/pkg/pred/in.go b/pkg/stream/pred/in.go
similarity index 100%
rename from pkg/pred/in.go
rename to pkg/stream/pred/in.go
diff --git a/pkg/pred/in_test.go b/pkg/stream/pred/in_test.go
similarity index 100%
rename from pkg/pred/in_test.go
rename to pkg/stream/pred/in_test.go
diff --git a/pkg/pred/logic.go b/pkg/stream/pred/logic.go
similarity index 100%
rename from pkg/pred/logic.go
rename to pkg/stream/pred/logic.go
diff --git a/pkg/pred/logic_test.go b/pkg/stream/pred/logic_test.go
similarity index 100%
rename from pkg/pred/logic_test.go
rename to pkg/stream/pred/logic_test.go
diff --git a/pkg/pred/string.go b/pkg/stream/pred/string.go
similarity index 100%
rename from pkg/pred/string.go
rename to pkg/stream/pred/string.go
diff --git a/pkg/pred/string_test.go b/pkg/stream/pred/string_test.go
similarity index 100%
rename from pkg/pred/string_test.go
rename to pkg/stream/pred/string_test.go
diff --git a/pkg/pred/zero.go b/pkg/stream/pred/zero.go
similarity index 100%
rename from pkg/pred/zero.go
rename to pkg/stream/pred/zero.go
diff --git a/pkg/pred/zero_test.go b/pkg/stream/pred/zero_test.go
similarity index 100%
rename from pkg/pred/zero_test.go
rename to pkg/stream/pred/zero_test.go
diff --git a/pkg/stream/reducer/reducer.go b/pkg/stream/reducer/reducer.go
new file mode 100644
index 0000000..3b2fcb6
--- /dev/null
+++ b/pkg/stream/reducer/reducer.go
@@ -0,0 +1,53 @@
+package reducer
+
+import (
+	"github.com/jpfourny/papaya/pkg/cmp"
+	"github.com/jpfourny/papaya/pkg/constraint"
+)
+
+// Sum returns a function that computes the sum of two given values.
+func Sum[E constraint.Numeric]() func(a, b E) E {
+	return func(a, b E) E {
+		return a + b
+	}
+}
+
+// Min returns a function that returns the minimum of two given values.
+func Min[E constraint.Ordered]() func(a, b E) E {
+	return func(a, b E) E {
+		if a <= b {
+			return a
+		}
+		return b
+	}
+}
+
+// MinBy returns a function that returns the minimum of two given values using the provided cmp.Comparer.
+func MinBy[E any](compare cmp.Comparer[E]) func(a, b E) E {
+	return func(a, b E) E {
+		if compare.LessThanOrEqual(a, b) {
+			return a
+		}
+		return b
+	}
+}
+
+// Max returns a function that returns the maximum of two given values.
+func Max[E constraint.Ordered]() func(a, b E) E {
+	return func(a, b E) E {
+		if a >= b {
+			return a
+		}
+		return b
+	}
+}
+
+// MaxBy returns a function that returns the maximum of two given values using the provided cmp.Comparer.
+func MaxBy[E any](compare cmp.Comparer[E]) func(a, b E) E {
+	return func(a, b E) E {
+		if compare.GreaterThanOrEqual(a, b) {
+			return a
+		}
+		return b
+	}
+}
diff --git a/pkg/stream/transform_test.go b/pkg/stream/transform_test.go
index 2b1b6ca..507f0cf 100644
--- a/pkg/stream/transform_test.go
+++ b/pkg/stream/transform_test.go
@@ -2,21 +2,21 @@ package stream
 
 import (
 	"fmt"
+	mapper2 "github.com/jpfourny/papaya/pkg/stream/mapper"
 	"testing"
 
 	"github.com/jpfourny/papaya/internal/assert"
-	"github.com/jpfourny/papaya/pkg/mapper"
 )
 
 func TestMap(t *testing.T) {
-	s := Map(Of(1, 2, 3), mapper.Sprintf[int]("%d"))
+	s := Map(Of(1, 2, 3), mapper2.Sprintf[int]("%d"))
 	got := CollectSlice(s)
 	want := []string{"1", "2", "3"}
 	assert.ElementsMatch(t, got, want)
 }
 
 func TestMapOrDiscard(t *testing.T) {
-	s := MapOrDiscard(Of("1", "2", "3", "foo"), mapper.TryParseInt[string, int](10, 64))
+	s := MapOrDiscard(Of("1", "2", "3", "foo"), mapper2.TryParseInt[string, int](10, 64))
 	got := CollectSlice(s)
 	want := []int{1, 2, 3}
 	assert.ElementsMatch(t, got, want)