backport (#49)

Author: csar

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

@@ -323,9 +323,10 @@ class Sharding private (
   private def registerEntity[R, Req: Tag](
     entityType: EntityType[Req],
     behavior: (String, Dequeue[Req]) => RIO[R, Nothing],
-    terminateMessage: Promise[Nothing, Unit] => Option[Req] = (_: Promise[Nothing, Unit]) => None
+    terminateMessage: Promise[Nothing, Unit] => Option[Req] = (_: Promise[Nothing, Unit]) => None,
+    entityMaxIdleTime: Option[Duration] = None
   ): ZManaged[Clock with R, Nothing, Unit] =
-    registerRecipient(entityType, behavior, terminateMessage) *>
+    registerRecipient(entityType, behavior, terminateMessage, entityMaxIdleTime) *>
       eventsHub.publish(ShardingRegistrationEvent.EntityRegistered(entityType)).unit.toManaged_
 
   private def registerTopic[R, Req: Tag](
@@ -341,10 +342,12 @@ class Sharding private (
   def registerRecipient[R, Req: Tag](
     recipientType: RecipientType[Req],
     behavior: (String, Dequeue[Req]) => RIO[R, Nothing],
-    terminateMessage: Promise[Nothing, Unit] => Option[Req] = (_: Promise[Nothing, Unit]) => None
+    terminateMessage: Promise[Nothing, Unit] => Option[Req] = (_: Promise[Nothing, Unit]) => None,
+    entityMaxIdleTime: Option[Duration] = None
   ): ZManaged[Clock with R, Nothing, Unit] =
     for {
-      entityManager <- EntityManager.make(recipientType, behavior, terminateMessage, self, config).toManaged_
+      entityManager <-
+        EntityManager.make(recipientType, behavior, terminateMessage, self, config, entityMaxIdleTime).toManaged_
       binaryQueue   <- Queue
                          .unbounded[(BinaryMessage, Promise[Throwable, Option[Array[Byte]]], Promise[Nothing, Unit])]
                          .toManaged(_.shutdown)
@@ -482,11 +485,12 @@ object Sharding {
   def registerEntity[R, Req: Tag](
     entityType: EntityType[Req],
     behavior: (String, Dequeue[Req]) => RIO[R, Nothing],
-    terminateMessage: Promise[Nothing, Unit] => Option[Req] = (_: Promise[Nothing, Unit]) => None
+    terminateMessage: Promise[Nothing, Unit] => Option[Req] = (_: Promise[Nothing, Unit]) => None,
+    entityMaxIdleTime: Option[Duration] = None
   ): ZManaged[Has[Sharding] with R with Clock, Nothing, Unit] =
     for {
       sharding <- ZIO.service[Sharding].toManaged_
-      _        <- sharding.registerEntity[R, Req](entityType, behavior, terminateMessage)
+      _        <- sharding.registerEntity[R, Req](entityType, behavior, terminateMessage, entityMaxIdleTime)
     } yield ()
 
   /**

entities/src/main/scala/com/devsisters/shardcake/internal/EntityManager.scala

@@ -23,7 +23,8 @@ private[shardcake] object EntityManager {
     behavior: (String, Queue[Req]) => RIO[R, Nothing],
     terminateMessage: Promise[Nothing, Unit] => Option[Req],
     sharding: Sharding,
-    config: Config
+    config: Config,
+    entityMaxIdleTime: Option[Duration]
   ): URIO[R with Clock, EntityManager[Req]] =
     for {
       entities <- RefM.make[Map[String, (Option[Queue[Req]], Fiber[Nothing, Unit])]](Map())
@@ -36,6 +37,7 @@ private[shardcake] object EntityManager {
       entities,
       sharding,
       config,
+      entityMaxIdleTime,
       clock
     )
 
@@ -46,11 +48,12 @@ private[shardcake] object EntityManager {
     entities: RefM[Map[String, (Option[Queue[Req]], Fiber[Nothing, Unit])]],
     sharding: Sharding,
     config: Config,
+    entityMaxIdleTime: Option[Duration],
     clock: Clock.Service
   ) extends EntityManager[Req] {
     private def startExpirationFiber(entityId: String): UIO[Fiber[Nothing, Unit]] =
       (for {
-        _ <- clock.sleep(config.entityMaxIdleTime)
+        _ <- clock.sleep(entityMaxIdleTime getOrElse config.entityMaxIdleTime)
         _ <- terminateEntity(entityId).forkDaemon.unit // fork daemon otherwise it will interrupt itself
       } yield ()).forkDaemon
 

entities/src/test/scala/com/devsisters/shardcake/ShardingSpec.scala

@@ -36,13 +36,18 @@ object ShardingSpec extends DefaultRunnableSpec {
         }
       },
       testM("Entity termination") {
-        (Sharding.registerEntity(Counter, behavior) *> Sharding.registerManaged).use { _ =>
+        (Sharding.registerEntity(
+          Counter,
+          behavior,
+          entityMaxIdleTime = Some(1.seconds)
+        ) *> Sharding.registerManaged).use { _ =>
           for {
             counter <- Sharding.messenger(Counter)
-            _       <- counter.send("c3")(GetCounter.apply)
+            _       <- counter.sendDiscard("c3")(IncrementCounter)
+            c0      <- counter.send("c3")(GetCounter.apply)
             _       <- clock.sleep(3 seconds)
-            c       <- counter.send("c3")(GetCounter.apply)
-          } yield assertTrue(c == 0)
+            c1 <- counter.send("c3")(GetCounter.apply) // counter should be restarted
+          } yield assertTrue(c0 == 1, c1 == 0)
         }
       },
       testM("Cluster singleton") {