Passivation configured in language supports via the discovery protocol

java-support/src/main/java/io/cloudstate/javasupport/CloudState.java

@@ -125,12 +125,15 @@ public CloudState registerEventSourcedEntity(
 
     final String persistenceId;
     final int snapshotEvery;
+    final int passivationTimeout;
     if (entity.persistenceId().isEmpty()) {
       persistenceId = entityClass.getSimpleName();
       snapshotEvery = 0; // Default
+      passivationTimeout = 0; // Default
     } else {
       persistenceId = entity.persistenceId();
       snapshotEvery = entity.snapshotEvery();
+      passivationTimeout = entity.passivationTimeout();
     }
 
     final AnySupport anySupport = newAnySupport(additionalDescriptors);
@@ -141,7 +144,8 @@ public CloudState registerEventSourcedEntity(
             descriptor,
             anySupport,
             persistenceId,
-            snapshotEvery);
+            snapshotEvery,
+            passivationTimeout);
 
     services.put(descriptor.getFullName(), system -> service);
 
@@ -169,6 +173,7 @@ public CloudState registerEventSourcedEntity(
       Descriptors.ServiceDescriptor descriptor,
       String persistenceId,
       int snapshotEvery,
+      int passivationTimeout,
       Descriptors.FileDescriptor... additionalDescriptors) {
     services.put(
         descriptor.getFullName(),
@@ -178,7 +183,8 @@ public CloudState registerEventSourcedEntity(
                 descriptor,
                 newAnySupport(additionalDescriptors),
                 persistenceId,
-                snapshotEvery));
+                snapshotEvery,
+                passivationTimeout));
 
     return this;
   }
@@ -205,13 +211,21 @@ public CloudState registerCrdtEntity(
           entityClass + " does not declare an " + CrdtEntity.class + " annotation!");
     }
 
+    final int passivationTimeout;
+    if (entity.passivationTimeout() == 0) {
+      passivationTimeout = 0; // Default
+    } else {
+      passivationTimeout = entity.passivationTimeout();
+    }
+
     final AnySupport anySupport = newAnySupport(additionalDescriptors);
 
     CrdtStatefulService service =
         new CrdtStatefulService(
             new AnnotationBasedCrdtSupport(entityClass, anySupport, descriptor),
             descriptor,
-            anySupport);
+            anySupport,
+            passivationTimeout);
 
     services.put(descriptor.getFullName(), system -> service);
 
@@ -233,11 +247,13 @@ public CloudState registerCrdtEntity(
   public CloudState registerCrdtEntity(
       CrdtEntityFactory factory,
       Descriptors.ServiceDescriptor descriptor,
+      int passivationTimeout,
       Descriptors.FileDescriptor... additionalDescriptors) {
     services.put(
         descriptor.getFullName(),
         system ->
-            new CrdtStatefulService(factory, descriptor, newAnySupport(additionalDescriptors)));
+            new CrdtStatefulService(
+                factory, descriptor, newAnySupport(additionalDescriptors), passivationTimeout));
 
     return this;
   }
@@ -327,10 +343,13 @@ public CloudState registerEntity(
     }
 
     final String persistenceId;
+    final int passivationTimeout;
     if (entity.persistenceId().isEmpty()) {
       persistenceId = entityClass.getSimpleName();
+      passivationTimeout = 0; // Default
     } else {
       persistenceId = entity.persistenceId();
+      passivationTimeout = entity.passivationTimeout();
     }
 
     final AnySupport anySupport = newAnySupport(additionalDescriptors);
@@ -339,7 +358,8 @@ public CloudState registerEntity(
             new AnnotationBasedEntitySupport(entityClass, anySupport, descriptor),
             descriptor,
             anySupport,
-            persistenceId);
+            persistenceId,
+            passivationTimeout);
 
     services.put(descriptor.getFullName(), system -> service);
 
@@ -363,12 +383,17 @@ public CloudState registerEntity(
       EntityFactory factory,
       Descriptors.ServiceDescriptor descriptor,
       String persistenceId,
+      int passivationTimeout,
       Descriptors.FileDescriptor... additionalDescriptors) {
     services.put(
         descriptor.getFullName(),
         system ->
             new ValueEntityStatefulService(
-                factory, descriptor, newAnySupport(additionalDescriptors), persistenceId));
+                factory,
+                descriptor,
+                newAnySupport(additionalDescriptors),
+                persistenceId,
+                passivationTimeout));
 
     return this;
   }

java-support/src/main/java/io/cloudstate/javasupport/crdt/CrdtEntity.java

@@ -40,4 +40,12 @@
 @CloudStateAnnotation
 @Target(ElementType.TYPE)
 @Retention(RetentionPolicy.RUNTIME)
-public @interface CrdtEntity {}
+public @interface CrdtEntity {
+
+  /**
+   * Specifies the idle time of the entity before it is passivated. Zero and any negative value mean
+   * use default from configuration file. Any positive value means the entity is passivated after
+   * the idle time.
+   */
+  int passivationTimeout() default 0;
+}

java-support/src/main/java/io/cloudstate/javasupport/entity/Entity.java

@@ -35,4 +35,11 @@
    * that you specify it explicitly.
    */
   String persistenceId() default "";
+
+  /**
+   * Specifies the idle time of the entity before it is passivated. Zero and any negative value mean
+   * use default from configuration file. Any positive value means the entity is passivated after
+   * the idle time.
+   */
+  int passivationTimeout() default 0;
 }

java-support/src/main/java/io/cloudstate/javasupport/eventsourced/EventSourcedEntity.java

@@ -31,7 +31,7 @@
   /**
    * The name of the persistence id.
    *
-   * <p>If not specifed, defaults to the entities unqualified classname. It's strongly recommended
+   * <p>If not specified, defaults to the entities unqualified classname. It's strongly recommended
    * that you specify it explicitly.
    */
   String persistenceId() default "";
@@ -42,4 +42,11 @@
    * snapshot at-or-after that number of events.
    */
   int snapshotEvery() default 0;
+
+  /**
+   * Specifies the idle time of the entity before it is passivated. Zero and any negative value mean
+   * use default from configuration file. Any positive value means the entity is passivated after
+   * the idle time.
+   */
+  int passivationTimeout() default 0;
 }

java-support/src/main/scala/io/cloudstate/javasupport/CloudStateRunner.scala

@@ -162,8 +162,7 @@ final class CloudStateRunner private[this] (
 }
 
 /**
- * StatefulService describes an entitiy type in a way which makes it possible
- * to deploy.
+ * Service describes an entity type in a way which makes it possible to deploy.
  */
 trait Service {
 
@@ -174,15 +173,24 @@ trait Service {
 
   /**
    * Possible values are: "", "", "".
-   * @return the type of entity represented by this StatefulService
+   * @return the type of entity represented by this service
    */
   def entityType: String
 
   /**
-   * @return the persistence identifier used for the the entities represented by this service
+   * @return the persistence identifier used for the entities represented by this service
    */
   def persistenceId: String = descriptor.getName
 
-  // TODO JavaDoc
+  /**
+   * @return the idle time after which entities represented by this service should be passivated
+   */
+  def passivationTimeout: Int
+
+  /**
+   * @return a dictionary of service methods (Protobuf Descriptors.MethodDescriptor) classified by method name.
+   *         The dictionary values represent a mapping of Protobuf Descriptors.MethodDescriptor with its input
+   *         and output types (see [[io.cloudstate.javasupport.impl.ResolvedServiceMethod]])
+   */
   def resolvedMethods: Option[Map[String, ResolvedServiceMethod[_, _]]]
 }

java-support/src/main/scala/io/cloudstate/javasupport/impl/EntityDiscoveryImpl.scala

@@ -72,7 +72,7 @@ class EntityDiscoveryImpl(system: ActorSystem, services: Map[String, Service]) e
 
     val entities = services.map {
       case (name, service) =>
-        Entity(service.entityType, name, service.persistenceId)
+        Entity(service.entityType, name, service.persistenceId, service.passivationTimeout)
     }.toSeq
 
     Future.successful(EntitySpec(fileDescriptorSet, entities, Some(serviceInfo)))

java-support/src/main/scala/io/cloudstate/javasupport/impl/action/ActionImpl.scala

@@ -31,7 +31,7 @@ import io.cloudstate.protocol.entity.{Failure, Forward, Reply, SideEffect, Metad
 
 import scala.concurrent.Future
 import scala.compat.java8.FutureConverters._
-import scala.collection.JavaConverters._
+import scala.jdk.CollectionConverters._
 
 final class ActionService(val actionHandler: ActionHandler,
                           override val descriptor: Descriptors.ServiceDescriptor,
@@ -45,6 +45,8 @@ final class ActionService(val actionHandler: ActionHandler,
     }
 
   override final val entityType = ActionProtocol.name
+
+  override final val passivationTimeout = 0
 }
 
 final class ActionProtocolImpl(_system: ActorSystem, services: Map[String, ActionService], rootContext: Context)

java-support/src/main/scala/io/cloudstate/javasupport/impl/crdt/CrdtImpl.scala

@@ -49,11 +49,12 @@ import io.cloudstate.protocol.entity.{Command, Failure, StreamCancelled}
 import com.google.protobuf.any.{Any => ScalaPbAny}
 import com.google.protobuf.{Any => JavaPbAny}
 
-import scala.collection.JavaConverters._
+import scala.jdk.CollectionConverters._
 
 final class CrdtStatefulService(val factory: CrdtEntityFactory,
                                 override val descriptor: Descriptors.ServiceDescriptor,
-                                val anySupport: AnySupport)
+                                val anySupport: AnySupport,
+                                override val passivationTimeout: Int)
     extends Service {
   override final val entityType = Crdt.name
 

java-support/src/main/scala/io/cloudstate/javasupport/impl/entity/EntityImpl.scala

@@ -44,7 +44,8 @@ import scala.util.control.NonFatal
 final class ValueEntityStatefulService(val factory: EntityFactory,
                                        override val descriptor: Descriptors.ServiceDescriptor,
                                        val anySupport: AnySupport,
-                                       override val persistenceId: String)
+                                       override val persistenceId: String,
+                                       override val passivationTimeout: Int)
     extends Service {
 
   override def resolvedMethods: Option[Map[String, ResolvedServiceMethod[_, _]]] =

java-support/src/main/scala/io/cloudstate/javasupport/impl/eventsourced/EventSourcedImpl.scala

@@ -52,7 +52,8 @@ final class EventSourcedStatefulService(val factory: EventSourcedEntityFactory,
                                         override val descriptor: Descriptors.ServiceDescriptor,
                                         val anySupport: AnySupport,
                                         override val persistenceId: String,
-                                        val snapshotEvery: Int)
+                                        val snapshotEvery: Int,
+                                        override val passivationTimeout: Int)
     extends Service {
 
   override def resolvedMethods: Option[Map[String, ResolvedServiceMethod[_, _]]] =
@@ -64,7 +65,12 @@ final class EventSourcedStatefulService(val factory: EventSourcedEntityFactory,
   override final val entityType = EventSourced.name
   final def withSnapshotEvery(snapshotEvery: Int): EventSourcedStatefulService =
     if (snapshotEvery != this.snapshotEvery)
-      new EventSourcedStatefulService(this.factory, this.descriptor, this.anySupport, this.persistenceId, snapshotEvery)
+      new EventSourcedStatefulService(this.factory,
+                                      this.descriptor,
+                                      this.anySupport,
+                                      this.persistenceId,
+                                      snapshotEvery,
+                                      this.passivationTimeout)
     else
       this
 }

protocols/protocol/cloudstate/entity.proto

@@ -262,6 +262,9 @@ message Entity {
     // The ID to namespace state by. How this is used depends on the type of entity, for example,
     // event sourced entities will prefix this to the persistence id.
     string persistence_id = 3;
+
+    // The idle timeout, in seconds, after which an entity is passivated.
+    int32 passivation_timeout = 4;
 }
 
 message UserFunctionError {

proxy/core/src/main/scala/io/cloudstate/proxy/crdt/CrdtSupportFactory.scala

@@ -18,6 +18,7 @@ package io.cloudstate.proxy.crdt
 
 import java.net.URLEncoder
 import java.nio.charset.StandardCharsets.UTF_8
+import java.util.concurrent.TimeUnit
 
 import akka.{Done, NotUsed}
 import akka.actor.{ActorRef, ActorSystem, CoordinatedShutdown}
@@ -34,7 +35,6 @@ import io.cloudstate.protocol.entity.{Entity, EntityDiscovery, Metadata}
 import io.cloudstate.proxy._
 import io.cloudstate.proxy.entity.{EntityCommand, UserFunctionReply}
 
-import scala.collection.JavaConverters._
 import scala.concurrent.{ExecutionContext, Future}
 import scala.concurrent.duration._
 
@@ -54,12 +54,7 @@ class CrdtSupportFactory(system: ActorSystem,
 
     validate(serviceDescriptor, methodDescriptors)
 
-    val crdtEntityConfig = CrdtEntity.Configuration(entity.serviceName,
-                                                    entity.persistenceId,
-                                                    config.crdtSettings.passivationTimeout,
-                                                    config.relayOutputBufferSize,
-                                                    3.seconds,
-                                                    5.seconds)
+    val crdtEntityConfig = crdtEntityConfiguration(config, entity)
 
     log.debug("Starting CrdtEntity for {}", entity.serviceName)
 
@@ -99,6 +94,24 @@ class CrdtSupportFactory(system: ActorSystem,
       )
     }
   }
+
+  private def crdtEntityConfiguration(config: EntityDiscoveryManager.Configuration,
+                                      entity: Entity): CrdtEntity.Configuration = {
+    val passivationTimeout = if (entity.passivationTimeout > 0) {
+      Timeout(Duration(entity.passivationTimeout, TimeUnit.SECONDS).toMillis.millis)
+    } else {
+      config.crdtSettings.passivationTimeout
+    }
+
+    CrdtEntity.Configuration(
+      entity.serviceName,
+      entity.persistenceId,
+      passivationTimeout,
+      config.relayOutputBufferSize,
+      3.seconds, // TODO make it configurable
+      5.seconds // TODO make it configurable
+    )
+  }
 }
 
 private class CrdtSupport(crdtEntity: ActorRef, parallelism: Int, private implicit val relayTimeout: Timeout)