Adding support for PubSub. Still needs JavaDocs.

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

@@ -0,0 +1,50 @@
+/*
+ * Copyright (c) Microsoft Corporation.
+ * Licensed under the MIT License.
+ */
+
+package io.dapr.examples.pubsub.http;
+
+import io.dapr.client.DaprClient;
+import io.dapr.client.DaprClientBuilder;
+
+import java.util.Collections;
+
+/**
+ * Message publisher.
+ * 1. Build and install jars:
+ * mvn clean install
+ * 2. Run the program:
+ * dapr run --app-id publisher --port 3006 -- mvn exec:java -pl=examples -Dexec.mainClass=io.dapr.examples.pubsub.http.Publisher
+ */
+public class Publisher {
+
+  private static final int NUM_MESSAGES = 10;
+
+  private static final String TOPIC_NAME = "message";
+
+  public static void main(String[] args) throws Exception {
+    DaprClient client = new DaprClientBuilder().build();
+    for (int i = 0; i < NUM_MESSAGES; i++) {
+      String message = String.format("This is message #%d", i);
+      client.publishEvent(TOPIC_NAME, message).block();
+      System.out.println("Published message: " + message);
+
+      try {
+        Thread.sleep((long)(1000 * Math.random()));
+      } catch (InterruptedException e) {
+        e.printStackTrace();
+        Thread.currentThread().interrupt();
+        return;
+      }
+    }
+
+    client.publishEvent(
+        TOPIC_NAME,
+        new byte[] { 1 },
+        Collections.singletonMap("content-type", "application/octet-stream")).block();
+    System.out.println("Published one byte.");
+
+    System.out.println("Done.");
+  }
+}

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

@@ -0,0 +1,48 @@
+/*
+ * Copyright (c) Microsoft Corporation.
+ * Licensed under the MIT License.
+ */
+
+package io.dapr.examples.pubsub.http;
+
+import io.dapr.runtime.Dapr;
+import io.dapr.springboot.DaprApplication;
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.DefaultParser;
+import org.apache.commons.cli.Options;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import reactor.core.publisher.Mono;
+
+/**
+ * Service for subscriber.
+ * 1. Build and install jars:
+ * mvn clean install
+ * 2. Run the server:
+ * dapr run --app-id subscriber --app-port 3000 --port 3005 -- mvn exec:java -pl=examples -Dexec.mainClass=io.dapr.examples.pubsub.http.Subscriber -Dexec.args="-p 3000"
+ */
+@SpringBootApplication
+public class Subscriber {
+
+  public static void main(String[] args) throws Exception {
+    Options options = new Options();
+    options.addRequiredOption("p", "port", true, "Port Dapr will listen to.");
+
+    CommandLineParser parser = new DefaultParser();
+    CommandLine cmd = parser.parse(options, args);
+
+    // If port string is not valid, it will throw an exception.
+    int port = Integer.parseInt(cmd.getOptionValue("port"));
+
+    // Subscribe to topic.
+    Dapr.getInstance().subscribeToTopic("message", (id, dataType, data, metadata) -> Mono
+        .fromSupplier(() -> {
+          System.out.println("Subscriber got message (" + id + "): " + (data == null ? "" : new String(data)));
+          return Boolean.TRUE;
+        })
+        .then(Mono.empty()));
+
+    // Start Dapr's callback endpoint.
+    DaprApplication.start(port);
+  }
+}

sdk-springboot/src/main/java/io/dapr/springboot/DaprApplication.java

@@ -33,7 +33,7 @@ public static void start(int port) {
    * @param args Command line arguments.
    */
   public static void main(String[] args) {
-    SpringApplication.run(DaprApplication.class, args);
+    DaprApplication.start(3000);
   }
 
 }

sdk-springboot/src/main/java/io/dapr/springboot/DaprController.java

@@ -10,13 +10,15 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.dapr.actors.runtime.ActorRuntime;
+import io.dapr.runtime.Dapr;
 import org.springframework.web.bind.annotation.*;
 import reactor.core.publisher.Mono;
 
 import java.io.IOException;
 import java.io.StringWriter;
 import java.io.Writer;
 import java.nio.charset.StandardCharsets;
+import java.util.Map;
 
 /**
  * SpringBoot Controller to handle callback APIs for Dapr.
@@ -30,12 +32,12 @@ public class DaprController {
 
   private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
-  @RequestMapping("/")
+  @GetMapping("/")
   public String index() {
     return "Greetings from Dapr!";
   }
 
-  @RequestMapping("/dapr/config")
+  @GetMapping("/dapr/config")
   public String daprConfig() throws Exception {
     try (Writer writer = new StringWriter()) {
       JsonGenerator generator = JSON_FACTORY.createGenerator(writer);
@@ -45,30 +47,56 @@ public String daprConfig() throws Exception {
         generator.writeString(actorClass);
       }
       generator.writeEndArray();
-      generator.writeStringField("actorIdleTimeout", "10s");
-      generator.writeStringField("actorScanInterval", "1s");
-      generator.writeStringField("drainOngoingCallTimeout", "1s");
-      generator.writeBooleanField("drainBalancedActors", true);
+      // TODO: handle configuration.
       generator.writeEndObject();
       generator.close();
       writer.flush();
       return writer.toString();
     }
   }
 
-  @RequestMapping(method = RequestMethod.POST, path = "/actors/{type}/{id}")
+  @GetMapping("/dapr/subscribe")
+  public String daprSubscribe() throws Exception {
+    try (Writer writer = new StringWriter()) {
+      JsonGenerator generator = JSON_FACTORY.createGenerator(writer);
+      generator.writeStartArray();
+      for (String topic : Dapr.getInstance().getSubscribedTopics()) {
+        generator.writeString(topic);
+      }
+      generator.writeEndArray();
+      generator.close();
+      writer.flush();
+      return writer.toString();
+    }
+  }
+
+  @PostMapping(path = "/{name}")
+  public Mono<byte[]> invokeMethodOrTopic(@PathVariable("name") String name,
+                                          @RequestBody(required = false) byte[] body,
+                                          @RequestHeader Map<String, String> header) {
+    return Dapr.getInstance().handleInvocation(name, body, header);
+  }
+
+  @PutMapping(path = "/{name}")
+  public Mono<byte[]> invokeMethodOrTopicViaPut(@PathVariable("name") String name,
+                                          @RequestBody(required = false) byte[] body,
+                                          @RequestHeader Map<String, String> header) {
+    return Dapr.getInstance().handleInvocation(name, body, header);
+  }
+
+  @PostMapping(path = "/actors/{type}/{id}")
   public Mono<Void> activateActor(@PathVariable("type") String type,
                                   @PathVariable("id") String id) throws Exception {
     return ActorRuntime.getInstance().activate(type, id);
   }
 
-  @RequestMapping(method = RequestMethod.DELETE, path = "/actors/{type}/{id}")
+  @DeleteMapping(path = "/actors/{type}/{id}")
   public Mono<Void> deactivateActor(@PathVariable("type") String type,
                                     @PathVariable("id") String id) throws Exception {
     return ActorRuntime.getInstance().deactivate(type, id);
   }
 
-  @RequestMapping(method = RequestMethod.PUT, path = "/actors/{type}/{id}/method/{method}")
+  @PutMapping(path = "/actors/{type}/{id}/method/{method}")
   public Mono<String> invokeActorMethod(@PathVariable("type") String type,
                                         @PathVariable("id") String id,
                                         @PathVariable("method") String method,
@@ -81,14 +109,14 @@ public Mono<String> invokeActorMethod(@PathVariable("type") String type,
     }
   }
 
-  @RequestMapping(method = RequestMethod.PUT, path = "/actors/{type}/{id}/method/timer/{timer}")
+  @PutMapping(path = "/actors/{type}/{id}/method/timer/{timer}")
   public Mono<Void> invokeActorTimer(@PathVariable("type") String type,
                                      @PathVariable("id") String id,
                                      @PathVariable("timer") String timer) {
     return ActorRuntime.getInstance().invokeTimer(type, id, timer);
   }
 
-  @RequestMapping(method = RequestMethod.PUT, path = "/actors/{type}/{id}/method/remind/{reminder}")
+  @PutMapping(path = "/actors/{type}/{id}/method/remind/{reminder}")
   public Mono<Void> invokeActorReminder(@PathVariable("type") String type,
                                         @PathVariable("id") String id,
                                         @PathVariable("reminder") String reminder,

sdk/src/main/java/io/dapr/actors/runtime/AbstractActor.java

@@ -95,9 +95,9 @@ protected <T> Mono<Void> registerReminder(
             Duration dueTime,
             Duration period) {
         try {
-            String data = this.actorRuntimeContext.getActorSerializer().serialize(state);
+            String data = this.actorRuntimeContext.getActorSerializer().serializeString(state);
             ActorReminderParams params = new ActorReminderParams(data, dueTime, period);
-            String serialized = this.actorRuntimeContext.getActorSerializer().serialize(params);
+            String serialized = this.actorRuntimeContext.getActorSerializer().serializeString(params);
             return this.actorRuntimeContext.getDaprClient().registerActorReminder(
                     this.actorRuntimeContext.getActorTypeInformation().getName(),
                     this.id.toString(),
@@ -139,7 +139,7 @@ protected <T> Mono<Void> registerActorTimer(
 
         try {
             ActorTimer actorTimer = new ActorTimer(this, name, callback, state, dueTime, period);
-            String serializedTimer = this.actorRuntimeContext.getActorSerializer().serialize(actorTimer);
+            String serializedTimer = this.actorRuntimeContext.getActorSerializer().serializeString(actorTimer);
 
             this.timers.put(name, actorTimer);
             return this.actorRuntimeContext.getDaprClient().registerActorTimer(

sdk/src/main/java/io/dapr/actors/runtime/ActorManager.java

@@ -226,15 +226,15 @@ private Mono<String> invokeMethod(ActorId actorId, ActorMethodContext context, S
                 if (response instanceof Mono) {
                     return ((Mono<Object>) response).map(r -> {
                         try {
-                            return this.runtimeContext.getActorSerializer().serialize(r);
+                            return this.runtimeContext.getActorSerializer().serializeString(r);
                         } catch (IOException e) {
                             throw new RuntimeException(e);
                         }
                     });
                 }
 
                 // Method was not Mono, so we serialize response.
-                return Mono.just(this.runtimeContext.getActorSerializer().serialize(response));
+                return Mono.just(this.runtimeContext.getActorSerializer().serializeString(response));
             } catch (Exception e) {
                 return Mono.error(e);
             }

sdk/src/main/java/io/dapr/actors/runtime/ActorRuntime.java

@@ -103,10 +103,9 @@ public Collection<String> getRegisteredActorTypes() {
      *
      * @param clazz The type of actor.
      * @param <T>   Actor class type.
-     * @return Async void task.
      */
-    public <T extends AbstractActor> Mono<Void> registerActor(Class<T> clazz) {
-        return registerActor(clazz, null);
+    public <T extends AbstractActor> void registerActor(Class<T> clazz) {
+        registerActor(clazz, null);
     }
 
     /**
@@ -115,10 +114,9 @@ public <T extends AbstractActor> Mono<Void> registerActor(Class<T> clazz) {
      * @param clazz        The type of actor.
      * @param actorFactory An optional factory to create actors.
      * @param <T>          Actor class type.
-     * @return Async void task.
      * This can be used for dependency injection into actors.
      */
-    public <T extends AbstractActor> Mono<Void> registerActor(Class<T> clazz, ActorFactory<T> actorFactory) {
+    public <T extends AbstractActor> void registerActor(Class<T> clazz, ActorFactory<T> actorFactory) {
         ActorTypeInformation<T> actorTypeInfo = ActorTypeInformation.create(clazz);
 
         ActorFactory<T> actualActorFactory = actorFactory != null ? actorFactory : new DefaultActorFactory<T>();
@@ -133,7 +131,6 @@ public <T extends AbstractActor> Mono<Void> registerActor(Class<T> clazz, ActorF
 
         // Create ActorManagers, override existing entry if registered again.
         this.actorManagers.put(actorTypeInfo.getName(), new ActorManager<T>(context));
-        return Mono.empty();
     }
 
     /**

sdk/src/main/java/io/dapr/actors/runtime/ActorStateSerializer.java

@@ -23,7 +23,7 @@ public class ActorStateSerializer extends ObjectSerializer {
      * {@inheritDoc}
      */
     @Override
-    public <T> String serialize(T state) throws IOException {
+    public <T> String serializeString(T state) throws IOException {
         if (state == null) {
             return null;
         }
@@ -39,7 +39,7 @@ public <T> String serialize(T state) throws IOException {
         }
 
         // Is not an special case.
-        return super.serialize(state);
+        return super.serializeString(state);
     }
 
     /**
@@ -105,13 +105,13 @@ public <T> String wrapMethodRequest(final T request) throws IOException {
             return null;
         }
 
-        String json = this.serialize(request);
+        byte[] data = this.serialize(request);
 
         try (Writer writer = new StringWriter()) {
             JsonGenerator generator = JSON_FACTORY.createGenerator(writer);
             generator.writeStartObject();
-            if (json != null) {
-                generator.writeBinaryField("data", json.getBytes());
+            if (data != null) {
+                generator.writeBinaryField("data", data);
             }
             generator.writeEndObject();
             generator.close();
@@ -139,7 +139,7 @@ private String serialize(ActorTimer<?> timer) throws IOException {
             generator.writeStringField("period", DurationUtils.ConvertDurationToDaprFormat(timer.getPeriod()));
             generator.writeStringField("callback", timer.getCallback());
             if (timer.getState() != null) {
-                generator.writeStringField("data", this.serialize(timer.getState()));
+                generator.writeStringField("data", this.serializeString(timer.getState()));
             }
             generator.writeEndObject();
             generator.close();

sdk/src/main/java/io/dapr/actors/runtime/DaprStateAsyncProvider.java

@@ -112,7 +112,7 @@ Mono<Void> apply(String actorType, ActorId actorId, ActorStateChange... stateCha
                 generator.writeObjectFieldStart("request");
                 generator.writeStringField("key", stateChange.getStateName());
                 if ((stateChange.getChangeKind() == ActorStateChangeKind.UPDATE) || (stateChange.getChangeKind() == ActorStateChangeKind.ADD)) {
-                    generator.writeStringField("value", this.serializer.serialize(stateChange.getValue()));
+                    generator.writeStringField("value", this.serializer.serializeString(stateChange.getValue()));
                 }
                 // End request object.
                 generator.writeEndObject();