Addresses multiple items from design review

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

@@ -5,6 +5,8 @@
 
 package io.dapr.examples.actors.http;
 
+import reactor.core.publisher.Mono;
+
 /**
  * Example of implementation of an Actor.
  */
@@ -15,4 +17,6 @@ public interface DemoActor {
   String say(String something);
 
   void clock(String message);
+
+  Mono<Integer> incrementAndGet(int delta);
 }

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

@@ -8,6 +8,7 @@
 import io.dapr.actors.ActorId;
 import io.dapr.actors.client.ActorProxy;
 import io.dapr.actors.client.ActorProxyBuilder;
+import io.dapr.client.DefaultObjectSerializer;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -34,12 +35,12 @@ public class DemoActorClient {
   private static final ExecutorService POOL = Executors.newFixedThreadPool(NUM_ACTORS);
 
   public static void main(String[] args) throws Exception {
-    ActorProxyBuilder builder = new ActorProxyBuilder();
+    ActorProxyBuilder builder = new ActorProxyBuilder("DemoActor", new DefaultObjectSerializer());
 
     List<CompletableFuture<Void>> futures = new ArrayList<>(NUM_ACTORS);
 
     for (int i = 0; i < NUM_ACTORS; i++) {
-      ActorProxy actor = builder.withActorType("DemoActor").withActorId(ActorId.createRandom()).build();
+      ActorProxy actor = builder.build(ActorId.createRandom());
       futures.add(callActorNTimes(actor));
     }
 
@@ -54,6 +55,7 @@ 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++) {
+        actor.invokeActorMethod("incrementAndGet", 1).block();
         String result = actor.invokeActorMethod(METHOD_NAME,
           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));
@@ -65,6 +67,9 @@ private static final CompletableFuture<Void> callActorNTimes(ActorProxy actor) {
           return;
         }
       }
+
+      System.out.println(
+        "Messages sent: " + actor.invokeActorMethod("incrementAndGet", 0, int.class).block());
     }, POOL);
   }
 }

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

@@ -6,7 +6,10 @@
 package io.dapr.examples.actors.http;
 
 import io.dapr.actors.ActorId;
-import io.dapr.actors.runtime.*;
+import io.dapr.actors.runtime.AbstractActor;
+import io.dapr.actors.runtime.ActorRuntimeContext;
+import io.dapr.actors.runtime.ActorType;
+import io.dapr.actors.runtime.Remindable;
 import reactor.core.publisher.Mono;
 
 import java.text.DateFormat;
@@ -18,8 +21,8 @@
 /**
  * Implementation of the DemoActor for the server side.
  */
-@ActorType(Name = "DemoActor")
-public class DemoActorImpl extends AbstractActor implements DemoActor, Actor, Remindable<Integer> {
+@ActorType(name = "DemoActor")
+public class DemoActorImpl extends AbstractActor implements DemoActor, Remindable<Integer> {
 
   /**
    * Format to output date and time.
@@ -56,10 +59,20 @@ public String say(String something) {
       super.getId() + ": " +
       (something == null ? "" : something + " @ " + utcNowAsString));
 
+    super.getActorStateManager().set("lastmessage", something).block();
+
     // Now respond with current timestamp.
     return utcNowAsString;
   }
 
+  @Override
+  public Mono<Integer> incrementAndGet(int delta) {
+    return super.getActorStateManager().contains("counter")
+      .flatMap(exists -> exists ? super.getActorStateManager().get("counter", int.class) : Mono.just(0))
+      .map(c -> c + delta)
+      .flatMap(c -> super.getActorStateManager().set("counter", c).thenReturn(c));
+  }
+
   @Override
   public void clock(String message) {
     Calendar utcNow = Calendar.getInstance(TimeZone.getTimeZone("GMT"));

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

@@ -6,6 +6,7 @@
 package io.dapr.examples.actors.http;
 
 import io.dapr.actors.runtime.ActorRuntime;
+import io.dapr.client.DefaultObjectSerializer;
 import io.dapr.springboot.DaprApplication;
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.CommandLineParser;
@@ -35,7 +36,7 @@ public static void main(String[] args) throws Exception {
     int port = Integer.parseInt(cmd.getOptionValue("port"));
 
     // Register the Actor class.
-    ActorRuntime.getInstance().registerActor(DemoActorImpl.class);
+    ActorRuntime.getInstance().registerActor(DemoActorImpl.class, new DefaultObjectSerializer());
 
     // Start Dapr's callback endpoint.
     DaprApplication.start(port);

examples/src/main/java/io/dapr/examples/bindings/http/OutputBindingExample.java

@@ -7,7 +7,7 @@
 
 import io.dapr.client.DaprClient;
 import io.dapr.client.DaprClientBuilder;
-import io.dapr.utils.ObjectSerializer;
+import io.dapr.client.DefaultObjectSerializer;
 
 /**
  * Service for output binding example.
@@ -25,7 +25,7 @@ public MyClass(){}
   }
 
   public static void main(String[] args) throws Exception {
-    DaprClient client = new DaprClientBuilder().build();
+    DaprClient client = new DaprClientBuilder(new DefaultObjectSerializer()).build();
 
     final String BINDING_NAME = "sample123";
 

examples/src/main/java/io/dapr/examples/invoke/http/InvokeClient.java

@@ -7,6 +7,7 @@
 
 import io.dapr.client.DaprClient;
 import io.dapr.client.DaprClientBuilder;
+import io.dapr.client.DefaultObjectSerializer;
 import io.dapr.client.domain.Verb;
 
 /**
@@ -27,7 +28,7 @@ public class InvokeClient {
    * @param args Messages to be sent as request for the invoke API.
    */
   public static void main(String[] args) {
-    DaprClient client = (new DaprClientBuilder()).build();
+    DaprClient client = (new DaprClientBuilder(new DefaultObjectSerializer())).build();
     for (String message : args) {
       client.invokeService(Verb.POST, SERVICE_APP_ID, "say", message, null, String.class).block();
     }

examples/src/main/java/io/dapr/examples/pubsub/http/Publisher.java

@@ -7,6 +7,7 @@
 
 import io.dapr.client.DaprClient;
 import io.dapr.client.DaprClientBuilder;
+import io.dapr.client.DefaultObjectSerializer;
 
 import java.util.Collections;
 
@@ -24,7 +25,7 @@ public class Publisher {
   private static final String TOPIC_NAME = "message";
 
   public static void main(String[] args) throws Exception {
-    DaprClient client = new DaprClientBuilder().build();
+    DaprClient client = new DaprClientBuilder(new DefaultObjectSerializer()).build();
     for (int i = 0; i < NUM_MESSAGES; i++) {
       String message = String.format("This is message #%d", i);
       client.publishEvent(TOPIC_NAME, message).block();

examples/src/main/java/io/dapr/examples/pubsub/http/SubscriberController.java

@@ -5,8 +5,8 @@
 
 package io.dapr.examples.pubsub.http;
 
-import io.dapr.client.domain.CloudEventEnvelope;
-import io.dapr.utils.ObjectSerializer;
+import io.dapr.client.DefaultObjectSerializer;
+import io.dapr.client.domain.CloudEvent;
 import org.springframework.web.bind.annotation.*;
 import reactor.core.publisher.Mono;
 
@@ -21,7 +21,7 @@ public class SubscriberController {
   /**
    * Dapr's default serializer/deserializer.
    */
-  private static final ObjectSerializer SERIALIZER = new ObjectSerializer();
+  private static final DefaultObjectSerializer SERIALIZER = new DefaultObjectSerializer();
 
   @GetMapping("/dapr/subscribe")
   public byte[] daprConfig() throws Exception {
@@ -34,7 +34,7 @@ public Mono<Void> handleMessage(@RequestBody(required = false) byte[] body,
     return Mono.fromRunnable(() -> {
       try {
         // Dapr's event is compliant to CloudEvent.
-        CloudEventEnvelope envelope = SERIALIZER.deserialize(body, CloudEventEnvelope.class);
+        CloudEvent envelope = CloudEvent.deserialize(body);
 
         String message = envelope.getData() == null ? "" : new String(envelope.getData());
         System.out.println("Subscriber got message: " + message);

examples/src/main/java/io/dapr/springboot/DaprController.java

@@ -5,9 +5,7 @@
 
 package io.dapr.springboot;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
 import io.dapr.actors.runtime.ActorRuntime;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.*;
 import reactor.core.publisher.Mono;
 
@@ -17,18 +15,13 @@
 @RestController
 public class DaprController {
 
-  @Autowired
-  private ObjectMapper objectMapper;
-
-  private String topics;
-
   @GetMapping("/")
   public String index() {
     return "Greetings from Dapr!";
   }
 
   @GetMapping("/dapr/config")
-  public String daprConfig() throws Exception {
+  public byte[] daprConfig() throws Exception {
     return ActorRuntime.getInstance().serializeConfig();
   }
 
@@ -45,10 +38,10 @@ public Mono<Void> deactivateActor(@PathVariable("type") String type,
   }
 
   @PutMapping(path = "/actors/{type}/{id}/method/{method}")
-  public Mono<String> invokeActorMethod(@PathVariable("type") String type,
+  public Mono<byte[]> invokeActorMethod(@PathVariable("type") String type,
                                         @PathVariable("id") String id,
                                         @PathVariable("method") String method,
-                                        @RequestBody(required = false) String body) {
+                                        @RequestBody(required = false) byte[] body) {
     return ActorRuntime.getInstance().invoke(type, id, method, body);
   }
 
@@ -63,7 +56,7 @@ public Mono<Void> invokeActorTimer(@PathVariable("type") String type,
   public Mono<Void> invokeActorReminder(@PathVariable("type") String type,
                                         @PathVariable("id") String id,
                                         @PathVariable("reminder") String reminder,
-                                        @RequestBody(required = false) String body) {
+                                        @RequestBody(required = false) byte[] body) {
     return ActorRuntime.getInstance().invokeReminder(type, id, reminder, body);
   }
 

sdk-actors/pom.xml

@@ -48,6 +48,18 @@
       <artifactId>mockito-core</artifactId>
       <scope>test</scope>
     </dependency>
+      <dependency>
+          <groupId>com.github.gmazzo</groupId>
+          <artifactId>okhttp-mock</artifactId>
+          <version>1.3.2</version>
+          <scope>test</scope>
+      </dependency>
+    <dependency>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter-api</artifactId>
+      <version>5.5.2</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>

sdk-actors/src/main/java/io/dapr/actors/client/ActorProxyBuilder.java

@@ -1,77 +1,63 @@
 package io.dapr.actors.client;
 
 import io.dapr.actors.ActorId;
-import io.dapr.actors.runtime.ActorStateSerializer;
-import io.dapr.client.DaprClientBuilder;
-import okhttp3.OkHttpClient;
+import io.dapr.client.DaprHttpBuilder;
+import io.dapr.client.DaprObjectSerializer;
 
 /**
- * Builder to generate an ActorProxy instance.
+ * Builder to generate an ActorProxy instance. Builder can be reused for multiple instances.
  */
 public class ActorProxyBuilder {
 
     /**
-     * Serializer for content to be sent back and forth between actors.
+     * Builder for Dapr's raw http client.
      */
-    private static final ActorStateSerializer SERIALIZER = new ActorStateSerializer();
-
-    /**
-     * Builder for the Dapr client.
-     */
-    private final DaprClientBuilder clientBuilder = new DaprClientBuilder();
+    private final DaprHttpBuilder daprHttpBuilder = new DaprHttpBuilder();
 
     /**
      * Actor's type.
      */
-    private String actorType;
+    private final String actorType;
 
     /**
-     * Actor's identifier.
+     * Dapr's object serializer.
      */
-    private ActorId actorId;
+    private final DaprObjectSerializer serializer;
 
     /**
-     * Changes build config to use given Actor's type.
+     * Instantiates a new builder for a given Actor type.
      *
      * @param actorType Actor's type.
-     * @return Same builder object.
+     * @param serializer Serializer for objects sent/received. Use null for default (not recommended).
      */
-    public ActorProxyBuilder withActorType(String actorType) {
-        this.actorType = actorType;
-        return this;
-    }
+    public ActorProxyBuilder(String actorType, DaprObjectSerializer serializer) {
+        if ((actorType == null) || actorType.isEmpty()) {
+            throw new IllegalArgumentException("ActorType is required.");
+        }
+        if (serializer == null) {
+            throw new IllegalArgumentException("Serializer is required.");
+        }
 
-    /**
-     * Changes build config to use given Actor's identifier.
-     *
-     * @param actorId Actor's identifier.
-     * @return Same builder object.
-     */
-    public ActorProxyBuilder withActorId(ActorId actorId) {
-        this.actorId = actorId;
-        return this;
+        this.actorType = actorType;
+        this.serializer = serializer;
     }
 
     /**
      * Instantiates a new ActorProxy.
      *
+     * @param actorId Actor's identifier.
      * @return New instance of ActorProxy.
      */
-    public ActorProxy build() {
-        if ((this.actorType == null) || this.actorType.isEmpty()) {
-            throw new IllegalArgumentException("Cannot instantiate an Actor without type.");
-        }
-
-        if (this.actorId == null) {
+    public ActorProxy build(ActorId actorId) {
+        if (actorId == null) {
             throw new IllegalArgumentException("Cannot instantiate an Actor without Id.");
         }
 
-        // TODO: Share client between actor proxy instances.
         return new ActorProxyImpl(
                 this.actorType,
-                this.actorId,
-                SERIALIZER,
-                this.clientBuilder.build());
+                actorId,
+                this.serializer,
+                new DaprHttpClient(this.daprHttpBuilder.build()));
     }
 
 }