ActorProxy + some refactoring

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

@@ -0,0 +1,70 @@
+/*
+ * Copyright (c) Microsoft Corporation.
+ * Licensed under the MIT License.
+ */
+
+package io.dapr.examples.actors.http;
+
+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;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Client for Actor runtime.
+ * 1. Build and install jars:
+ * mvn clean install
+ * 2. Run the client:
+ * dapr run --app-id demoactorclient --port 3006 -- mvn exec:java -pl=examples -Dexec.mainClass=io.dapr.examples.actors.http.DemoActorClient
+ */
+public class DemoActorClient {
+
+  private static final int NUM_ACTORS = 3;
+
+  private static final int NUM_MESSAGES_PER_ACTOR = 10;
+
+  private static final String METHOD_NAME = "say";
+
+  private static final ExecutorService POOL = Executors.newFixedThreadPool(NUM_ACTORS);
+
+  public static void main(String[] args) throws Exception {
+    ActorProxyBuilder builder = new ActorProxyBuilder();
+
+    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();
+      futures.add(callActorNTimes(actor));
+    }
+
+    futures.forEach(CompletableFuture::join);
+    POOL.shutdown();
+    POOL.awaitTermination(1, TimeUnit.MINUTES);
+
+    System.out.println("Done.");
+  }
+
+  private static final CompletableFuture<Void> callActorNTimes(ActorProxy actor) {
+    return CompletableFuture.runAsync(() -> {
+      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();
+        System.out.println(String.format("Actor %s got a reply: %s", actor.getActorId().toString(), result));
+        try {
+          Thread.sleep(1000);
+        } catch (InterruptedException e) {
+          e.printStackTrace();
+          Thread.currentThread().interrupt();
+          return;
+        }
+      }
+    }, POOL);
+  }
+}

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

@@ -9,6 +9,7 @@
 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 java.text.DateFormat;
 import java.text.SimpleDateFormat;
@@ -18,6 +19,7 @@
 /**
  * Implementation of the DemoActor for the server side.
  */
+@ActorType(Name = "DemoActor")
 public class DemoActorImpl extends AbstractActor implements DemoActor, Actor {
 
   /**

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

@@ -32,8 +32,8 @@
  * Service for Actor runtime.
  * 1. Build and install jars:
  * mvn clean install
- * 2. Run in server mode:
- * dapr run --app-id hellogrpc --app-port 3000 --port 3005 -- mvn exec:java -pl=examples -Dexec.mainClass=io.dapr.examples.actors.http.DemoActorService -Dexec.args="-p 3000"
+ * 2. Run the server:
+ * dapr run --app-id demoactorservice --app-port 3000 --port 3005 -- mvn exec:java -pl=examples -Dexec.mainClass=io.dapr.examples.actors.http.DemoActorService -Dexec.args="-p 3000"
  */
 public class DemoActorService {
 

examples/src/main/java/io/dapr/examples/state/http/OrderManager.java

@@ -20,6 +20,8 @@
 import java.net.http.HttpResponse.BodyHandlers;
 import java.nio.charset.Charset;
 import java.time.Duration;
+import java.util.Arrays;
+import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 
@@ -56,7 +58,7 @@ public static void main(String[] args) throws IOException {
       out.println("Fetching order!");
       fetch(stateUrl + "/order").thenAccept(response -> {
         int resCode = response.statusCode() == 200 ? 200 : 500;
-        String body = response.statusCode() == 200 ? response.body() : "Could not get state.";
+        String body = (response.statusCode() == 200) || (response.statusCode() == 201) ? response.body() : "Could not get state.";
 
         try {
           e.sendResponseHeaders(resCode, body.getBytes().length);
@@ -89,7 +91,7 @@ public static void main(String[] args) throws IOException {
         out.printf("Writing to state: %s\n", state.toString());
 
         post(stateUrl, state.toString()).thenAccept(response -> {
-          int resCode = response.statusCode() == 200 ? 200 : 500;
+          int resCode = (response.statusCode() == 200) || (response.statusCode() == 201) ? 201 : 500;
           String body = response.body();
           try {
             e.sendResponseHeaders(resCode, body.getBytes().length);

sdk/src/main/java/io/dapr/actors/AbstractClientBuilder.java

@@ -25,12 +25,20 @@ public AbstractClientBuilder withPort(int port) {
     return this;
   }
 
+  /**
+   * Returns configured port.
+   * @return
+   */
+  protected int getPort() {
+    return this.port;
+  }
+
   /**
    * Tries to get a valid port from environment variable or returns default.
    *
    * @return Port defined in env variable or default.
    */
-  protected static int GetEnvPortOrDefault() {
+  private static int GetEnvPortOrDefault() {
     String envPort = System.getenv(Constants.ENV_DAPR_HTTP_PORT);
     if (envPort == null) {
       return Constants.DEFAULT_PORT;
@@ -44,4 +52,5 @@ protected static int GetEnvPortOrDefault() {
 
     return Constants.DEFAULT_PORT;
   }
+
 }

sdk/src/main/java/io/dapr/actors/AbstractDaprClient.java

@@ -5,16 +5,13 @@
 package io.dapr.actors;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
-import okhttp3.*;
-import reactor.core.publisher.Mono;
-
 import java.io.IOException;
 import java.net.URL;
 import java.util.UUID;
+import okhttp3.*;
+import reactor.core.publisher.Mono;
 
-/**
- * Base for Dapr HTTP Client.
- */
+// base class of hierarchy
 public abstract class AbstractDaprClient {
 
   /**
@@ -109,10 +106,10 @@ private final void tryInvokeAPI(String method, String urlString, String json, fi
     RequestBody body = json != null ? RequestBody.create(MEDIA_TYPE_APPLICATION_JSON, json) : REQUEST_BODY_EMPTY_JSON;
 
     Request request = new Request.Builder()
-        .url(new URL(this.baseUrl + urlString))
-        .method(method, body)
-        .addHeader(Constants.HEADER_DAPR_REQUEST_ID, requestId)
-        .build();
+      .url(new URL(this.baseUrl + urlString))
+      .method(method, body)
+      .addHeader(Constants.HEADER_DAPR_REQUEST_ID, requestId)
+      .build();
 
     this.httpClient.newCall(request).enqueue(new Callback() {
 
@@ -181,4 +178,4 @@ public interface DaprHttpCallback {
     public void onSuccess(String response);
   }
 
-}
+}

sdk/src/main/java/io/dapr/actors/client/ActorMethodEnvelope.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright (c) Microsoft Corporation.
+ * Licensed under the MIT License.
+ */
+
+package io.dapr.actors.client;
+
+/**
+ * Request and Response object used to talk to Actors.
+ */
+public class ActorMethodEnvelope {
+
+  /**
+   * Data serialized for input/output of Actor methods.
+   */
+  private byte[] data;
+
+  /**
+   * Gets the data serialized for input/output of Actor methods.
+   * @return Data serialized for input/output of Actor methods.
+   */
+  public byte[] getData() {
+    return data;
+  }
+
+  /**
+   * Sets the data serialized for input/output of Actor methods.
+   * @param data Data serialized for input/output of Actor methods.
+   */
+  public void setData(byte[] data) {
+    this.data = data;
+  }
+}

sdk/src/main/java/io/dapr/actors/client/ActorProxy.java

@@ -0,0 +1,63 @@
+package io.dapr.actors.client;
+
+import io.dapr.actors.ActorId;
+import reactor.core.publisher.Mono;
+
+import java.io.IOException;
+
+/**
+ * Proxy to communicate to a given Actor instance in Dapr.
+ */
+public interface ActorProxy {
+
+    /**
+     * Returns the ActorId associated with the proxy object.
+     *
+     * @return An ActorId object.
+     */
+    ActorId getActorId();
+
+    /**
+     * Returns actor implementation type of the actor associated with the proxy object.
+     *
+     * @return Actor's type name.
+     */
+    String getActorType();
+
+    /**
+     * Invokes an Actor method on Dapr.
+     *
+     * @param methodName Method name to invoke.
+     * @param clazz The type of the return class.
+     * @return Asynchronous result with the Actor's response.
+     */
+    <T> Mono<T> invokeActorMethod(String methodName, Class<T> clazz);
+
+    /**
+     * Invokes an Actor method on Dapr.
+     *
+     * @param methodName Method name to invoke.
+     * @param data Object with the data.
+     * @param clazz The type of the return class.
+     * @return Asynchronous result with the Actor's response.
+     */
+    <T> Mono<T> invokeActorMethod(String methodName, Object data,  Class<T> clazz);
+
+    /**
+     * Invokes an Actor method on Dapr.
+     *
+     * @param methodName Method name to invoke.
+     * @return Asynchronous result with the Actor's response.
+     */
+    Mono<String> invokeActorMethod(String methodName);
+
+    /**
+     * Invokes an Actor method on Dapr.
+     *
+     * @param methodName Method name to invoke.
+     * @param data Object with the data.
+     * @return Asynchronous result with the Actor's response.
+     */
+    Mono<String> invokeActorMethod(String methodName, Object data);
+
+}

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

@@ -0,0 +1,86 @@
+package io.dapr.actors.client;
+
+import io.dapr.actors.ActorId;
+import io.dapr.actors.runtime.ActorStateSerializer;
+import io.dapr.actors.utils.ObjectSerializer;
+
+/**
+ * Builder to generate an ActorProxy instance.
+ */
+public class ActorProxyBuilder {
+
+  /**
+   * Serializer for content to be sent back and forth between actors.
+   */
+  private static final ObjectSerializer SERIALIZER = new ActorStateSerializer();
+
+  /**
+   * Builder for the Dapr client.
+   */
+  private final ActorProxyClientBuilder clientBuilder = new ActorProxyClientBuilder();
+
+  /**
+   * Actor's type.
+   */
+  private String actorType;
+
+  /**
+   * Actor's identifier.
+   */
+  private ActorId actorId;
+
+  /**
+   * Changes build config to use specific port.
+   *
+   * @param port Port to be used.
+   * @return Same builder object.
+   */
+  public ActorProxyBuilder withPort(int port) {
+    this.clientBuilder.withPort(port);
+    return this;
+  }
+
+  /**
+   * Changes build config to use given Actor's type.
+   *
+   * @param actorType Actor's type.
+   * @return Same builder object.
+   */
+  public ActorProxyBuilder withActorType(String actorType) {
+    this.actorType = actorType;
+    return this;
+  }
+
+  /**
+   * 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;
+  }
+
+  /**
+   * Instantiates a new ActorProxy.
+   *
+   * @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) {
+      throw new IllegalArgumentException("Cannot instantiate an Actor without Id.");
+    }
+
+    return new ActorProxyImpl(
+      this.actorType,
+      this.actorId,
+      SERIALIZER,
+      this.clientBuilder.buildAsyncClient());
+  }
+
+}