adopted to dotty-cps-async 0.9.1 (async fir future starts with spawn)

Author: rssh

build.sbt

@@ -2,15 +2,15 @@
 val dottyVersion = "3.0.1"
 //val dottyVersion = dottyLatestNightlyBuild.get
 
-ThisBuild/version := "2.0.5-SNAPSHOT"
+ThisBuild/version := "2.0.5"
 ThisBuild/versionScheme := Some("semver-spec")
 
 val sharedSettings = Seq(
     organization := "com.github.rssh",
     scalaVersion := dottyVersion,
     name := "scala-gopher",
     resolvers += "Local Ivy Repository" at "file://"+Path.userHome.absolutePath+"/.ivy2/local",
-    libraryDependencies += "com.github.rssh" %%% "dotty-cps-async" % "0.9.0",
+    libraryDependencies += "com.github.rssh" %%% "dotty-cps-async" % "0.9.1",
     libraryDependencies += "org.scalameta" %%% "munit" % "0.7.26" % Test,
     testFrameworks += new TestFramework("munit.Framework")
 )

jvm/src/main/scala/gopher/impl/PromiseChannel.scala

@@ -34,8 +34,10 @@ import scala.util.Failure
             case Some((a,f)) =>
               val ar: AnyRef = a.asInstanceOf[AnyRef] //
               if (ref.compareAndSet(null,ar) && !closed.get() ) then
-                closed.lazySet(true)
-                taskExecutor.execute(()=> f(Success(())))
+                closed.set(true)
+                taskExecutor.execute{ ()=> 
+                  f(Success(()))
+                }
                 writer.markUsed()
                 step()
               else 
@@ -48,21 +50,24 @@ import scala.util.Failure
 
 
     def addDoneReader(reader: Reader[Unit]): Unit =
-      if (!closed.get()) then
+      if (!closed.get() || !readed.get) then
         doneReaders.add(reader)
+        if (closed.get()) then
+          step()
       else 
         var done = false
         while(!done & !reader.isExpired) {
-          reader.capture() match
-            case Some(f) => 
-              reader.markUsed()
-              taskExecutor.execute(()=>f(Success(())))
-              done = true
-            case None =>
-              if (!reader.isExpired)
-                Thread.onSpinWait()
+            reader.capture() match
+              case Some(f) => 
+                reader.markUsed()
+                taskExecutor.execute(()=>f(Success(())))
+                done = true
+              case None =>
+                if (!reader.isExpired)
+                  Thread.onSpinWait()
         }
 
+          
 
 
     def close(): Unit =

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

@@ -214,6 +214,10 @@ object ReadChannel:
       retval.close()
       retval
 
+   /**
+    *@param c - iteratable to read from.
+    *@return channel, which will emit all elements from 'c' and then close.
+    **/   
    def fromIterable[F[_],A](c: IterableOnce[A])(using Gopher[F]): ReadChannel[F,A] =
       given asyncMonad: CpsSchedulingMonad[F] = summon[Gopher[F]].asyncMonad
       val retval = makeChannel[A]()
@@ -227,6 +231,28 @@ object ReadChannel:
       })
       retval
 
+   /**
+    *@return one copy of `a` and close.
+   **/   
+   def once[F[_],A](a: A)(using Gopher[F]): ReadChannel[F,A] =
+      fromIterable(List(a))   
+
+   /**
+    *@param a - value to produce
+    *@return channel which emit value of a in loop and never close
+    **/  
+   def always[F[_],A](a: A)(using Gopher[F]): ReadChannel[F,A] =
+      given asyncMonad: CpsSchedulingMonad[F] = summon[Gopher[F]].asyncMonad
+      val retval = makeChannel[A]()
+      summon[Gopher[F]].spawnAndLogFail(
+         async{
+            while(true) {
+               retval.write(a)
+            }
+         }
+      )
+      retval
+
 
 
    def fromFuture[F[_],A](f: F[A])(using Gopher[F]): ReadChannel[F,A] =

shared/src/main/scala/gopher/monads/ReadChannelCpsMonad.scala

@@ -23,6 +23,6 @@ given futureToReadChannel[F[_]](using Gopher[F]): CpsMonadConversion[F, [A] =>>
    def apply[T](ft: F[T]): ReadChannel[F,T] = futureInput(ft)
 
 
-   
 
+   
 

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

@@ -419,6 +419,7 @@ class MacroSelectSuite extends FunSuite
         }
         val f1 = ch.awrite(1)
         async {
+            await(f1)
             val r = await(sf)
             assert(r==1)
         }

shared/src/test/scala/gopher/monads/Queens.scala

@@ -22,7 +22,25 @@ class QueensSuite extends FunSuite {
      busyLRDiagonals:Set[Int],
      busyRLDiagonals:Set[Int],
      queens: Vector[(Int,Int)]
-  );
+  )  {
+
+    def isBusy(i:Int, j:Int): Boolean = 
+      busyRows.contains(i) ||
+      busyColumns.contains(j) ||
+      busyLRDiagonals.contains(i-j) ||
+      busyRLDiagonals.contains(i+j)
+      
+
+    def put(i:Int, j:Int): State =
+      copy( busyRows = busyRows + i,
+            busyColumns = busyColumns + j,
+            busyLRDiagonals = busyLRDiagonals + (i-j),
+            busyRLDiagonals = busyRLDiagonals + (i+j),
+            queens = queens :+ (i,j)
+          )
+
+
+  }
 
   val N = 8
 
@@ -31,29 +49,16 @@ class QueensSuite extends FunSuite {
     async[Future] {
       val i = state.queens.length
       if i < N then 
-        for{    
-          j <- 0 until N  if !state.busyColumns.contains(j) &&
-                         !state.busyLRDiagonals.contains(i-j) &&
-                         !state.busyRLDiagonals.contains(i+j)  
-          } {
-            val newPos = (i,j)
-            val nState = state.copy( busyRows = state.busyRows + i,
-                                 busyColumns = state.busyColumns + j,
-                                 busyLRDiagonals = state.busyLRDiagonals + (i-j),
-                                 busyRLDiagonals = state.busyRLDiagonals + (i+j),
-                                 queens = state.queens :+ newPos  )
-            ch.write(nState)
-        }
+        for{ j <- 0 until N  if !state.isBusy(i,j) } 
+          ch.write(state.put(i,j))
       ch.close()
     }
     ch
 
   def solutions(state: State): ReadChannel[Future,State] =
     async[[X] =>> ReadChannel[Future,X]] {
       if(state.queens.size < N) then
-        //println("state:"+state.queens)
         val nextState = await(putQueen(state))
-        //println("next-state:"+state.queens)
         await(solutions(nextState))
       else
         state