diff --git a/examples/src/main/java/io/dapr/examples/pubsub/http/Publisher.java b/examples/src/main/java/io/dapr/examples/pubsub/http/Publisher.java new file mode 100644 index 0000000000..56b2d4249e --- /dev/null +++ b/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."); + } +} diff --git a/examples/src/main/java/io/dapr/examples/pubsub/http/Subscriber.java b/examples/src/main/java/io/dapr/examples/pubsub/http/Subscriber.java new file mode 100644 index 0000000000..b0c8cdf054 --- /dev/null +++ b/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); + } +} diff --git a/sdk-springboot/src/main/java/io/dapr/springboot/DaprApplication.java b/sdk-springboot/src/main/java/io/dapr/springboot/DaprApplication.java index b8dd5fa27a..455a4032ee 100644 --- a/sdk-springboot/src/main/java/io/dapr/springboot/DaprApplication.java +++ b/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); } } diff --git a/sdk-springboot/src/main/java/io/dapr/springboot/DaprController.java b/sdk-springboot/src/main/java/io/dapr/springboot/DaprController.java index cb890b36de..3e73900e6a 100644 --- a/sdk-springboot/src/main/java/io/dapr/springboot/DaprController.java +++ b/sdk-springboot/src/main/java/io/dapr/springboot/DaprController.java @@ -10,6 +10,7 @@ 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; @@ -17,6 +18,7 @@ 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("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/dapr/config") + @GetMapping("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/dapr/config") public String daprConfig() throws Exception { try (Writer writer = new StringWriter()) { JsonGenerator generator = JSON_FACTORY.createGenerator(writer); @@ -45,10 +47,7 @@ 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(); @@ -56,19 +55,48 @@ public String daprConfig() throws Exception { } } - @RequestMapping(method = RequestMethod.POST, path = "/actors/{type}/{id}") + @GetMapping("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/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 invokeMethodOrTopic(@PathVariable("name") String name, + @RequestBody(required = false) byte[] body, + @RequestHeader Map header) { + return Dapr.getInstance().handleInvocation(name, body, header); + } + + @PutMapping(path = "/{name}") + public Mono invokeMethodOrTopicViaPut(@PathVariable("name") String name, + @RequestBody(required = false) byte[] body, + @RequestHeader Map header) { + return Dapr.getInstance().handleInvocation(name, body, header); + } + + @PostMapping(path = "/actors/{type}/{id}") public Mono 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 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 invokeActorMethod(@PathVariable("type") String type, @PathVariable("id") String id, @PathVariable("method") String method, @@ -81,14 +109,14 @@ public Mono 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 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 invokeActorReminder(@PathVariable("type") String type, @PathVariable("id") String id, @PathVariable("reminder") String reminder, diff --git a/sdk/src/main/java/io/dapr/actors/runtime/AbstractActor.java b/sdk/src/main/java/io/dapr/actors/runtime/AbstractActor.java index 095d3dc58e..4e894fcc40 100644 --- a/sdk/src/main/java/io/dapr/actors/runtime/AbstractActor.java +++ b/sdk/src/main/java/io/dapr/actors/runtime/AbstractActor.java @@ -95,9 +95,9 @@ protected Mono 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 Mono 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( diff --git a/sdk/src/main/java/io/dapr/actors/runtime/ActorManager.java b/sdk/src/main/java/io/dapr/actors/runtime/ActorManager.java index e07fcb6b05..b915a5d0cc 100644 --- a/sdk/src/main/java/io/dapr/actors/runtime/ActorManager.java +++ b/sdk/src/main/java/io/dapr/actors/runtime/ActorManager.java @@ -226,7 +226,7 @@ private Mono invokeMethod(ActorId actorId, ActorMethodContext context, S if (response instanceof Mono) { return ((Mono) response).map(r -> { try { - return this.runtimeContext.getActorSerializer().serialize(r); + return this.runtimeContext.getActorSerializer().serializeString(r); } catch (IOException e) { throw new RuntimeException(e); } @@ -234,7 +234,7 @@ private Mono invokeMethod(ActorId actorId, ActorMethodContext context, S } // 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); } diff --git a/sdk/src/main/java/io/dapr/actors/runtime/ActorRuntime.java b/sdk/src/main/java/io/dapr/actors/runtime/ActorRuntime.java index f193e65d1a..87ac7abe73 100644 --- a/sdk/src/main/java/io/dapr/actors/runtime/ActorRuntime.java +++ b/sdk/src/main/java/io/dapr/actors/runtime/ActorRuntime.java @@ -103,10 +103,9 @@ public Collection getRegisteredActorTypes() { * * @param clazz The type of actor. * @param Actor class type. - * @return Async void task. */ - public Mono registerActor(Class clazz) { - return registerActor(clazz, null); + public void registerActor(Class clazz) { + registerActor(clazz, null); } /** @@ -115,10 +114,9 @@ public Mono registerActor(Class clazz) { * @param clazz The type of actor. * @param actorFactory An optional factory to create actors. * @param Actor class type. - * @return Async void task. * This can be used for dependency injection into actors. */ - public Mono registerActor(Class clazz, ActorFactory actorFactory) { + public void registerActor(Class clazz, ActorFactory actorFactory) { ActorTypeInformation actorTypeInfo = ActorTypeInformation.create(clazz); ActorFactory actualActorFactory = actorFactory != null ? actorFactory : new DefaultActorFactory(); @@ -133,7 +131,6 @@ public Mono registerActor(Class clazz, ActorF // Create ActorManagers, override existing entry if registered again. this.actorManagers.put(actorTypeInfo.getName(), new ActorManager(context)); - return Mono.empty(); } /** diff --git a/sdk/src/main/java/io/dapr/actors/runtime/ActorStateSerializer.java b/sdk/src/main/java/io/dapr/actors/runtime/ActorStateSerializer.java index a5ec021793..7cf1657171 100644 --- a/sdk/src/main/java/io/dapr/actors/runtime/ActorStateSerializer.java +++ b/sdk/src/main/java/io/dapr/actors/runtime/ActorStateSerializer.java @@ -23,7 +23,7 @@ public class ActorStateSerializer extends ObjectSerializer { * {@inheritDoc} */ @Override - public String serialize(T state) throws IOException { + public String serializeString(T state) throws IOException { if (state == null) { return null; } @@ -39,7 +39,7 @@ public String serialize(T state) throws IOException { } // Is not an special case. - return super.serialize(state); + return super.serializeString(state); } /** @@ -105,13 +105,13 @@ public 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(); diff --git a/sdk/src/main/java/io/dapr/actors/runtime/DaprStateAsyncProvider.java b/sdk/src/main/java/io/dapr/actors/runtime/DaprStateAsyncProvider.java index 2d0a1c0a03..4965855124 100644 --- a/sdk/src/main/java/io/dapr/actors/runtime/DaprStateAsyncProvider.java +++ b/sdk/src/main/java/io/dapr/actors/runtime/DaprStateAsyncProvider.java @@ -112,7 +112,7 @@ Mono 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(); diff --git a/sdk/src/main/java/io/dapr/client/DaprClient.java b/sdk/src/main/java/io/dapr/client/DaprClient.java index a52a1ad599..afb243be00 100644 --- a/sdk/src/main/java/io/dapr/client/DaprClient.java +++ b/sdk/src/main/java/io/dapr/client/DaprClient.java @@ -9,6 +9,7 @@ import reactor.core.publisher.Mono; import java.util.List; +import java.util.Map; /** * Generic Client Adapter to be used regardless of the GRPC or the HTTP Client implementation required. @@ -19,15 +20,28 @@ public interface DaprClient { /** * Publish an event. + * * @param topic the topic where the event will be published * @param event the event to be published - * @param The type of event to be publishded. + * @param The type of event to be published. * @return a Mono plan of type Void */ Mono publishEvent(String topic, T event); + /** + * Publish an event. + * + * @param topic the topic where the event will be published + * @param event the event to be published + * @param metadata The metadata for the published event. + * @param The type of event to be published. + * @return a Mono plan of type Void + */ + Mono publishEvent(String topic, T event, Map metadata); + /** * Invoke a service + * * @param verb The Verb to be used for HTTP will be the HTTP Verb, for GRPC is just a metadata value. * @param appId The Application ID where the service is * @param method The actual Method to be call in the application. @@ -41,6 +55,7 @@ public interface DaprClient { /** * Invoke a service + * * @param verb The Verb to be used for HTTP will be the HTTP Verb, for GRPC is just a metadata value. * @param appId The Application ID where the service is * @param method The actual Method to be call in the application. @@ -52,6 +67,7 @@ public interface DaprClient { /** * Creating a Binding + * * @param name The name of the biding to call * @param request The request needed for the binding * @param The type of the request. @@ -62,11 +78,11 @@ public interface DaprClient { /** * Retrieve a State based on their key. * - * @param state The key of the State to be retrieved + * @param state The key of the State to be retrieved * @param stateOptions - * @param clazz the Type of State needed as return. - * @param the Type of the return - * @param The Type of the key of the State + * @param clazz the Type of State needed as return. + * @param the Type of the return + * @param The Type of the key of the State * @return A Mono Plan for the requested State */ Mono getState(StateKeyValue state, StateOptions stateOptions, Class clazz); @@ -74,20 +90,21 @@ public interface DaprClient { /** * Save/Update a list of states. * - * @param states the States to be saved + * @param states the States to be saved * @param options the Options to use for each state - * @param the Type of the State + * @param the Type of the State * @return a Mono plan of type Void */ Mono saveStates(List> states, StateOptions options); /** * Save/Update a state - * @param key the key of the state - * @param etag the etag to be used - * @param value the value of the state + * + * @param key the key of the state + * @param etag the etag to be used + * @param value the value of the state * @param options the Options to use for each state - * @param the Type of the State + * @param the Type of the State * @return a Mono plan of type Void */ Mono saveState(String key, String etag, T value, StateOptions options); @@ -95,9 +112,9 @@ public interface DaprClient { /** * Delete a state * - * @param state The key of the State to be removed + * @param state The key of the State to be removed * @param options The options of the state - * @param The Type of the key of the State + * @param The Type of the key of the State * @return a Mono plan of type Void */ Mono deleteState(StateKeyValue state, StateOptions options); diff --git a/sdk/src/main/java/io/dapr/client/DaprClientBuilder.java b/sdk/src/main/java/io/dapr/client/DaprClientBuilder.java index a2e71e5a1f..5a1256adec 100644 --- a/sdk/src/main/java/io/dapr/client/DaprClientBuilder.java +++ b/sdk/src/main/java/io/dapr/client/DaprClientBuilder.java @@ -22,16 +22,9 @@ public class DaprClientBuilder { private static final int port = DaprClientBuilder.getEnvPortOrDefault(); /** - * Default host for Dapr after checking environment variable. + * Unique instance of httpClient to be shared. */ - private static final String host = Constants.DEFAULT_HOSTNAME; - - /** - * The HTTP Client that will be used to injectto connect to Dapr - */ - private static OkHttpClient okHttpClient = new OkHttpClient.Builder().build(); - - private static DaprClientHttpAdapter daprHttClient; + private static volatile DaprClientHttpAdapter daprHttClient; /** * Tries to get a valid port from environment variable or returns default. @@ -70,13 +63,10 @@ public DaprClient build() { * @throws java.lang.IllegalStateException if either host is missing or if port is missing or a negative number. */ private DaprClient buildDaprClientGrpc() { - if (null == this.host || "".equals(this.host.trim())) { - throw new IllegalStateException("Host must is required."); - } if (port <= 0) { throw new IllegalStateException("Invalid port."); } - ManagedChannel channel = ManagedChannelBuilder.forAddress(host, port).usePlaintext().build(); + ManagedChannel channel = ManagedChannelBuilder.forAddress(Constants.DEFAULT_HOSTNAME, port).usePlaintext().build(); return new DaprClientGrpcAdapter(DaprGrpc.newFutureStub(channel)); } @@ -87,21 +77,19 @@ private DaprClient buildDaprClientGrpc() { * @return */ private DaprClient buildDaprClientHttp() { - if (null == this.host || "".equals(this.host.trim())) { - throw new IllegalStateException("Host must is required."); - } if (port <= 0) { throw new IllegalStateException("Invalid port."); } - if (daprHttClient == null) { - synchronized (okHttpClient) { - if (daprHttClient == null) { - DaprHttp daprHtt = new DaprHttp(Constants.DEFAULT_HTTP_PROTOCOL_IDENTIFIED+host, port, okHttpClient); - daprHttClient = new DaprClientHttpAdapter(daprHtt); + if (this.daprHttClient == null) { + synchronized (DaprClientBuilder.class) { + if (this.daprHttClient == null) { + OkHttpClient okHttpClient = new OkHttpClient.Builder().build(); + DaprHttp daprHtt = new DaprHttp(Constants.DEFAULT_BASE_HTTP_URL, port, okHttpClient); + this.daprHttClient = new DaprClientHttpAdapter(daprHtt); } } } - return daprHttClient; + return this.daprHttClient; } } diff --git a/sdk/src/main/java/io/dapr/client/DaprClientGrpcAdapter.java b/sdk/src/main/java/io/dapr/client/DaprClientGrpcAdapter.java index 92f0db4f0c..b919f36931 100644 --- a/sdk/src/main/java/io/dapr/client/DaprClientGrpcAdapter.java +++ b/sdk/src/main/java/io/dapr/client/DaprClientGrpcAdapter.java @@ -51,13 +51,22 @@ class DaprClientGrpcAdapter implements DaprClient { */ @Override public Mono publishEvent(String topic, T event) { + return this.publishEvent(topic, event, null); + } + + /** + * {@inheritDoc} + */ + @Override + public Mono publishEvent(String topic, T event, Map metadata) { try { - String serializedEvent = objectSerializer.serialize(event); + String serializedEvent = objectSerializer.serializeString(event); Map mapEvent = new HashMap<>(); mapEvent.put("Topic", topic); mapEvent.put("Data", serializedEvent); + // TODO: handle metadata. - byte[] byteEvent = objectSerializer.serialize(mapEvent).getBytes(); + byte[] byteEvent = objectSerializer.serialize(mapEvent); DaprProtos.PublishEventEnvelope envelope = DaprProtos.PublishEventEnvelope.parseFrom(byteEvent); ListenableFuture futureEmpty = client.publishEvent(envelope); @@ -81,12 +90,12 @@ public Mono publishEvent(String topic, T event) { public Mono invokeService(String verb, String appId, String method, K request, Class clazz) { try { Map mapMessage = new HashMap<>(); - mapMessage.put("Id", objectSerializer.serialize(appId)); - mapMessage.put("Method", objectSerializer.serialize(verb)); - mapMessage.put("Data", objectSerializer.serialize(request)); + mapMessage.put("Id", appId); + mapMessage.put("Method", verb); + mapMessage.put("Data", objectSerializer.serializeString(request)); DaprProtos.InvokeServiceEnvelope envelope = - DaprProtos.InvokeServiceEnvelope.parseFrom(objectSerializer.serialize(mapMessage).getBytes()); + DaprProtos.InvokeServiceEnvelope.parseFrom(objectSerializer.serialize(mapMessage)); ListenableFuture futureResponse = client.invokeService(envelope); return Mono.just(futureResponse).flatMap(f -> { @@ -118,9 +127,9 @@ public Mono invokeBinding(String name, T request) { try { Map mapMessage = new HashMap<>(); mapMessage.put("Name", name); - mapMessage.put("Data", objectSerializer.serialize(request)); + mapMessage.put("Data", objectSerializer.serializeString(request)); DaprProtos.InvokeBindingEnvelope envelope = - DaprProtos.InvokeBindingEnvelope.parseFrom(objectSerializer.serialize(mapMessage).getBytes()); + DaprProtos.InvokeBindingEnvelope.parseFrom(objectSerializer.serialize(mapMessage)); ListenableFuture futureEmpty = client.invokeBinding(envelope); return Mono.just(futureEmpty).flatMap(f -> { try { @@ -144,8 +153,8 @@ public Mono getState(StateKeyValue key, StateOptions stateOptions, C Map request = new HashMap<>(); request.put("Key", key.getKey()); request.put("Consistency", stateOptions.getConsistency()); - String serializedRequest = objectSerializer.serialize(request); - DaprProtos.GetStateEnvelope envelope = DaprProtos.GetStateEnvelope.parseFrom(serializedRequest.getBytes()); + byte[] serializedRequest = objectSerializer.serialize(request); + DaprProtos.GetStateEnvelope envelope = DaprProtos.GetStateEnvelope.parseFrom(serializedRequest); ListenableFuture futureResponse = client.getState(envelope); return Mono.just(futureResponse).flatMap(f -> { try { @@ -173,7 +182,7 @@ public Mono saveStates(List> states, StateOptions opt }; Map mapStates = new HashMap<>(); mapStates.put("Requests", listStates); - byte[] byteRequests = objectSerializer.serialize(mapStates).getBytes(); + byte[] byteRequests = objectSerializer.serialize(mapStates); DaprProtos.SaveStateEnvelope envelope = DaprProtos.SaveStateEnvelope.parseFrom(byteRequests); ListenableFuture futureEmpty = client.saveState(envelope); return Mono.just(futureEmpty).flatMap(f -> { @@ -204,8 +213,8 @@ public Mono deleteState(StateKeyValue state, StateOptions options) try { Map mapOptions = transformStateOptionsToMap(options); Map mapState = transformStateKeyValueToMap(state, mapOptions); - String serializedState = objectSerializer.serialize(mapState); - DaprProtos.DeleteStateEnvelope envelope = DaprProtos.DeleteStateEnvelope.parseFrom(serializedState.getBytes()); + byte[] serializedState = objectSerializer.serialize(mapState); + DaprProtos.DeleteStateEnvelope envelope = DaprProtos.DeleteStateEnvelope.parseFrom(serializedState); ListenableFuture futureEmpty = client.deleteState(envelope); return Mono.just(futureEmpty).flatMap(f -> { try { diff --git a/sdk/src/main/java/io/dapr/client/DaprClientHttpAdapter.java b/sdk/src/main/java/io/dapr/client/DaprClientHttpAdapter.java index 41227c8d8f..643deace1d 100644 --- a/sdk/src/main/java/io/dapr/client/DaprClientHttpAdapter.java +++ b/sdk/src/main/java/io/dapr/client/DaprClientHttpAdapter.java @@ -28,22 +28,22 @@ public class DaprClientHttpAdapter implements DaprClient { * * @see io.dapr.client.DaprHttp */ - private DaprHttp client; + private final DaprHttp client; /** * A utitlity class for serialize and deserialize the messages sent and retrived by the client. */ - private ObjectSerializer objectSerializer; + private final ObjectSerializer objectSerializer; /** * Default access level constructor, in order to create an instance of this class use io.dapr.client.DaprClientBuilder * - * @param httpClient + * @param client Dapr's http client. * @see io.dapr.client.DaprClientBuilder */ - DaprClientHttpAdapter(DaprHttp httpClient) { + DaprClientHttpAdapter(DaprHttp client) { this.client = client; - objectSerializer = new ObjectSerializer(); + this.objectSerializer = new ObjectSerializer(); } /** @@ -51,22 +51,23 @@ public class DaprClientHttpAdapter implements DaprClient { */ @Override public Mono publishEvent(String topic, T event) { + return this.publishEvent(topic, event, null); + } + + /** + * {@inheritDoc} + */ + @Override + public Mono publishEvent(String topic, T event, Map metadata) { try { if (topic == null || topic.trim().isEmpty()) { - throw new DaprException("500", "Name cannot be null or empty."); + throw new DaprException("INVALID_TOPIC", "Topic name cannot be null or empty."); } - String serializedEvent = objectSerializer.serialize(event); + + byte[] serializedEvent = objectSerializer.serialize(event); StringBuilder url = new StringBuilder(Constants.PUBLISH_PATH).append("/").append(topic); - CompletableFuture futureVoid = client.invokeAPIVoid( - Constants.defaultHttpMethodSupported.POST.name(), url.toString(), serializedEvent, null); - return Mono.just(futureVoid).flatMap(f -> { - try { - f.get(); - } catch (Exception ex) { - return Mono.error(ex); - } - return Mono.empty(); - }); + return this.client.invokeAPI( + Constants.defaultHttpMethodSupported.POST.name(), url.toString(), serializedEvent, metadata).then(); } catch (Exception ex) { return Mono.error(ex); } @@ -91,20 +92,16 @@ public Mono invokeService(String verb, String appId, String method, K if (method == null || method.trim().isEmpty()) { throw new DaprException("500", "App Id cannot be null or empty."); } - StringBuilder urlSB = new StringBuilder("/invoke/"); - urlSB.append(objectSerializer.serialize(appId)); - urlSB.append("/method/"); - urlSB.append(objectSerializer.serialize(method)); - String serializedRequestBody = objectSerializer.serialize(request); - CompletableFuture futureResponse = - client.invokeAPI(httMethod.name(), urlSB.toString(), serializedRequestBody, null); - return Mono.just(futureResponse).flatMap(f -> { - try { - return Mono.just(objectSerializer.deserialize(f.get(), clazz)); - } catch (Exception ex) { - return Mono.error(ex); - } - }); + String path = String.format("/invoke/%s/method/%s", appId, method); + byte[] serializedRequestBody = objectSerializer.serialize(request); + return this.client.invokeAPI(httMethod.name(), path, serializedRequestBody, null) + .flatMap(r -> { + try { + return Mono.just(objectSerializer.deserialize(r, clazz)); + } catch (Exception ex) { + return Mono.error(ex); + } + }); } catch (Exception ex) { return Mono.error(ex); } @@ -129,21 +126,9 @@ public Mono invokeService(String verb, String appId, String method, T if (method == null || method.trim().isEmpty()) { throw new DaprException("500", "Method to invoke cannot be null or empty."); } - StringBuilder urlSB = new StringBuilder("/invoke/"); - urlSB.append(objectSerializer.serialize(appId)); - urlSB.append("/method/"); - urlSB.append(objectSerializer.serialize(method)); - String serializedRequestBody = objectSerializer.serialize(request); - CompletableFuture futureVoid = - client.invokeAPIVoid(httMethod.name(), urlSB.toString(), serializedRequestBody, null); - return Mono.just(futureVoid).flatMap(f -> { - try { - f.get(); - } catch (Exception ex) { - return Mono.error(ex); - } - return Mono.empty(); - }); + String path = String.format("/invoke/%s/method/%s", appId, method); + byte[] serializedRequestBody = objectSerializer.serialize(request); + return this.client.invokeAPI(httMethod.name(), path, serializedRequestBody, null).then(); } catch (Exception ex) { return Mono.error(ex); } @@ -159,21 +144,18 @@ public Mono invokeBinding(String name, T request) { throw new DaprException("500", "Name to bind cannot be null or empty."); } - String serializedBidingRequestBody = objectSerializer.serialize(request); + String serializedBidingRequestBody = objectSerializer.serializeString(request); Map jsonMap = new HashMap<>(); jsonMap.put("Data", serializedBidingRequestBody); StringBuilder url = new StringBuilder(Constants.BINDING_PATH).append("/").append(name); - CompletableFuture futureVoid = client.invokeAPIVoid( - Constants.defaultHttpMethodSupported.POST.name(), url.toString(), objectSerializer.serialize(jsonMap), null); - return Mono.just(futureVoid).flatMap(f -> { - try { - f.get(); - } catch (Exception ex) { - return Mono.error(ex); - } - return Mono.empty(); - }); + return this.client + .invokeAPI( + Constants.defaultHttpMethodSupported.POST.name(), + url.toString(), + objectSerializer.serialize(jsonMap), + null) + .then(); } catch (Exception ex) { return Mono.error(ex); } @@ -192,21 +174,20 @@ public Mono getState(StateKeyValue state, StateOptions options, Cla if (state.getEtag() != null && !state.getEtag().trim().isEmpty()) { headers.put(Constants.HEADER_HTTP_ETAG_ID, state.getEtag()); } - String serializedKeyBody = objectSerializer.serialize(state.getKey()); - serializedKeyBody += getOptionsAsQueryParameter(options); - if (options.getConsistency() != null && !options.getConsistency().trim().isEmpty()) { - serializedKeyBody += "?consistency=" + objectSerializer.serialize(options.getConsistency()); - } - StringBuilder url = new StringBuilder(Constants.STATE_PATH).append("/").append(serializedKeyBody); - CompletableFuture futureResponse = - client.invokeAPI(Constants.defaultHttpMethodSupported.GET.name(), url.toString(), null, headers); - return Mono.just(futureResponse).flatMap(f -> { - try { - return Mono.just(objectSerializer.deserialize(f.get(), clazz)); - } catch (Exception ex) { - return Mono.error(ex); - } - }); + + StringBuilder url = new StringBuilder(Constants.STATE_PATH) + .append("/") + .append(state.getKey()) + .append(getOptionsAsQueryParameter(options)); + return this.client + .invokeAPI(Constants.defaultHttpMethodSupported.GET.name(), url.toString(), headers) + .flatMap(s -> { + try { + return Mono.just(objectSerializer.deserialize(s, clazz)); + } catch (Exception ex) { + return Mono.error(ex); + } + }); } catch (Exception ex) { return Mono.error(ex); } @@ -228,17 +209,9 @@ public Mono saveStates(List> states, StateOptions opt headers.put(Constants.HEADER_HTTP_ETAG_ID, etag); } String url = Constants.STATE_PATH + getOptionsAsQueryParameter(options);; - String serializedStateBody = objectSerializer.serialize(states); - CompletableFuture futureVoid = client.invokeAPIVoid( - Constants.defaultHttpMethodSupported.POST.name(), url, serializedStateBody, headers); - return Mono.just(futureVoid).flatMap(f -> { - try { - f.get(); - } catch (Exception ex) { - return Mono.error(ex); - } - return Mono.empty(); - }); + byte[] serializedStateBody = objectSerializer.serialize(states); + return this.client.invokeAPI( + Constants.defaultHttpMethodSupported.POST.name(), url, serializedStateBody, headers).then(); } catch (Exception ex) { return Mono.error(ex); } @@ -263,19 +236,8 @@ public Mono deleteState(StateKeyValue state, StateOptions options) if (state.getEtag() != null && !state.getEtag().trim().isEmpty()) { headers.put(Constants.HEADER_HTTP_ETAG_ID, state.getEtag()); } - String serializedKey = objectSerializer.serialize(state.getKey()); - serializedKey += getOptionsAsQueryParameter(options); - String url = Constants.STATE_PATH + "/" + serializedKey; - CompletableFuture futureVoid = client.invokeAPIVoid( - Constants.defaultHttpMethodSupported.DELETE.name(), url, null, headers); - return Mono.just(futureVoid).flatMap(f -> { - try { - f.get(); - } catch (Exception ex) { - return Mono.error(ex); - } - return Mono.empty(); - }); + String url = Constants.STATE_PATH + "/" + state.getKey() + getOptionsAsQueryParameter(options); + return this.client.invokeAPI(Constants.defaultHttpMethodSupported.DELETE.name(), url, headers).then(); } catch (Exception ex) { return Mono.error(ex); } @@ -284,76 +246,43 @@ public Mono deleteState(StateKeyValue state, StateOptions options) @Override public Mono invokeActorMethod(String actorType, String actorId, String methodName, String jsonPayload) { String url = String.format(Constants.ACTOR_METHOD_RELATIVE_URL_FORMAT, actorType, actorId, methodName); - return actorActionString(Constants.defaultHttpMethodSupported.POST.name(), url, jsonPayload); + return this.client.invokeAPI(Constants.defaultHttpMethodSupported.POST.name(), url, jsonPayload, null); } @Override public Mono getActorState(String actorType, String actorId, String keyName) { String url = String.format(Constants.ACTOR_STATE_KEY_RELATIVE_URL_FORMAT, actorType, actorId, keyName); - return actorActionString(Constants.defaultHttpMethodSupported.GET.name(), url, null); + return this.client.invokeAPI(Constants.defaultHttpMethodSupported.GET.name(), url, "", null); } @Override public Mono saveActorStateTransactionally(String actorType, String actorId, String data) { String url = String.format(Constants.ACTOR_STATE_RELATIVE_URL_FORMAT, actorType, actorId); - return actorActionVoid(Constants.defaultHttpMethodSupported.PUT.name(), url, data); + return this.client.invokeAPI(Constants.defaultHttpMethodSupported.PUT.name(), url, data, null).then(); } @Override public Mono registerActorReminder(String actorType, String actorId, String reminderName, String data) { String url = String.format(Constants.ACTOR_REMINDER_RELATIVE_URL_FORMAT, actorType, actorId, reminderName); - return actorActionVoid(Constants.defaultHttpMethodSupported.PUT.name(), url, data); + return this.client.invokeAPI(Constants.defaultHttpMethodSupported.PUT.name(), url, data, null).then(); } @Override public Mono unregisterActorReminder(String actorType, String actorId, String reminderName) { String url = String.format(Constants.ACTOR_REMINDER_RELATIVE_URL_FORMAT, actorType, actorId, reminderName); - return actorActionVoid(Constants.defaultHttpMethodSupported.DELETE.name(), url, null); + return this.client.invokeAPI(Constants.defaultHttpMethodSupported.DELETE.name(), url, null).then(); } @Override public Mono registerActorTimer(String actorType, String actorId, String timerName, String data) { String url = String.format(Constants.ACTOR_TIMER_RELATIVE_URL_FORMAT, actorType, actorId, timerName); - return actorActionVoid(Constants.defaultHttpMethodSupported.PUT.name(), url, data); + return this.client.invokeAPI(Constants.defaultHttpMethodSupported.PUT.name(), url, data, null).then(); } @Override public Mono unregisterActorTimer(String actorType, String actorId, String timerName) { String url = String.format(Constants.ACTOR_TIMER_RELATIVE_URL_FORMAT, actorType, actorId, timerName); - return actorActionVoid(Constants.defaultHttpMethodSupported.DELETE.name(), url, null); - } - - private Mono actorActionString(String httpVerb, String url, String payload) { - try { - CompletableFuture futureResponse = - client.invokeAPI(httpVerb, url, objectSerializer.serialize(payload), null); - return Mono.just(futureResponse).flatMap(f -> { - try { - return Mono.just(objectSerializer.deserialize(f.get(), String.class)); - } catch (Exception ex) { - return Mono.error(ex); - } - }); - } catch (Exception ex) { - return Mono.error(ex); - } - } - - private Mono actorActionVoid(String httpVerb, String url, String payload) { - try { - CompletableFuture futureVoid = - client.invokeAPIVoid(httpVerb, url, objectSerializer.serialize(payload), null); - return Mono.just(futureVoid).flatMap(f -> { - try { - f.get(); - } catch (Exception ex) { - return Mono.error(ex); - } - return Mono.empty(); - }); - } catch (Exception ex) { - return Mono.error(ex); - } + return this.client.invokeAPI(Constants.defaultHttpMethodSupported.DELETE.name(), url, null).then(); } private String getOptionsAsQueryParameter(StateOptions options) diff --git a/sdk/src/main/java/io/dapr/client/DaprHttp.java b/sdk/src/main/java/io/dapr/client/DaprHttp.java index 1ac3c7beb8..613e007ccf 100644 --- a/sdk/src/main/java/io/dapr/client/DaprHttp.java +++ b/sdk/src/main/java/io/dapr/client/DaprHttp.java @@ -18,6 +18,7 @@ import java.io.IOException; import java.net.URL; +import java.nio.charset.StandardCharsets; import java.util.*; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; @@ -25,11 +26,6 @@ class DaprHttp { - /** - * ObjectMapper to Serialize data - */ - private static final ObjectSerializer MAPPER = new ObjectSerializer(); - /** * Defines the standard application/json type for HTTP calls in Dapr. */ @@ -42,6 +38,11 @@ class DaprHttp { private static final RequestBody REQUEST_BODY_EMPTY_JSON = RequestBody.Companion.create("", MEDIA_TYPE_APPLICATION_JSON); + /** + * Empty input or output. + */ + private static final byte[] EMPTY_BYTES = new byte[0]; + /** * JSON Object Mapper. */ @@ -65,9 +66,9 @@ class DaprHttp { /** * Creates a new instance of {@link DaprHttp}. * - * @param baseUrl Base url calling Dapr (e.g. http://localhost) - * @param port Port for calling Dapr. (e.g. 3500) - * @param httpClient RestClient used for all API calls in this new instance. + * @param baseUrl Base url calling Dapr (e.g. http://localhost) + * @param port Port for calling Dapr. (e.g. 3500) + * @param httpClient RestClient used for all API calls in this new instance. */ DaprHttp(String baseUrl, int port, OkHttpClient httpClient) { this.baseUrl = String.format("%s:%d/", baseUrl, port); @@ -76,16 +77,14 @@ class DaprHttp { } /** - * Invokes an API asynchronously that returns Void. + * Invokes an API asynchronously without payload that returns a text payload. * * @param method HTTP method. * @param urlString url as String. - * @param json JSON payload or null. - * @return Asynchronous Void + * @return Asynchronous text */ - protected final CompletableFuture invokeAPIVoid(String method, String urlString, String json, Map headers) { - CompletableFuture future = this.invokeAPI(method, urlString, json, headers); - return future.thenAcceptAsync(future::complete); + public final Mono invokeAPI(String method, String urlString, Map headers) { + return this.invokeAPI(method, urlString, (byte[])null, headers); } /** @@ -93,24 +92,47 @@ protected final CompletableFuture invokeAPIVoid(String method, String urlS * * @param method HTTP method. * @param urlString url as String. - * @param json JSON payload or null. + * @param content payload to be posted. * @return Asynchronous text */ - public final CompletableFuture invokeAPI(String method, String urlString, String json, Map headers) { - CompletableFuture future = CompletableFuture.supplyAsync( + public final Mono invokeAPI(String method, String urlString, String content, Map headers) { + return this.invokeAPI(method, urlString, content == null ? EMPTY_BYTES : content.getBytes(StandardCharsets.UTF_8), headers); + } + + /** + * Invokes an API asynchronously that returns a text payload. + * + * @param method HTTP method. + * @param urlString url as String. + * @param content payload to be posted. + * @return Asynchronous text + */ + public final Mono invokeAPI(String method, String urlString, byte[] content, Map headers) { + return Mono.fromFuture(CompletableFuture.supplyAsync( () -> { try { String requestId = UUID.randomUUID().toString(); - RequestBody body = - json != null ? RequestBody.Companion.create(json, MEDIA_TYPE_APPLICATION_JSON) : REQUEST_BODY_EMPTY_JSON; + RequestBody body = REQUEST_BODY_EMPTY_JSON; + + String contentType = headers != null ? headers.get("content-type") : null; + MediaType mediaType = contentType == null ? MEDIA_TYPE_APPLICATION_JSON : MediaType.get(contentType); + if (content == null) { + body = mediaType.equals(MEDIA_TYPE_APPLICATION_JSON) ? + REQUEST_BODY_EMPTY_JSON : RequestBody.Companion.create(new byte[0], mediaType); + } else { + body = RequestBody.Companion.create(content, mediaType); + } Request.Builder requestBuilder = new Request.Builder() .url(new URL(this.baseUrl + urlString)) .method(method, body) .addHeader(Constants.HEADER_DAPR_REQUEST_ID, requestId); - Optional.ofNullable(headers.entrySet()).orElse(Collections.emptySet()).stream().forEach(header ->{ - requestBuilder.addHeader(header.getKey(), header.getValue()); - }); + if (headers != null) { + Optional.ofNullable(headers.entrySet()).orElse(Collections.emptySet()).stream() + .forEach(header -> { + requestBuilder.addHeader(header.getKey(), header.getValue()); + }); + } Request request = requestBuilder.build(); @@ -129,9 +151,7 @@ public final CompletableFuture invokeAPI(String method, String urlString } catch (Exception e) { throw new RuntimeException(e); } - }, this.pool); - - return future; + }, this.pool)); } /** diff --git a/sdk/src/main/java/io/dapr/runtime/Dapr.java b/sdk/src/main/java/io/dapr/runtime/Dapr.java new file mode 100644 index 0000000000..bf0ee809e7 --- /dev/null +++ b/sdk/src/main/java/io/dapr/runtime/Dapr.java @@ -0,0 +1,150 @@ +/* + * Copyright (c) Microsoft Corporation. + * Licensed under the MIT License. + */ + +package io.dapr.runtime; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.dapr.exceptions.DaprException; +import io.dapr.utils.ObjectSerializer; +import reactor.core.publisher.Mono; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.*; +import java.util.function.Function; +import java.util.stream.Collectors; + +public final class Dapr implements DaprRuntime { + + /** + * Shared Json serializer/deserializer as per Jackson's documentation. + */ + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + + private static final Mono EMPTY_BYTES_ASYNC = Mono.just(new byte[0]); + + private static final byte[] EMPTY_BYTES = new byte[0]; + + private static volatile DaprRuntime instance; + + private final ObjectSerializer serializer = new ObjectSerializer(); + + private final Map>> handlers = Collections.synchronizedMap(new HashMap<>()); + + private Dapr() { + } + + /** + * Returns an DaprRuntime object. + * + * @return DaprRuntime object. + */ + public static DaprRuntime getInstance() { + if (instance == null) { + synchronized (Dapr.class) { + if (instance == null) { + instance = new Dapr(); + } + } + } + + return instance; + } + + @Override + public Collection getSubscribedTopics() { + return this.handlers + .entrySet() + .stream().filter(kv -> kv.getValue() instanceof TopicHandler) + .map(kv -> kv.getKey()) + .collect(Collectors.toCollection(ArrayList::new)); + } + + @Override + public void subscribeToTopic(String topic, TopicListener listener) { + this.handlers.putIfAbsent(topic, new TopicHandler(listener)); + } + + @Override + public void registerServiceMethod(String name, MethodListener listener) { + this.handlers.putIfAbsent(name, new MethodHandler(listener)); + } + + @Override + public Mono handleInvocation(String name, byte[] payload, Map metadata) { + Function> handler = this.handlers.get(name); + + if (handler == null) { + return Mono.error(new DaprException("INVALID_METHOD_OR_TOPIC", "Did not find handler for : " + (name == null ? "" : name))); + } + + try { + Map map = parse(payload); + String messageId = map.getOrDefault("id", "").toString(); + String dataType = map.getOrDefault("datacontenttype", "").toString(); + byte[] data = this.serializer.deserialize(map.get("data"), byte[].class); + return handler.apply(new HandleRequest(messageId, dataType, data, metadata)).switchIfEmpty(EMPTY_BYTES_ASYNC); + } catch (Exception e) { + // Handling exception in user code by propagating up via Mono. + return Mono.error(e); + } + } + + private static final class HandleRequest { + + private final String messageId; + + private final String dataType; + + private final byte[] payload; + + private final Map metadata; + + public HandleRequest(String messageId, String dataType, byte[] payload, Map metadata) { + this.messageId = messageId; + this.dataType = dataType; + this.payload = payload; + this.metadata = Collections.unmodifiableMap(metadata); + } + } + + private static final class MethodHandler implements Function> { + + private final MethodListener listener; + + private MethodHandler(MethodListener listener) { + this.listener = listener; + } + + @Override + public Mono apply(HandleRequest r) { + return listener.process(r.payload, r.metadata); + } + } + + private static final class TopicHandler implements Function> { + + private final TopicListener listener; + + private TopicHandler(TopicListener listener) { + this.listener = listener; + } + + @Override + public Mono apply(HandleRequest r) { + return listener.process(r.messageId, r.dataType, r.payload, r.metadata).then(Mono.empty()); + } + } + + public Map parse(final byte[] response) throws IOException { + if (response == null) { + return new HashMap(); + } + + return OBJECT_MAPPER.readValue(response, new TypeReference>() {}); + } +} diff --git a/sdk/src/main/java/io/dapr/runtime/DaprRuntime.java b/sdk/src/main/java/io/dapr/runtime/DaprRuntime.java new file mode 100644 index 0000000000..d7303a6fe2 --- /dev/null +++ b/sdk/src/main/java/io/dapr/runtime/DaprRuntime.java @@ -0,0 +1,22 @@ +/* + * Copyright (c) Microsoft Corporation. + * Licensed under the MIT License. + */ + +package io.dapr.runtime; + +import reactor.core.publisher.Mono; + +import java.util.Collection; +import java.util.Map; + +public interface DaprRuntime { + + void subscribeToTopic(String topic, TopicListener listener); + + Collection getSubscribedTopics(); + + void registerServiceMethod(String name, MethodListener listener); + + Mono handleInvocation(String name, byte[] payload, Map metadata); +} diff --git a/sdk/src/main/java/io/dapr/runtime/MethodListener.java b/sdk/src/main/java/io/dapr/runtime/MethodListener.java new file mode 100644 index 0000000000..ff77937b24 --- /dev/null +++ b/sdk/src/main/java/io/dapr/runtime/MethodListener.java @@ -0,0 +1,16 @@ +/* + * Copyright (c) Microsoft Corporation. + * Licensed under the MIT License. + */ + +package io.dapr.runtime; + +import reactor.core.publisher.Mono; + +import java.util.Map; + +public interface MethodListener { + + Mono process(byte[] data, Map metadata); + +} diff --git a/sdk/src/main/java/io/dapr/runtime/TopicListener.java b/sdk/src/main/java/io/dapr/runtime/TopicListener.java new file mode 100644 index 0000000000..7ba34fc581 --- /dev/null +++ b/sdk/src/main/java/io/dapr/runtime/TopicListener.java @@ -0,0 +1,16 @@ +/* + * Copyright (c) Microsoft Corporation. + * Licensed under the MIT License. + */ + +package io.dapr.runtime; + +import reactor.core.publisher.Mono; + +import java.util.Map; + +public interface TopicListener { + + Mono process(String messageId, String dataType, byte[] data, Map metadata); + +} diff --git a/sdk/src/main/java/io/dapr/utils/Constants.java b/sdk/src/main/java/io/dapr/utils/Constants.java index 0bc429e030..f75b486774 100644 --- a/sdk/src/main/java/io/dapr/utils/Constants.java +++ b/sdk/src/main/java/io/dapr/utils/Constants.java @@ -20,9 +20,9 @@ public final class Constants { public static final String DEFAULT_HOSTNAME = "localhost"; /** - * Dapr's default http protocol. + * Dapr's default http base url. */ - public static final String DEFAULT_HTTP_PROTOCOL_IDENTIFIED = "http://"; + public static final String DEFAULT_BASE_HTTP_URL = "http://" + DEFAULT_HOSTNAME; /** * Dapr's default port. diff --git a/sdk/src/main/java/io/dapr/utils/ObjectSerializer.java b/sdk/src/main/java/io/dapr/utils/ObjectSerializer.java index ba2f31b88c..9ee36707c4 100644 --- a/sdk/src/main/java/io/dapr/utils/ObjectSerializer.java +++ b/sdk/src/main/java/io/dapr/utils/ObjectSerializer.java @@ -9,6 +9,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import java.io.IOException; +import java.nio.charset.StandardCharsets; /** * Serializes and deserializes an object. @@ -33,13 +34,42 @@ public class ObjectSerializer { * @return Array of bytes[] with the serialized content. * @throws IOException In case state cannot be serialized. */ - public String serialize(T state) throws IOException { + public byte[] serialize(T state) throws IOException { if (state == null) { return null; } + if (state instanceof byte[]) { + return (byte[])state; + } + if (state.getClass() == String.class) { - return state.toString(); + return ((String) state).getBytes(StandardCharsets.UTF_8); + } + + if (isPrimitiveOrEquivalent(state.getClass())) { + return state.toString().getBytes(StandardCharsets.UTF_8); + } + + // Not string, not primitive, so it is a complex type: we use JSON for that. + return OBJECT_MAPPER.writeValueAsBytes(state); + } + + /** + * Serializes a given state object into String. + * + * @param state State object to be serialized. + * @param Type of the state object. + * @return Array of bytes[] with the serialized content. + * @throws IOException In case state cannot be serialized. + */ + public String serializeString(T state) throws IOException { + if (state == null) { + return null; + } + + if (state.getClass() == String.class) { + return (String) state; } if (isPrimitiveOrEquivalent(state.getClass())) { @@ -60,10 +90,6 @@ public String serialize(T state) throws IOException { * @throws IOException In case value cannot be deserialized. */ public T deserialize(Object value, Class clazz) throws IOException { - if (clazz == String.class) { - return (T) value; - } - if (isPrimitiveOrEquivalent(clazz)) { return parse(value, clazz); } @@ -72,7 +98,21 @@ public T deserialize(Object value, Class clazz) throws IOException { return (T) null; } - // Not string, not primitive, so it is a complex type: we use JSON for that. + if (clazz == String.class) { + return (value instanceof byte[]) + ? (T) new String((byte[])value, StandardCharsets.UTF_8) : (T) value.toString(); + } + + if (clazz == byte[].class) { + if (value instanceof String) { + return (T) value.toString().getBytes(StandardCharsets.UTF_8); + } + + return (value instanceof byte[]) + ? (T) value : null; + } + + // Not string, not primitive, not byte[], so it is a complex type: we use JSON for that. if (value instanceof byte[]) { return OBJECT_MAPPER.readValue((byte[]) value, clazz); } @@ -134,13 +174,16 @@ private static T parse(Object value, Class clazz) { if (isDoubleOrPrimitive(clazz) && isDoubleOrPrimitive(value.getClass())) return (T) value; } - if (isBooleanOrPrimitive(clazz)) return (T) Boolean.valueOf(value.toString()); - if (isByteOrPrimitive(clazz)) return (T) Byte.valueOf(value.toString()); - if (isShortOrPrimitive(clazz)) return (T) Short.valueOf(value.toString()); - if (isIntegerOrPrimitive(clazz)) return (T) Integer.valueOf(value.toString()); - if (isLongOrPrimitive(clazz)) return (T) Long.valueOf(value.toString()); - if (isFloatOrPrimitive(clazz)) return (T) Float.valueOf(value.toString()); - if (isDoubleOrPrimitive(clazz)) return (T) Double.valueOf(value.toString()); + String valueString = (value instanceof byte[]) ? + new String((byte[])value, StandardCharsets.UTF_8) : value.toString(); + + if (isBooleanOrPrimitive(clazz)) return (T) Boolean.valueOf(valueString); + if (isByteOrPrimitive(clazz)) return (T) Byte.valueOf(valueString); + if (isShortOrPrimitive(clazz)) return (T) Short.valueOf(valueString); + if (isIntegerOrPrimitive(clazz)) return (T) Integer.valueOf(valueString); + if (isLongOrPrimitive(clazz)) return (T) Long.valueOf(valueString); + if (isFloatOrPrimitive(clazz)) return (T) Float.valueOf(valueString); + if (isDoubleOrPrimitive(clazz)) return (T) Double.valueOf(valueString); return null; } diff --git a/sdk/src/test/java/io/dapr/actors/runtime/ActorManagerTest.java b/sdk/src/test/java/io/dapr/actors/runtime/ActorManagerTest.java index c916d3b4c3..0366552864 100644 --- a/sdk/src/test/java/io/dapr/actors/runtime/ActorManagerTest.java +++ b/sdk/src/test/java/io/dapr/actors/runtime/ActorManagerTest.java @@ -177,7 +177,7 @@ public void activateInvokeTimerDeactivateThenInvokeTimer() { private String createReminderParams(String data) throws IOException { ActorReminderParams params = new ActorReminderParams(data, Duration.ofSeconds(1), Duration.ofSeconds(1)); - return this.context.getActorSerializer().serialize(params); + return this.context.getActorSerializer().serializeString(params); } private static ActorId newActorId() { diff --git a/sdk/src/test/java/io/dapr/actors/runtime/ActorReminderParamsTest.java b/sdk/src/test/java/io/dapr/actors/runtime/ActorReminderParamsTest.java index 6e61b7e669..2e30825a6a 100644 --- a/sdk/src/test/java/io/dapr/actors/runtime/ActorReminderParamsTest.java +++ b/sdk/src/test/java/io/dapr/actors/runtime/ActorReminderParamsTest.java @@ -30,7 +30,7 @@ public void noState() { ActorReminderParams original = new ActorReminderParams(null, Duration.ZERO.plusMinutes(2), Duration.ZERO.plusMinutes((5))); ActorReminderParams recreated = null; try { - String serialized = SERIALIZER.serialize(original); + String serialized = SERIALIZER.serializeString(original); recreated = SERIALIZER.deserialize(serialized, ActorReminderParams.class); } catch(Exception e) { @@ -48,7 +48,7 @@ public void withState() { ActorReminderParams original = new ActorReminderParams("maru", Duration.ZERO.plusMinutes(2), Duration.ZERO.plusMinutes((5))); ActorReminderParams recreated = null; try { - String serialized = SERIALIZER.serialize(original); + String serialized = SERIALIZER.serializeString(original); recreated = SERIALIZER.deserialize(serialized, ActorReminderParams.class); } catch(Exception e) { diff --git a/sdk/src/test/java/io/dapr/actors/runtime/ActorTimerTest.java b/sdk/src/test/java/io/dapr/actors/runtime/ActorTimerTest.java index e27baf27f9..f39f446b7a 100644 --- a/sdk/src/test/java/io/dapr/actors/runtime/ActorTimerTest.java +++ b/sdk/src/test/java/io/dapr/actors/runtime/ActorTimerTest.java @@ -28,7 +28,7 @@ public void serialize() throws IOException { null, dueTime, period); - String s = new ActorStateSerializer().serialize(timer); + String s = new ActorStateSerializer().serializeString(timer); String expected = "{\"period\":\"1h0m3s0ms\",\"dueTime\":\"0h7m17s0ms\", \"callback\": \"myfunction\"}"; // Deep comparison via JsonNode.equals method. @@ -53,7 +53,7 @@ public void serializeWithOneTimePeriod() throws IOException { null, dueTime, period); - String s = new ActorStateSerializer().serialize(timer); + String s = new ActorStateSerializer().serializeString(timer); // A negative period will be serialized to an empty string which is interpreted by Dapr to mean fire once only. String expected = "{\"period\":\"\",\"dueTime\":\"0h7m17s0ms\", \"callback\": \"myfunction\"}"; diff --git a/sdk/src/test/java/io/dapr/utils/ObjectSerializerTest.java b/sdk/src/test/java/io/dapr/utils/ObjectSerializerTest.java index 3344c4a827..2c5662f80c 100644 --- a/sdk/src/test/java/io/dapr/utils/ObjectSerializerTest.java +++ b/sdk/src/test/java/io/dapr/utils/ObjectSerializerTest.java @@ -185,7 +185,7 @@ public void serializeObjectTest() { ObjectSerializer serializer = new ObjectSerializer(); String serializedValue; try { - serializedValue = serializer.serialize(obj); + serializedValue = serializer.serializeString(obj); assertEquals("FOUND:[[" + serializedValue + "]] \n but was EXPECING: [[" + expectedResult + "]]", expectedResult, serializedValue); } catch (IOException exception) { fail(exception.getMessage()); @@ -197,7 +197,7 @@ public void serializeNullTest() { ObjectSerializer serializer = new ObjectSerializer(); String serializedValue; try { - serializedValue = serializer.serialize(null); + serializedValue = serializer.serializeString(null); assertNull("The expected result is null", serializedValue); } catch (IOException exception) { fail(exception.getMessage()); @@ -210,7 +210,7 @@ public void serializeStringTest() { ObjectSerializer serializer = new ObjectSerializer(); String serializedValue; try { - serializedValue = serializer.serialize(valueToSerialize); + serializedValue = serializer.serializeString(valueToSerialize); assertEquals(valueToSerialize, serializedValue); } catch (IOException exception) { fail(exception.getMessage()); @@ -224,7 +224,7 @@ public void serializeIntTest() { ObjectSerializer serializer = new ObjectSerializer(); String serializedValue; try { - serializedValue = serializer.serialize(valueToSerialize.intValue()); + serializedValue = serializer.serializeString(valueToSerialize.intValue()); assertEquals(expectedResult, serializedValue); } catch (IOException exception) { fail(exception.getMessage());