Add Source.repeatEval, improve exception handling in sources (#96)

Author: adamw

core/src/main/scala/ox/channels/SourceCompanionOps.scala

@@ -125,12 +125,31 @@ trait SourceCompanionOps:
     }
     c
 
-  def repeat[T](element: T = ())(using Ox, StageCapacity): Source[T] =
+  /** Creates a channel, to which the given `element` is sent repeatedly.
+    *
+    * @param element
+    *   The element to send
+    * @return
+    *   A source to which the given element is sent repeatedly.
+    */
+  def repeat[T](element: T = ())(using Ox, StageCapacity): Source[T] = repeatEval(element)
+
+  /** Creates a channel, to which the result of evaluating `f` is sent repeatedly. As the parameter is passed by-name, the evaluation is
+    * deferred until the element is sent, and happens multiple times.
+    *
+    * @param f
+    *   The code block, computing the element to send
+    * @return
+    *   A source to which the result of evaluating `f` is sent repeatedly.
+    */
+  def repeatEval[T](f: => T)(using Ox, StageCapacity): Source[T] =
     val c = StageCapacity.newChannel[T]
     fork {
-      forever {
-        c.sendSafe(element)
-      }
+      try
+        forever {
+          c.sendSafe(f)
+        }
+      catch case t: Throwable => c.errorSafe(t)
     }
     c
 

core/src/test/scala/ox/channels/SourceOpsRepeatEvalTest.scala

@@ -0,0 +1,19 @@
+package ox.channels
+
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import ox.supervised
+import ox.channels.Source
+
+class SourceOpsRepeatEvalTest extends AnyFlatSpec with Matchers {
+  behavior of "SourceOps.repeatEval"
+
+  it should "evaluate the element before each send" in supervised {
+    var i = 0
+    val s = Source.repeatEval {
+      i += 1
+      i
+    }
+    s.take(3).toList shouldBe List(1, 2, 3)
+  }
+}

kafka/src/main/scala/ox/kafka/KafkaConsumerActor.scala

@@ -7,7 +7,6 @@ import ox.*
 import ox.channels.*
 
 import scala.jdk.CollectionConverters.*
-import scala.util.control.NonFatal
 
 object KafkaConsumerActor:
   private val logger = LoggerFactory.getLogger(classOf[KafkaConsumerActor.type])
@@ -30,30 +29,30 @@ object KafkaConsumerActor:
                 consumer.subscribe(topics.asJava)
                 true
               catch
-                case NonFatal(e) =>
-                  logger.error(s"Exception when subscribing to $topics", e)
-                  c.errorSafe(e)
+                case t: Throwable =>
+                  logger.error(s"Exception when subscribing to $topics", t)
+                  c.errorSafe(t)
                   false
             case KafkaConsumerRequest.Poll(results) =>
               try
                 results.send(consumer.poll(java.time.Duration.ofMillis(100)))
                 true
               catch
-                case NonFatal(e) =>
-                  logger.error("Exception when polling for records in Kafka", e)
-                  results.errorSafe(e)
-                  c.errorSafe(e)
+                case t: Throwable =>
+                  logger.error("Exception when polling for records in Kafka", t)
+                  results.errorSafe(t)
+                  c.errorSafe(t)
                   false
             case KafkaConsumerRequest.Commit(offsets, result) =>
               try
                 consumer.commitSync(offsets.view.mapValues(o => new OffsetAndMetadata(o + 1)).toMap.asJava)
                 result.sendSafe(())
                 true
               catch
-                case NonFatal(e) =>
-                  logger.error("Exception when committing offsets", e)
-                  result.errorSafe(e)
-                  c.errorSafe(e)
+                case t: Throwable =>
+                  logger.error("Exception when committing offsets", t)
+                  result.errorSafe(t)
+                  c.errorSafe(t)
                   false
         }
       finally

kafka/src/main/scala/ox/kafka/KafkaSource.scala

@@ -5,8 +5,6 @@ import org.slf4j.LoggerFactory
 import ox.*
 import ox.channels.*
 
-import scala.util.control.NonFatal
-
 object KafkaSource:
   private val logger = LoggerFactory.getLogger(classOf[KafkaSource.type])
 
@@ -38,9 +36,9 @@ object KafkaSource:
           records.forEach(r => c.send(ReceivedMessage(kafkaConsumer, r)))
         }
       catch
-        case NonFatal(e) =>
-          logger.error("Exception when polling for records", e)
-          c.errorSafe(e)
+        case t: Throwable =>
+          logger.error("Exception when polling for records", t)
+          c.errorSafe(t)
     }
 
     c