log failures which can be in spawned operations

Author: rssh

shared/src/main/scala/gopher/Gopher.scala

@@ -2,7 +2,7 @@ package gopher
 
 import cps._
 import scala.concurrent.duration.Duration
-import scala.util.Try
+import scala.util._
 
 import java.util.logging.{Level => LogLevel}
 
@@ -33,6 +33,14 @@ trait Gopher[F[_]:CpsSchedulingMonad]:
   protected[gopher] def logImpossible(ex: Throwable): Unit =
     log(LogLevel.WARNING, "impossible", ex)
 
+  protected[gopher] def spawnAndLogFail[T](op: =>F[T]): F[Unit] =
+     asyncMonad.mapTry(asyncMonad.spawn(op)){
+           case Success(_) => ()
+           case Failure(ex) =>
+             log(LogLevel.WARNING, "exception in spawned process", ex)
+             ()
+     }
+
 
 def makeChannel[A](bufSize:Int = 0, 
                   autoClose: Boolean = false)(using g:Gopher[?]):Channel[g.Monad,A,A] =
@@ -46,7 +54,7 @@ def select(using g:Gopher[?]):Select[g.Monad] =
 
 def futureInput[F[_],A](f: F[A])(using g: Gopher[F]): ReadChannel[F,A] =
       val ch = g.makeOnceChannel[Try[A]]()
-      g.asyncMonad.spawn{
+      g.spawnAndLogFail{
             g.asyncMonad.flatMapTry(f)(r => ch.awrite(r))
       } 
       ch.map(_.get)

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

@@ -149,7 +149,7 @@ trait ReadChannel[F[_], A]:
    def zip[B](x: ReadChannel[F,B]): ReadChannel[F,(A,B)] = 
       given CpsSchedulingMonad[F] = asyncMonad
       val retval = gopherApi.makeChannel[(A,B)]()
-      asyncMonad.spawn(async[F]{
+      gopherApi.spawnAndLogFail(async[F]{
          var done = false
          while(!done) {
             this.optRead() match
@@ -213,7 +213,7 @@ object ReadChannel:
    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]()
-      asyncMonad.spawn(async{
+      summon[Gopher[F]].spawnAndLogFail(async{
          val it = c.iterator
          while(it.hasNext) {
             val a = it.next()

shared/src/main/scala/gopher/Select.scala

@@ -60,7 +60,7 @@ class Select[F[_]](api: Gopher[F]):
   def mapAsync[A](step: SelectGroup[F,A] => F[A]): ReadChannel[F,A] =
     val r = makeChannel[A]()(using api)
     given CpsSchedulingMonad[F] = api.asyncMonad
-    api.asyncMonad.spawn{
+    api.spawnAndLogFail{
       async{
         var done = false
         while(!done) 

shared/src/main/scala/gopher/SelectGroup.scala

@@ -169,7 +169,7 @@ class SelectGroup[F[_], S](api: Gopher[F])  extends SelectListeners[F,S,S]:
              if waitState.compareAndSet(0,1) then
                  Some(v => {
                     timeoutScheduled.foreach(_.cancel())
-                    m.spawn(
+                    api.spawnAndLogFail(
                       m.mapTry(action(v))(x => call(x))
                     )
                  })
@@ -189,7 +189,7 @@ class SelectGroup[F[_], S](api: Gopher[F])  extends SelectListeners[F,S,S]:
             if waitState.compareAndSet(0,1) then
               Some((element, (v:Try[Unit]) => {
                         timeoutScheduled.foreach(_.cancel())
-                        m.spawn(
+                        api.spawnAndLogFail(
                           m.mapTry(action(v))(x=>call(x))
                         )}
                   ))
@@ -204,7 +204,7 @@ class SelectGroup[F[_], S](api: Gopher[F])  extends SelectListeners[F,S,S]:
       def capture(): Option[Try[FiniteDuration] => Unit] =
         if (waitState.compareAndSet(0,1)) then
           Some((v:Try[FiniteDuration]) =>
-            m.spawn(m.mapTry(action(v))(x => call(x)))
+            api.spawnAndLogFail(m.mapTry(action(v))(x => call(x)))
           )
         else
           None

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

@@ -40,7 +40,7 @@ class ChFlatMappedReadChannel[F[_], A, B](prev: ReadChannel[F,A], f: A=>ReadChan
       bChannel.close()
     }
 
-  gopherApi.asyncMonad.spawn(run())
+  gopherApi.spawnAndLogFail(run())
 
 
 

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

@@ -45,7 +45,7 @@ class ChFlatMappedTryReadChannel[F[_], A, B](prev: ReadChannel[F,Try[A]], f: Try
     }
   } 
 
-  gopherApi.asyncMonad.spawn(run())
+  gopherApi.spawnAndLogFail(run())
 
 
 }

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

@@ -32,6 +32,6 @@ class DuppedInput[F[_],A](origin:ReadChannel[F,A], bufSize: Int=1)(using api:Gop
       val f2 = sink2.write(a)
       true
     }}.runAsync()
-    api.asyncMonad.spawn(runner)
+    api.spawnAndLogFail(runner)
 
 }

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

@@ -62,7 +62,7 @@ class FilteredAsyncReadChannel[F[_],A](internal: ReadChannel[F,A], p: A=>F[Boole
 
     def wrappedFun(fun: (Try[A] => Unit) ): (Try[A] => Unit) = {
       case Success(a) =>
-        gopherApi.asyncMonad.spawn(
+        gopherApi.spawnAndLogFail(
           gopherApi.asyncMonad.mapTry(p(a)){
             case Success(v) =>
               if (v) {

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

@@ -51,7 +51,7 @@ 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{
-          asyncMonad.spawn(
+          gopherApi.spawnAndLogFail(
             asyncMonad.mapTry(f(a))(fun)
           )
         }catch{

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

@@ -37,7 +37,7 @@ given ReadTryChannelCpsMonad[F[_]](using Gopher[F]): CpsAsyncMonad[ [A] =>> Read
   def error[A](e: Throwable): ReadChannel[F,Try[A]] = 
     val r = makeChannel[Try[A]]()
     given fm: CpsSchedulingMonad[F] = summon[Gopher[F]].asyncMonad
-    fm.spawn{ async[F] {
+    summon[Gopher[F]].spawnAndLogFail{ async[F] {
       r.write(Failure(e))
       r.close()
     } }
@@ -48,7 +48,7 @@ given ReadTryChannelCpsMonad[F[_]](using Gopher[F]): CpsAsyncMonad[ [A] =>> Read
     val r = makeOnceChannel[Try[A]]()
     given fm: CpsSchedulingMonad[F] = summon[Gopher[F]].asyncMonad
     val fv = fm.adoptCallbackStyle(source)
-    fm.spawn{
+    summon[Gopher[F]].spawnAndLogFail{
         fm.flatMapTry( fv ){ tryV =>
            r.awrite(tryV)
         }