Resource management in streams

[Primetalk]

Currently streams can only use resources provided at the top level via ordinary resource.Use.
Sometimes we need to pass streams around and we wish that resources are correctly released when stream is either finished or thrown away.

[Primetalk]

This requires adding a finalizer to all stream steps. Then in every stream executor, these finalizers should be respected and appended to io.

[Primetalk]

Signature might look like

stream.Finally[A any](stm stream.Stream[A], func () io.IOUnit) stream.Stream[A]

[Primetalk]

There are two FStreams - one that is just a pair of stream and IO to finalize, and another one - IO of such pair. The difference is that IO represents a stream that has not yet been started.

[TelpeNight]

Hi. I see that repo is abandoned. But it is interesting to follow)

So, what about:

func ResourceUseInStream[Res any, A any](res resource.Resource[Res], f func(Res) stream.Stream[A]) stream.Stream[A] {
	return stream.Stream[A](io.FlatMap(io.IO[resource.Closable[Res]](res),
		func(cl resource.Closable[Res]) io.IO[stream.StepResult[A]] {
			return io.IO[stream.StepResult[A]](Finally(f(cl.Value), cl.Close))
		},
	))
}

func Finally[A any](str stream.Stream[A], finalizer func() io.IOUnit) stream.Stream[A] {
	return stream.Stream[A](stream.StreamFold(str,
		func() io.IO[stream.StepResult[A]] {
			return io.FlatMap(finalizer(), func(fun.Unit) io.IO[stream.StepResult[A]] {
				return io.Lift(stream.NewStepResultFinished[A]())
			})
		},
		func(a A, tail stream.Stream[A]) io.IO[stream.StepResult[A]] {
			return io.Lift(stream.NewStepResult(a, Finally(tail, finalizer)))
		},
		func(tail stream.Stream[A]) io.IO[stream.StepResult[A]] {
			return io.Lift(stream.NewStepResultEmpty[A](Finally(tail, finalizer)))
		},
		func(err error) io.IO[stream.StepResult[A]] {
			iocl := finalizer()
			ioclSafe := io.Recover(iocl, func(err2 error) io.IO[fun.Unit] {
				log.Printf("double error during resource release: %+v", err2)
				return io.IOUnit1
			})
			return io.FlatMap(ioclSafe, func(fun.Unit) io.IO[stream.StepResult[A]] {
				return io.Fail[stream.StepResult[A]](err)
			})
		},
	))
}

I've just combined the logic of resource.Use with stream.MapEval) Seems to work in proper order.