Naming and code refactor

.scalafmt.conf

@@ -21,4 +21,4 @@ rewrite.sortModifiers.order = [
 ]
 docstrings = ScalaDoc
 
-rewrite.neverInfix.excludeFilters = [andThen, or]
+rewrite.neverInfix.excludeFilters = [andThen, or, ne]

core/shared/src/main/scala/canoe/api/Bot.scala

@@ -5,6 +5,7 @@ import canoe.models.messages.TelegramMessage
 import canoe.models.{InputFile, Update}
 import cats.effect.concurrent.{Deferred, Ref}
 import cats.effect.{Concurrent, ConcurrentEffect, Resource, Timer}
+import cats.instances.option._
 import cats.syntax.all._
 import fs2.concurrent.Topic
 import fs2.{Pipe, Stream}
@@ -16,7 +17,6 @@ import scala.concurrent.duration.FiniteDuration
   * interact with other Telegram users in a certain predefined way
   */
 class Bot[F[_]: Concurrent] private[api] (source: UpdateSource[F]) {
-
   /**
     * Stream of all updates which your bot receives from Telegram service
     */
@@ -59,36 +59,36 @@ class Bot[F[_]: Concurrent] private[api] (source: UpdateSource[F]) {
     * @return Stream of all updates which your bot receives from Telegram service
     */
   def follow(scenarios: Scenario[F, Unit]*): Stream[F, Update] = {
+    def runScenarios(updates: Topic[F, Update]): Stream[F, Nothing] =
+      updates
+        .subscribe(1)
+        .through(pipes.messages)
+        .map(m => Stream.emits(scenarios).map(sc => fork(updates, m).through(sc.pipe)).parJoinUnbounded.drain)
+        .parJoinUnbounded
+
+    def fork(updates: Topic[F, Update], m: TelegramMessage): Stream[F, TelegramMessage] =
+      updates
+        .subscribe(1)
+        .through(filterMessages(m.chat.id))
+        .through(debounce)
+        .cons1(m)
 
     def filterMessages(id: Long): Pipe[F, Update, TelegramMessage] =
       _.through(pipes.messages).filter(_.chat.id == id)
 
-    def debounce[F[_]: Concurrent, A](in: Stream[F, A]): Stream[F, A] =
-      Stream.eval(Ref[F].of[Option[Deferred[F, A]]](None)).flatMap { ref =>
-        val hook = Stream
-          .repeatEval(Deferred[F, A])
-          .evalMap(df => ref.set(Some(df)) *> df.get)
+    def debounce[F[_]: Concurrent, A]: Pipe[F, A, A] =
+      input =>
+        Stream.eval(Ref[F].of[Option[Deferred[F, A]]](None)).flatMap { ref =>
+          val hook = Stream
+            .repeatEval(Deferred[F, A])
+            .evalMap(df => ref.set(Some(df)) *> df.get)
 
-        val update = in.evalMap(
-          a =>
-            ref.getAndSet(None).flatMap {
-              case Some(df) => df.complete(a)
-              case None     => Concurrent[F].unit
-            }
-        )
+          val update = input.evalMap { a =>
+            ref.getAndSet(None).flatMap(_.traverse_(_.complete(a)))
+          }
 
-        hook.concurrently(update)
-      }
-
-    def track(updates: Topic[F, Update], m: TelegramMessage): Stream[F, TelegramMessage] =
-      debounce(updates.subscribe(1).through(filterMessages(m.chat.id))).cons1(m)
-
-    def runScenarios(updates: Topic[F, Update]): Stream[F, Nothing] =
-      updates
-        .subscribe(1)
-        .through(pipes.messages)
-        .map(m => Stream.emits(scenarios).map(sc => track(updates, m).through(sc.pipe).take(1)).parJoinUnbounded.drain)
-        .parJoinUnbounded
+          hook.concurrently(update)
+        }
 
     Stream.eval(Broadcast[F, Update]).flatMap { topic =>
       val pop = updates.evalTap(topic.publish1)
@@ -99,7 +99,6 @@ class Bot[F[_]: Concurrent] private[api] (source: UpdateSource[F]) {
 }
 
 object Bot {
-
   /**
     * Creates a bot which receives incoming updates using long polling mechanism.
     *

core/shared/src/main/scala/canoe/api/Broadcast.scala

@@ -12,22 +12,22 @@ import fs2.concurrent.{Queue, Topic}
   * Each subscriber has own queue where each incoming elements is stored.
   *
   * Unlike stanard Topic impemenatation, doesn't contain initial value,
-  * thus each subscriber sees only elements that were published after 
-  * the subscription was done. 
+  * thus each subscriber sees only elements that were published after
+  * the subscription was done.
   */
 private[api] class Broadcast[F[_], A](subs: Ref[F, List[Queue[F, A]]])(implicit C: Concurrent[F]) extends Topic[F, A] {
   def publish: Pipe[F, A, Unit] =
     _.evalMap(publish1)
 
   def publish1(a: A): F[Unit] =
-    subs.get.flatMap(_.traverse(_.enqueue1(a)).void)
+    subs.get.flatMap(_.traverse_(_.enqueue1(a)))
 
   def subscribe(maxQueued: Int): Stream[F, A] =
-    emptyQueue(maxQueued).evalTap(q => subs.update(q :: _)).flatMap(_.dequeue)
+    subscription(maxQueued).evalTap(q => subs.update(q :: _)).flatMap(_.dequeue)
 
-  private def emptyQueue(maxQueued: Int): Stream[F, Queue[F, A]] =
+  private def subscription(maxQueued: Int): Stream[F, Queue[F, A]] =
     Stream.bracket(Queue.bounded[F, A](maxQueued))(
-      queue => subs.update(_.filter(_.ne(queue)))
+      queue => subs.update(_.filter(_ ne queue))
     )
 
   def subscribeSize(maxQueued: Int): Stream[F, (A, Int)] =

core/shared/src/test/scala/canoe/api/BotSpec.scala

@@ -22,7 +22,7 @@ class BotSpec extends AnyFunSuite {
           .emits(messages.zipWithIndex.map {
             case ((m, id), i) => MessageReceived(i, TextMessage(i, PrivateChat(id, None, None, None), -1, m))
           })
-          .metered[IO](0.1.second)
+          .metered[IO](0.2.second)
     }
 
   test("updates returns updates from the source") {