added ReadChannel.unfold

 fiexe situation, where exception in map is not propagated to reader

Author: rssh

shared/src/main/scala/gopher/ReadChannel.scala

@@ -5,6 +5,7 @@ import gopher.impl._
 import scala.util.Try
 import scala.util.Success
 import scala.util.Failure
+import scala.util.control.NonFatal
 import scala.concurrent.duration.Duration
 
 import java.util.logging.{Level => LogLevel}
@@ -231,7 +232,35 @@ object ReadChannel:
    def fromValues[F[_],A](values: A*)(using Gopher[F]): ReadChannel[F,A] =
       fromIterable(values)
 
-   
+   def unfold[S,F[_],A](s:S)(f:S => Option[(A,S)])(using Gopher[F]): ReadChannel[F,A] =
+      unfoldAsync[S,F,A](s)( state => summon[Gopher[F]].asyncMonad.tryPure(f(state)) )
+      
+   def unfoldAsync[S,F[_],A](s:S)(f:S => F[Option[(A,S)]])(using Gopher[F]): ReadChannel[F,A]=
+      given asyncMonad: CpsSchedulingMonad[F] = summon[Gopher[F]].asyncMonad
+      val retval = makeChannel[Try[A]]()
+      summon[Gopher[F]].spawnAndLogFail(async{
+         var done = false
+         var state = s
+         try
+            while(!done) {
+               await(f(state)) match
+                  case Some((a,next)) =>
+                     retval.write(Success(a))
+                     state = next
+                  case None =>
+                     done = true
+            }
+         catch
+            case NonFatal(ex) =>
+               retval.write(Failure(ex))
+         finally
+            retval.close();
+      })
+      retval.map{
+         case Success(x) => x
+         case Failure(ex) =>
+            throw ex
+      }   
 
 end ReadChannel
 

shared/src/main/scala/gopher/impl/FilteredReadChannel.scala

@@ -68,7 +68,7 @@ class FilteredAsyncReadChannel[F[_],A](internal: ReadChannel[F,A], p: A=>F[Boole
               if (v) {
                 if (markedUsed.get()) {
                   nested.markUsed()
-                }
+                } 
                 fun(Success(a))
               } else {
                 nested.markFree()

shared/src/main/scala/gopher/impl/MappedReadChannel.scala

@@ -11,8 +11,12 @@ class MappedReadChannel[F[_],A, B](internal: ReadChannel[F,A], f: A=> B) extends
 
     def wrappedFun(fun: (Try[B] => Unit) ): (Try[A] => Unit) = {
       case Success(a) =>
-        val b = f(a)
-        fun(Success(b))
+        try
+          val b = f(a)
+          fun(Success(b))
+        catch
+          case NonFatal(ex) =>
+            fun(Failure(ex))
       case Failure(ex) =>
         fun(Failure(ex))
     }
@@ -50,14 +54,14 @@ class MappedAsyncReadChannel[F[_],A, B](internal: ReadChannel[F,A], f: A=> F[B])
 
     def wrappedFun(fun: (Try[B] => Unit) ): (Try[A] => Unit) = {
       case Success(a) =>
-        try{
-          gopherApi.spawnAndLogFail(
-            asyncMonad.mapTry(f(a))(fun)
-          )
-        }catch{
-          case NonFatal(ex) =>
-            fun(Failure(ex))
-        }
+        gopherApi.spawnAndLogFail(
+            try 
+              asyncMonad.mapTry(f(a))(fun)
+            catch
+              case NonFatal(ex) =>
+                fun(Failure(ex))
+                asyncMonad.pure(())
+        )
       case Failure(ex) =>
         fun(Failure(ex))
     }

shared/src/test/scala/gopher/channels/ReadChannelFactoryTest.scala

@@ -0,0 +1,68 @@
+package gopher.channels
+
+import gopher._
+import cps._
+import munit._
+
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.{Channel=>_,_}
+import scala.concurrent.duration._
+
+import cps.monads.FutureAsyncMonad
+
+class ReadChannelFactoryTest extends FunSuite {
+
+    given Gopher[Future] = Gopher[Future]()
+
+
+    test("unfoldAsync produce stream simple") {
+      val  ch = ReadChannel.unfoldAsync(0){
+        (x: Int) =>
+          if (x > 10)  then
+            Future successful None
+          else
+            Future successful Some(x,x+1)
+      }
+
+      ch.atake(20).map{ values =>
+         assert(values(0) == 0)
+         assert(values(1) == 1)
+         assert(values(2) == 2)
+         assert(values.size == 11)
+      }
+
+    }
+
+
+    test("unfoldAsync prodce stream with error") {
+      val  ch = ReadChannel.unfoldAsync(0){
+        (x: Int) =>
+          if (x > 3)  then
+            Future failed new RuntimeException("state is too big")
+          else
+            Future successful Some(x,x+1)
+      }
+
+      async {
+        val r0 = ch.read()
+        assert(r0 == 0)
+        val r1 = ch.read()
+        assert(r1 == 1)
+        val r2 = ch.read()
+        assert(r2 == 2)
+        val r3 = ch.read()
+        assert(r3 == 3)
+        var wasTooBig = false
+        try {
+          val r4 = ch.read()
+        }catch{
+           case e: RuntimeException =>
+            wasTooBig = true
+        }
+        assert(wasTooBig)
+      }
+
+
+    }
+
+}