Unit tests for ActorManager + fixes for Timer and Reminder.

examples/pom.xml

@@ -34,6 +34,11 @@
       <artifactId>commons-cli</artifactId>
       <version>1.4</version>
     </dependency>
+    <dependency>
+      <groupId>commons-io</groupId>
+      <artifactId>commons-io</artifactId>
+      <version>2.6</version>
+    </dependency>
     <dependency>
       <groupId>org.json</groupId>
       <artifactId>json</artifactId>

examples/src/main/java/io/dapr/examples/actors/http/DemoActor.java

@@ -10,5 +10,9 @@
  */
 public interface DemoActor {
 
+  void registerReminder();
+
   String say(String something);
+
+  void clock(String message);
 }

examples/src/main/java/io/dapr/examples/actors/http/DemoActorClient.java

@@ -8,7 +8,6 @@
 import io.dapr.actors.ActorId;
 import io.dapr.actors.client.ActorProxy;
 import io.dapr.actors.client.ActorProxyBuilder;
-import reactor.core.publisher.Mono;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -53,12 +52,13 @@ public static void main(String[] args) throws Exception {
 
   private static final CompletableFuture<Void> callActorNTimes(ActorProxy actor) {
     return CompletableFuture.runAsync(() -> {
+      actor.invokeActorMethod("registerReminder").block();
       for (int i = 0; i < NUM_MESSAGES_PER_ACTOR; i++) {
         String result = actor.invokeActorMethod(METHOD_NAME,
-          String.format("Actor %s said message #%d", actor.getActorId().toString(), i)).block();
+          String.format("Actor %s said message #%d", actor.getActorId().toString(), i), String.class).block();
         System.out.println(String.format("Actor %s got a reply: %s", actor.getActorId().toString(), result));
         try {
-          Thread.sleep(1000);
+          Thread.sleep((long)(1000 * Math.random()));
         } catch (InterruptedException e) {
           e.printStackTrace();
           Thread.currentThread().interrupt();

examples/src/main/java/io/dapr/examples/actors/http/DemoActorImpl.java

@@ -6,21 +6,20 @@
 package io.dapr.examples.actors.http;
 
 import io.dapr.actors.ActorId;
-import io.dapr.actors.runtime.AbstractActor;
-import io.dapr.actors.runtime.Actor;
-import io.dapr.actors.runtime.ActorRuntimeContext;
-import io.dapr.actors.runtime.ActorType;
+import io.dapr.actors.runtime.*;
+import reactor.core.publisher.Mono;
 
 import java.text.DateFormat;
 import java.text.SimpleDateFormat;
+import java.time.Duration;
 import java.util.Calendar;
 import java.util.TimeZone;
 
 /**
  * Implementation of the DemoActor for the server side.
  */
 @ActorType(Name = "DemoActor")
-public class DemoActorImpl extends AbstractActor implements DemoActor, Actor {
+public class DemoActorImpl extends AbstractActor implements DemoActor, Actor, Remindable<Integer> {
 
   /**
    * Format to output date and time.
@@ -29,6 +28,22 @@ public class DemoActorImpl extends AbstractActor implements DemoActor, Actor {
 
   public DemoActorImpl(ActorRuntimeContext runtimeContext, ActorId id) {
     super(runtimeContext, id);
+
+    super.registerActorTimer(
+      null,
+      "clock",
+      "ping!",
+      Duration.ofSeconds(2),
+      Duration.ofSeconds(1));
+  }
+
+  @Override
+  public void registerReminder() {
+    super.registerReminder(
+      "myremind",
+      (int)(Integer.MAX_VALUE * Math.random()),
+      Duration.ofSeconds(5),
+      Duration.ofSeconds(2));
   }
 
   @Override
@@ -37,9 +52,39 @@ public String say(String something) {
     String utcNowAsString = DATE_FORMAT.format(utcNow.getTime());
 
     // Handles the request by printing message.
-    System.out.println("Server: " + something == null ? "" : something + " @ " + utcNowAsString);
+    System.out.println("Server say method for actor " +
+      super.getId() + ": " +
+      (something == null ? "" : something + " @ " + utcNowAsString));
 
     // Now respond with current timestamp.
     return utcNowAsString;
   }
+
+  @Override
+  public void clock(String message) {
+    Calendar utcNow = Calendar.getInstance(TimeZone.getTimeZone("GMT"));
+    String utcNowAsString = DATE_FORMAT.format(utcNow.getTime());
+
+    // Handles the request by printing message.
+    System.out.println("Server timer for actor " +
+      super.getId() + ": " +
+      (message == null ? "" : message + " @ " + utcNowAsString));
+  }
+
+  @Override
+  public Class<Integer> getStateType() {
+    return Integer.class;
+  }
+
+  @Override
+  public Mono<Void> receiveReminder(String reminderName, Integer state, Duration dueTime, Duration period) {
+    Calendar utcNow = Calendar.getInstance(TimeZone.getTimeZone("GMT"));
+    String utcNowAsString = DATE_FORMAT.format(utcNow.getTime());
+
+    // Handles the request by printing message.
+    System.out.println(String.format(
+      "Server reminded actor %s of: %s for %d @ %s",
+      this.getId(), reminderName, state, utcNowAsString));
+    return Mono.empty();
+  }
 }

examples/src/main/java/io/dapr/examples/actors/http/DemoActorService.java

@@ -19,6 +19,7 @@
 import org.apache.commons.cli.CommandLineParser;
 import org.apache.commons.cli.DefaultParser;
 import org.apache.commons.cli.Options;
+import org.apache.commons.io.IOUtils;
 
 import java.io.IOException;
 import java.io.InputStream;
@@ -46,7 +47,9 @@ public class DemoActorService {
     .get("/dapr/config", DemoActorService::handleDaprConfig)
     .post("/actors/{actorType}/{id}", DemoActorService::handleActorActivate)
     .delete("/actors/{actorType}/{id}", DemoActorService::handleActorDeactivate)
-    .put("/actors/{actorType}/{id}/method/{methodName}", DemoActorService::handleActorInvoke);
+    .put("/actors/{actorType}/{id}/method/{methodName}", DemoActorService::handleActorInvoke)
+    .put("/actors/{actorType}/{id}/method/timer/{timerName}", DemoActorService::handleActorTimer)
+    .put("/actors/{actorType}/{id}/method/remind/{reminderName}", DemoActorService::handleActorReminder);
 
   private final int port;
 
@@ -135,11 +138,39 @@ private static void handleActorInvoke(HttpServerExchange exchange) throws IOExce
     String actorId = findParamValueOrNull(exchange, "id");
     String methodName = findParamValueOrNull(exchange, "methodName");
     exchange.startBlocking();
-    String data = findData(exchange.getInputStream());
+    String data = findMethodData(exchange.getInputStream());
     String result = ActorRuntime.getInstance().invoke(actorType, actorId, methodName, data).block();
     exchange.getResponseSender().send(buildResponse(result));
   }
 
+  private static void handleActorTimer(HttpServerExchange exchange) throws IOException {
+    if (exchange.isInIoThread()) {
+      exchange.dispatch(DemoActorService::handleActorTimer);
+      return;
+    }
+
+    String actorType = findParamValueOrNull(exchange, "actorType");
+    String actorId = findParamValueOrNull(exchange, "id");
+    String timerName = findParamValueOrNull(exchange, "timerName");
+    ActorRuntime.getInstance().invokeTimer(actorType, actorId, timerName).block();
+    exchange.getResponseSender().send("");
+  }
+
+  private static void handleActorReminder(HttpServerExchange exchange) throws IOException {
+    if (exchange.isInIoThread()) {
+      exchange.dispatch(DemoActorService::handleActorReminder);
+      return;
+    }
+
+    String actorType = findParamValueOrNull(exchange, "actorType");
+    String actorId = findParamValueOrNull(exchange, "id");
+    String reminderName = findParamValueOrNull(exchange, "reminderName");
+    exchange.startBlocking();
+    String params = IOUtils.toString(exchange.getInputStream(), StandardCharsets.UTF_8);
+    ActorRuntime.getInstance().invokeReminder(actorType, actorId, reminderName, params).block();
+    exchange.getResponseSender().send("");
+  }
+
   private static String findParamValueOrNull(HttpServerExchange exchange, String name) {
     Map<String, Deque<String>> params = exchange.getQueryParameters();
     if (params == null) {
@@ -154,7 +185,7 @@ private static String findParamValueOrNull(HttpServerExchange exchange, String n
     return values.getFirst();
   }
 
-  private static String findData(InputStream stream) throws IOException {
+  private static String findMethodData(InputStream stream) throws IOException {
     JsonNode root = OBJECT_MAPPER.readTree(stream);
     if (root == null) {
       return null;