Some tweaks to code and docs (#1)

README.md

@@ -300,13 +300,11 @@ The general interface of a channel is as follows:
 trait Channel[T]:
   def read()(using Async): T
   def send(x: T)(using Async): Unit
-  def close(): Unit
 ```
 Channels provide
 
  - a `read` method, which might suspend while waiting for a message to arrive,
  - a `send` method, which also might suspend in case this is a sync channel or there is some other mechanism that forces a sender to wait,
- - a `close` method which closes the channel. Trying to send to a closed channel or to read from it results in a `ChannelClosedException` to be thrown.
 
 ### Async Channels
 
@@ -409,12 +407,17 @@ A stream represents a sequence of values that are computed one-by-one in a separ
     case More(elem: T, rest: Stream[T])
     case End extends StreamResult[Nothing]
 ```
-One can see a stream as a static representation of the values that are transmitted over a channel. Indeed, there is an easy way to convert a channel to a stream:
+One can see a stream as a static representation of the values that are transmitted over a channel. This poses the question of termination -- when do we know that a channel receives no further values, so the stream can be terminated
+with an `StreamResult.End` value? The following implementation shows one
+possibility: Here we map a channel of `Try` results to a stream, mapping
+failures with a special =`ChannelClosedException` to `StreamResult.End`.
 ```scala
-  extension [T](c: Channel[T])
+  extension [T](c: Channel[Try[T]])
     def toStream(using Async.Config): Stream[T] = Future:
-      try StreamResult.More(read(), toStream)
-      catch case ex: ChannelClosedException => StreamResult.End
+      c.read() match
+        case Success(x) => StreamResult.More(x, toStream)
+        case Failure(ex: ChannelClosedException) => StreamResult.End
+        case Failure(ex) => throw ex
 ```
 
 ### Coroutines or Fibers

src/main/scala/async/Async.scala

@@ -70,7 +70,8 @@ object Async:
 
   def group[T](body: Async ?=> T)(using async: Async): T =
     val newGroup = CancellationGroup().link()
-    body(using async.withConfig(async.config.copy(group = newGroup)))
+    try body(using async.withConfig(async.config.copy(group = newGroup)))
+    finally newGroup.cancel()
 
   /** A function `T => Boolean` whose lineage is recorded by its implementing
    *  classes. The Listener function accepts values of type `T` and returns

src/main/scala/async/channels.scala

@@ -3,13 +3,20 @@ import scala.collection.mutable, mutable.ListBuffer
 import fiberRuntime.boundary, boundary.Label
 import fiberRuntime.suspend
 import scala.concurrent.ExecutionContext
+import scala.util.{Try, Failure}
 import Async.{Listener, await}
 
 /** A common interface for channels */
 trait Channel[T]:
   def read()(using Async): T
   def send(x: T)(using Async): Unit
-  def close(): Unit
+  protected def shutDown(finalValue: T): Unit
+
+object Channel:
+
+  extension [T](c: Channel[Try[T]])
+    def close(): Unit =
+      c.shutDown(Failure(ChannelClosedException()))
 
 class ChannelClosedException extends Exception
 
@@ -57,8 +64,9 @@ class AsyncChannel[T] extends Async.OriginalSource[T], Channel[T]:
   def dropListener(k: Listener[T]): Unit = synchronized:
     waiting -= k
 
-  def close() =
+  protected def shutDown(finalValue: T) =
     isClosed = true
+    waiting.foreach(_(finalValue))
 
 end AsyncChannel
 
@@ -117,8 +125,9 @@ object SyncChannel:
       def dropListener(k: Listener[Listener[T]]): Unit = synchronized:
         pendingSends -= k
 
-    def close() =
+    protected def shutDown(finalValue: T) =
       isClosed = true
+      pendingReads.foreach(_(finalValue))
 
 end SyncChannel
 

src/main/scala/async/streams.scala

@@ -9,7 +9,10 @@ enum StreamResult[+T]:
 
 import StreamResult.*
 
-extension [T](c: Channel[T])
+extension [T](c: Channel[Try[T]])
   def toStream(using Async.Config): Stream[T] = Future:
-    try StreamResult.More(c.read(), toStream)
-    catch case ex: ChannelClosedException => StreamResult.End
+    c.read() match
+      case Success(x) => StreamResult.More(x, toStream)
+      case Failure(ex: ChannelClosedException) => StreamResult.End
+      case Failure(ex) => throw ex
+