Add Last

README.md

@@ -326,6 +326,7 @@ After constructing the desired pipeline, the stream needs to be executed.
 - `stream.AppendToSlice[A any](stm Stream[A], start []A) io.IO[[]A]`
 - `stream.ToSlice[A any](stm Stream[A]) io.IO[[]A]`
 - `stream.Head[A any](stm Stream[A]) io.IO[A]` - returns the first element if it exists. Otherwise - an error.
+- `stream.Last[A any](stm Stream[A]) io.IO[A]` - Last keeps track of the current element of the stream and returns it when the stream completes.
 - `stream.Collect[A any](stm Stream[A], collector func (A) error) io.IO[fun.Unit]` - collects all element from the stream and for each element invokes the provided function.
 - `stream.ForEach[A any](stm Stream[A], collector func (A)) io.IO[fun.Unit]` - invokes a simple function for each element of the stream.
 - `stream.Partition[A any, C any, D any](stm Stream[A], predicate func(A) bool, trueHandler func(Stream[A]) io.IO[C], falseHandler func(Stream[A]) io.IO[D]) io.IO[fun.Pair[C, D]]` - Partition divides the stream into two that are handled independently.

stream/execution.go

@@ -6,6 +6,7 @@ import (
 	"github.com/primetalk/goio/either"
 	"github.com/primetalk/goio/fun"
 	"github.com/primetalk/goio/io"
+	"github.com/primetalk/goio/option"
 )
 
 // Collector reads the stream and produces some value.
@@ -87,6 +88,26 @@ func Head[A any](stm Stream[A]) io.IO[A] {
 	})
 }
 
+// Last keeps track of the current element of the stream and returns it when the stream completes.
+func Last[A any](stm Stream[A]) io.IO[A] {
+	return Head(
+		StateFlatMapWithFinish(
+			stm, option.None[A](),
+			func(a A, st option.Option[A]) io.IO[fun.Pair[option.Option[A], Stream[A]]] {
+				return io.Lift(fun.NewPair(option.Some(a), Empty[A]()))
+			},
+			func(st option.Option[A]) Stream[A] {
+				return option.Fold(st,
+					Lift[A],
+					func() Stream[A] {
+						return Fail[A](fmt.Errorf("last of an empty stream"))
+					},
+				)
+			},
+		),
+	)
+}
+
 // Partition divides the stream into two that are handled independently.
 func Partition[A any, C any, D any](stm Stream[A],
 	predicate func(A) bool,

stream/stream_test.go

@@ -34,6 +34,10 @@ func TestGenerate(t *testing.T) {
 	res, err = io.UnsafeRunSync(stream.Head(powers2_10))
 	assert.NoError(t, err)
 	assert.Equal(t, 1024, res)
+
+	res, err = io.UnsafeRunSync(stream.Last(stream.Take(powers2, 10)))
+	assert.NoError(t, err)
+	assert.Equal(t, 1024, res)
 }
 
 func TestDrainAll(t *testing.T) {