Prevent binary queue processing stream from stopping in case of error

Author: ghostdogpr

entities/src/main/scala/com/devsisters/shardcake/Sharding.scala

@@ -264,18 +264,18 @@ class Sharding private (
                            serialization
                              .decode[Req](msg.body)
                              .map(req => Some((req, msg.entityId, p, msg.replyId)))
-                             .catchAll(p.fail(_).as(None))
+                             .catchAllCause((cause: Cause[Throwable]) => p.fail(cause.squash).as(None))
                          }
                          .collectSome
                          .foreach { case (msg, entityId, p, replyId) =>
                            Promise
                              .make[Throwable, Option[Any]]
                              .flatMap(p2 =>
-                               entityManager.send(entityId, msg, replyId, p2).catchAll(p.fail) *>
-                                 p2.await.flatMap(
-                                   ZIO.foreach(_)(serialization.encode).flatMap(p.succeed(_)).catchAll(p.fail(_)).fork
-                                 )
+                               entityManager.send(entityId, msg, replyId, p2) *>
+                                 p2.await
+                                   .flatMap(ZIO.foreach(_)(serialization.encode).flatMap(p.succeed(_)).catchAll(p.fail(_)).fork)
                              )
+                             .catchAllCause((cause: Cause[Throwable]) => p.fail(cause.squash).unit)
                          }
                          .forkManaged
     } yield ()

examples/src/main/scala/example/complex/GuildApp.scala

@@ -17,7 +17,7 @@ object GuildApp extends zio.App {
 
   val program =
     (Sharding.registerEntity(Guild, behavior) *> Sharding.registerManaged).use { _ =>
-      Sharding.messenger(Guild).map { guild =>
+      Sharding.messenger(Guild).flatMap { guild =>
         for {
           user1 <- random.nextUUID.map(_.toString)
           user2 <- random.nextUUID.map(_.toString)

examples/src/main/scala/example/simple/GuildApp.scala

@@ -11,7 +11,7 @@ import zio.random.Random
 object GuildApp extends zio.App {
   val program =
     (Sharding.registerEntity(Guild, behavior) *> Sharding.registerManaged).use { _ =>
-      Sharding.messenger(Guild).map { guild =>
+      Sharding.messenger(Guild).flatMap { guild =>
         for {
           _ <- guild.send("guild1")(Join("user1", _)).debug
           _ <- guild.send("guild1")(Join("user2", _)).debug

examples/src/main/scala/example/simple/ShardManagerApp.scala

@@ -11,7 +11,7 @@ object ShardManagerApp extends zio.App {
   val logging = Logging.debug
   val pods    = ZLayer.succeed(GrpcConfig.default) ++ logging >>> GrpcPods.live // use gRPC protocol
   val health  = pods >>> PodsHealth.local                                       // just ping a pod to see if it's alive
-  val storage = Storage.memory                                                  // / store data in memory
+  val storage = Storage.memory                                                  // store data in memory
   val layer   =
     ZLayer.succeed(ManagerConfig.default) ++ Clock.live ++ Console.live ++ logging ++ pods ++ health ++ storage >+>
       ShardManager.live // Shard Manager logic