From b09f2071c3d47dc2e24403fbf463f0e4e277f0d2 Mon Sep 17 00:00:00 2001 From: Artur Souza Date: Wed, 8 Jan 2020 18:12:32 -0800 Subject: [PATCH] Unit tests for PubSub + JavaDocs. --- .../io/dapr/client/DaprClientGrpcAdapter.java | 50 ++++- .../io/dapr/client/DaprClientHttpAdapter.java | 77 +++++-- .../main/java/io/dapr/client/DaprHttp.java | 15 +- .../io/dapr/client/domain/StateOptions.java | 1 + sdk/src/main/java/io/dapr/runtime/Dapr.java | 184 +++++++++++++---- .../java/io/dapr/runtime/DaprRuntime.java | 24 +++ .../java/io/dapr/runtime/MethodListener.java | 9 + .../java/io/dapr/runtime/TopicListener.java | 11 + .../main/java/io/dapr/utils/Constants.java | 10 +- .../java/io/dapr/utils/ObjectSerializer.java | 100 ++++++++- .../io/dapr/client/DaprClientTestBuilder.java | 21 ++ .../java/io/dapr/client/DaprHttpStub.java | 48 +++++ .../java/io/dapr/runtime/DaprRuntimeTest.java | 192 ++++++++++++++++++ 13 files changed, 671 insertions(+), 71 deletions(-) create mode 100644 sdk/src/test/java/io/dapr/client/DaprClientTestBuilder.java create mode 100644 sdk/src/test/java/io/dapr/client/DaprHttpStub.java create mode 100644 sdk/src/test/java/io/dapr/runtime/DaprRuntimeTest.java diff --git a/sdk/src/main/java/io/dapr/client/DaprClientGrpcAdapter.java b/sdk/src/main/java/io/dapr/client/DaprClientGrpcAdapter.java index b919f36931..526f6850bf 100644 --- a/sdk/src/main/java/io/dapr/client/DaprClientGrpcAdapter.java +++ b/sdk/src/main/java/io/dapr/client/DaprClientGrpcAdapter.java @@ -113,6 +113,8 @@ public Mono invokeService(String verb, String appId, String method, K /** * Operation not supported for GRPC + * + * TODO: Implement this since this IS supported. * @throws UnsupportedOperationException every time is called. */ public Mono invokeService(String verb, String appId, String method, T request) { @@ -198,6 +200,9 @@ public Mono saveStates(List> states, StateOptions opt } } + /** + * {@inheritDoc} + */ @Override public Mono saveState(String key, String etag, T value, StateOptions options) { StateKeyValue state = new StateKeyValue<>(value, key, etag); @@ -238,38 +243,70 @@ public Mono invokeActorMethod(String actorType, String actorId, String m return Mono.error(new UnsupportedOperationException("Operation not supported for GRPC")); } + /** + * Operation not supported for GRPC + * @throws UnsupportedOperationException every time is called. + */ @Override public Mono getActorState(String actorType, String actorId, String keyName) { return Mono.error(new UnsupportedOperationException("Operation not supported for GRPC")); } + /** + * Operation not supported for GRPC + * @throws UnsupportedOperationException every time is called. + */ @Override public Mono saveActorStateTransactionally(String actorType, String actorId, String data) { return Mono.error(new UnsupportedOperationException("Operation not supported for GRPC")); } + /** + * Operation not supported for GRPC + * @throws UnsupportedOperationException every time is called. + */ @Override public Mono registerActorReminder(String actorType, String actorId, String reminderName, String data) { return Mono.error(new UnsupportedOperationException("Operation not supported for GRPC")); } + /** + * Operation not supported for GRPC + * @throws UnsupportedOperationException every time is called. + */ @Override public Mono unregisterActorReminder(String actorType, String actorId, String reminderName) { return Mono.error(new UnsupportedOperationException("Operation not supported for GRPC")); } + /** + * Operation not supported for GRPC + * @throws UnsupportedOperationException every time is called. + */ @Override public Mono registerActorTimer(String actorType, String actorId, String timerName, String data) { return Mono.error(new UnsupportedOperationException("Operation not supported for GRPC")); } + /** + * Operation not supported for GRPC + * @throws UnsupportedOperationException every time is called. + */ @Override public Mono unregisterActorTimer(String actorType, String actorId, String timerName) { return Mono.error(new UnsupportedOperationException("Operation not supported for GRPC")); } + /** + * Converts state options to map. + * + * TODO: Move this logic to StateOptions. + * @param options Instance to have is methods converted into map. + * @return Map for the state options. + * @throws IllegalAccessException Cannot extract params. + */ private Map transformStateOptionsToMap(StateOptions options) - throws IllegalAccessException, IllegalArgumentException { + throws IllegalAccessException { Map mapOptions = null; if (options != null) { mapOptions = new HashMap<>(); @@ -283,8 +320,17 @@ private Map transformStateOptionsToMap(StateOptions options) return mapOptions; } + /** + * Creates an map for the given key-value operation. + * + * // TODO: Move this logic into StateKeyValue. + * @param state Key value for the state change. + * @param mapOptions Options to be applied to this operation. + * @return Map for the key-value operation. + * @throws IllegalAccessException Cannot identify key-value attributes. + */ private Map transformStateKeyValueToMap(StateKeyValue state, Map mapOptions) - throws IllegalAccessException, IllegalArgumentException { + throws IllegalAccessException { Map mapState = new HashMap<>(); for (Field field : state.getClass().getFields()) { mapState.put(field.getName(), field.get(state)); diff --git a/sdk/src/main/java/io/dapr/client/DaprClientHttpAdapter.java b/sdk/src/main/java/io/dapr/client/DaprClientHttpAdapter.java index 643deace1d..80b8d99407 100644 --- a/sdk/src/main/java/io/dapr/client/DaprClientHttpAdapter.java +++ b/sdk/src/main/java/io/dapr/client/DaprClientHttpAdapter.java @@ -7,13 +7,11 @@ import io.dapr.utils.ObjectSerializer; import reactor.core.publisher.Mono; -import java.io.IOException; import java.lang.reflect.Field; import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.concurrent.CompletableFuture; /** * An adapter for the GRPC Client. @@ -67,7 +65,7 @@ public Mono publishEvent(String topic, T event, Map me byte[] serializedEvent = objectSerializer.serialize(event); StringBuilder url = new StringBuilder(Constants.PUBLISH_PATH).append("/").append(topic); return this.client.invokeAPI( - Constants.defaultHttpMethodSupported.POST.name(), url.toString(), serializedEvent, metadata).then(); + DaprHttp.HttpMethods.POST.name(), url.toString(), serializedEvent, metadata).then(); } catch (Exception ex) { return Mono.error(ex); } @@ -82,7 +80,7 @@ public Mono invokeService(String verb, String appId, String method, K if (verb == null || verb.trim().isEmpty()) { throw new DaprException("500", "App Id cannot be null or empty."); } - Constants.defaultHttpMethodSupported httMethod = Constants.defaultHttpMethodSupported.valueOf(verb.toUpperCase()); + DaprHttp.HttpMethods httMethod = DaprHttp.HttpMethods.valueOf(verb.toUpperCase()); if (httMethod == null) { throw new DaprException("405", "HTTP Method not allowed."); } @@ -116,7 +114,7 @@ public Mono invokeService(String verb, String appId, String method, T if (verb == null || verb.trim().isEmpty()) { throw new DaprException("500", "App Id cannot be null or empty."); } - Constants.defaultHttpMethodSupported httMethod = Constants.defaultHttpMethodSupported.valueOf(verb.toUpperCase()); + DaprHttp.HttpMethods httMethod = DaprHttp.HttpMethods.valueOf(verb.toUpperCase()); if (httMethod == null) { throw new DaprException("405", "HTTP Method not allowed."); } @@ -151,7 +149,7 @@ public Mono invokeBinding(String name, T request) { StringBuilder url = new StringBuilder(Constants.BINDING_PATH).append("/").append(name); return this.client .invokeAPI( - Constants.defaultHttpMethodSupported.POST.name(), + DaprHttp.HttpMethods.POST.name(), url.toString(), objectSerializer.serialize(jsonMap), null) @@ -180,7 +178,7 @@ public Mono getState(StateKeyValue state, StateOptions options, Cla .append(state.getKey()) .append(getOptionsAsQueryParameter(options)); return this.client - .invokeAPI(Constants.defaultHttpMethodSupported.GET.name(), url.toString(), headers) + .invokeAPI(DaprHttp.HttpMethods.GET.name(), url.toString(), headers) .flatMap(s -> { try { return Mono.just(objectSerializer.deserialize(s, clazz)); @@ -211,12 +209,15 @@ public Mono saveStates(List> states, StateOptions opt String url = Constants.STATE_PATH + getOptionsAsQueryParameter(options);; byte[] serializedStateBody = objectSerializer.serialize(states); return this.client.invokeAPI( - Constants.defaultHttpMethodSupported.POST.name(), url, serializedStateBody, headers).then(); + DaprHttp.HttpMethods.POST.name(), url, serializedStateBody, headers).then(); } catch (Exception ex) { return Mono.error(ex); } } + /** + * {@inheritDoc} + */ @Override public Mono saveState(String key, String etag, T value, StateOptions options) { StateKeyValue state = new StateKeyValue<>(value, key, etag); @@ -237,70 +238,108 @@ public Mono deleteState(StateKeyValue state, StateOptions options) headers.put(Constants.HEADER_HTTP_ETAG_ID, state.getEtag()); } String url = Constants.STATE_PATH + "/" + state.getKey() + getOptionsAsQueryParameter(options); - return this.client.invokeAPI(Constants.defaultHttpMethodSupported.DELETE.name(), url, headers).then(); + return this.client.invokeAPI(DaprHttp.HttpMethods.DELETE.name(), url, headers).then(); } catch (Exception ex) { return Mono.error(ex); } } + /** + * {@inheritDoc} + */ @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 this.client.invokeAPI(Constants.defaultHttpMethodSupported.POST.name(), url, jsonPayload, null); + return this.client.invokeAPI(DaprHttp.HttpMethods.POST.name(), url, jsonPayload, null); } + /** + * {@inheritDoc} + */ @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 this.client.invokeAPI(Constants.defaultHttpMethodSupported.GET.name(), url, "", null); + return this.client.invokeAPI(DaprHttp.HttpMethods.GET.name(), url, "", null); } + /** + * {@inheritDoc} + */ @Override public Mono saveActorStateTransactionally(String actorType, String actorId, String data) { String url = String.format(Constants.ACTOR_STATE_RELATIVE_URL_FORMAT, actorType, actorId); - return this.client.invokeAPI(Constants.defaultHttpMethodSupported.PUT.name(), url, data, null).then(); + return this.client.invokeAPI(DaprHttp.HttpMethods.PUT.name(), url, data, null).then(); } + /** + * {@inheritDoc} + */ @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 this.client.invokeAPI(Constants.defaultHttpMethodSupported.PUT.name(), url, data, null).then(); + return this.client.invokeAPI(DaprHttp.HttpMethods.PUT.name(), url, data, null).then(); } + /** + * {@inheritDoc} + */ @Override public Mono unregisterActorReminder(String actorType, String actorId, String reminderName) { String url = String.format(Constants.ACTOR_REMINDER_RELATIVE_URL_FORMAT, actorType, actorId, reminderName); - return this.client.invokeAPI(Constants.defaultHttpMethodSupported.DELETE.name(), url, null).then(); + return this.client.invokeAPI(DaprHttp.HttpMethods.DELETE.name(), url, null).then(); } + /** + * {@inheritDoc} + */ @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 this.client.invokeAPI(Constants.defaultHttpMethodSupported.PUT.name(), url, data, null).then(); + return this.client.invokeAPI(DaprHttp.HttpMethods.PUT.name(), url, data, null).then(); } + /** + * {@inheritDoc} + */ @Override public Mono unregisterActorTimer(String actorType, String actorId, String timerName) { String url = String.format(Constants.ACTOR_TIMER_RELATIVE_URL_FORMAT, actorType, actorId, timerName); - return this.client.invokeAPI(Constants.defaultHttpMethodSupported.DELETE.name(), url, null).then(); + return this.client.invokeAPI(DaprHttp.HttpMethods.DELETE.name(), url, null).then(); } + /** + * Gets the string with params for a given URL. + * + * TODO: Move this logic down the stack to use okhttp's builder instead: + * https://square.github.io/okhttp/4.x/okhttp/okhttp3/-http-url/-builder/add-query-parameter/ + * @param options State options to be converted. + * @return String with query params. + * @throws IllegalAccessException Cannot extract params. + */ private String getOptionsAsQueryParameter(StateOptions options) - throws IllegalAccessException, IllegalArgumentException, IOException { + throws IllegalAccessException { StringBuilder sb = new StringBuilder(); Map mapOptions = transformStateOptionsToMap(options); if (mapOptions != null && !mapOptions.isEmpty()) { sb.append("?"); for (Map.Entry option : mapOptions.entrySet()) { - sb.append(option.getKey()).append("=").append(objectSerializer.serialize(option.getValue())).append("&"); + sb.append(option.getKey()).append("=").append(option.getValue()).append("&"); } sb.deleteCharAt(sb.length()-1); } return sb.toString(); } + /** + * Converts state options to map. + * + * TODO: Move this logic to StateOptions. + * @param options Instance to have is methods converted into map. + * @return Map for the state options. + * @throws IllegalAccessException Cannot extract params. + */ private Map transformStateOptionsToMap(StateOptions options) - throws IllegalAccessException, IllegalArgumentException { + throws IllegalAccessException { Map mapOptions = null; if (options != null) { mapOptions = new HashMap<>(); diff --git a/sdk/src/main/java/io/dapr/client/DaprHttp.java b/sdk/src/main/java/io/dapr/client/DaprHttp.java index f0f03220af..6156fece2a 100644 --- a/sdk/src/main/java/io/dapr/client/DaprHttp.java +++ b/sdk/src/main/java/io/dapr/client/DaprHttp.java @@ -26,6 +26,11 @@ class DaprHttp { + /** + * HTTP Methods supported. + */ + enum HttpMethods { GET, PUT, POST, DELETE; } + /** * Defines the standard application/json type for HTTP calls in Dapr. */ @@ -83,7 +88,7 @@ class DaprHttp { * @param urlString url as String. * @return Asynchronous text */ - public final Mono invokeAPI(String method, String urlString, Map headers) { + public Mono invokeAPI(String method, String urlString, Map headers) { return this.invokeAPI(method, urlString, (byte[])null, headers); } @@ -95,7 +100,7 @@ public final Mono invokeAPI(String method, String urlString, Map invokeAPI(String method, String urlString, String content, Map headers) { + public 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); } @@ -107,7 +112,7 @@ public final Mono invokeAPI(String method, String urlString, String cont * @param content payload to be posted. * @return Asynchronous text */ - public final Mono invokeAPI(String method, String urlString, byte[] content, Map headers) { + public Mono invokeAPI(String method, String urlString, byte[] content, Map headers) { return Mono.fromFuture(CompletableFuture.supplyAsync( () -> { try { @@ -126,9 +131,9 @@ public final Mono invokeAPI(String method, String urlString, byte[] cont Request.Builder requestBuilder = new Request.Builder() .url(new URL(this.baseUrl + urlString)) .addHeader(Constants.HEADER_DAPR_REQUEST_ID, requestId); - if (Constants.defaultHttpMethodSupported.GET.name().equals(method)) { + if (HttpMethods.GET.name().equals(method)) { requestBuilder.get(); - } else if (Constants.defaultHttpMethodSupported.DELETE.name().equals(method)) { + } else if (HttpMethods.DELETE.name().equals(method)) { requestBuilder.delete(); } else { requestBuilder.method(method, body); diff --git a/sdk/src/main/java/io/dapr/client/domain/StateOptions.java b/sdk/src/main/java/io/dapr/client/domain/StateOptions.java index 24150e127a..6987dc05bc 100644 --- a/sdk/src/main/java/io/dapr/client/domain/StateOptions.java +++ b/sdk/src/main/java/io/dapr/client/domain/StateOptions.java @@ -1,6 +1,7 @@ package io.dapr.client.domain; public class StateOptions { + private final String consistency; public StateOptions(String consistency) { diff --git a/sdk/src/main/java/io/dapr/runtime/Dapr.java b/sdk/src/main/java/io/dapr/runtime/Dapr.java index bf0ee809e7..025d37467f 100644 --- a/sdk/src/main/java/io/dapr/runtime/Dapr.java +++ b/sdk/src/main/java/io/dapr/runtime/Dapr.java @@ -5,7 +5,9 @@ package io.dapr.runtime; +import com.fasterxml.jackson.annotation.JsonInclude; import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.DeserializationFeature; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import io.dapr.exceptions.DaprException; @@ -18,28 +20,41 @@ import java.util.function.Function; import java.util.stream.Collectors; +/** + * Main interface to register and interface with local Dapr instance. + * + * This class DO NOT make I/O operations by itself, only via user-provided listeners. + */ public final class Dapr implements DaprRuntime { /** - * Shared Json serializer/deserializer as per Jackson's documentation. + * Empty response. */ - 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]; - + /** + * Singleton instance for this class. + */ private static volatile DaprRuntime instance; + /** + * Serializes and deserializes internal objects. + */ private final ObjectSerializer serializer = new ObjectSerializer(); - private final Map>> handlers = Collections.synchronizedMap(new HashMap<>()); + /** + * Topics, methods and binding handles. + */ + private final Map>> handlers = Collections.synchronizedMap(new HashMap<>()); + /** + * Private constructor to keep it singleton. + */ private Dapr() { } /** - * Returns an DaprRuntime object. + * Returns the only DaprRuntime object. * * @return DaprRuntime object. */ @@ -55,6 +70,9 @@ public static DaprRuntime getInstance() { return instance; } + /** + * {@inheritDoc} + */ @Override public Collection getSubscribedTopics() { return this.handlers @@ -64,87 +82,179 @@ public Collection getSubscribedTopics() { .collect(Collectors.toCollection(ArrayList::new)); } + /** + * {@inheritDoc} + */ @Override public void subscribeToTopic(String topic, TopicListener listener) { this.handlers.putIfAbsent(topic, new TopicHandler(listener)); } + /** + * {@inheritDoc} + */ @Override public void registerServiceMethod(String name, MethodListener listener) { this.handlers.putIfAbsent(name, new MethodHandler(listener)); } + /** + * {@inheritDoc} + */ @Override public Mono handleInvocation(String name, byte[] payload, Map metadata) { - Function> handler = this.handlers.get(name); + 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); + Message message = this.serializer.deserialize(payload, Message.class); + if (message == null) { + return EMPTY_BYTES_ASYNC; + } + message.setMetadata(metadata); + return handler.apply(message).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 { + /** + * Internal class to encapsulate a request message. + */ + public static final class Message { + + /** + * Identifier of the message being processed. + */ + private String id; + + /** + * Type of the input payload. + */ + private String datacontenttype; + + /** + * Raw input payload. + */ + private byte[] data; + + /** + * Headers (or metadata). + */ + private Map metadata; + + /** + * Instantiates a new input request (useful for JSON deserialization). + */ + public Message() { + } + + /** + * Instantiates a new input request. + * @param id Identifier of the message being processed. + * @param datacontenttype Type of the input payload. + * @param data Input body. + * @param metadata Headers (or metadata) for the call. + */ + public Message(String id, String datacontenttype, byte[] data, Map metadata) { + this.id = id; + this.datacontenttype = datacontenttype; + this.data = data; + this.metadata = metadata == null ? null : Collections.unmodifiableMap(metadata); + } - private final String messageId; + public String getId() { + return id; + } - private final String dataType; + public void setId(String id) { + this.id = id; + } - private final byte[] payload; + public byte[] getData() { + return data; + } - private final Map metadata; + public void setData(byte[] data) { + this.data = data; + } + + public String getDatacontenttype() { + return datacontenttype; + } + + public void setDatacontenttype(String datacontenttype) { + this.datacontenttype = datacontenttype; + } - public HandleRequest(String messageId, String dataType, byte[] payload, Map metadata) { - this.messageId = messageId; - this.dataType = dataType; - this.payload = payload; - this.metadata = Collections.unmodifiableMap(metadata); + public Map getMetadata() { + return metadata; + } + + public void setMetadata(Map metadata) { + this.metadata = metadata; } } - private static final class MethodHandler implements Function> { + /** + * Internal class to handle a method call. + */ + private static final class MethodHandler implements Function> { + /** + * User-provided listener. + */ private final MethodListener listener; + /** + * Instantiates a new method handler. + * @param listener Method to be executed on a given API call. + */ private MethodHandler(MethodListener listener) { this.listener = listener; } + /** + * Executes the listener's method. + * @param r Internal request object. + * @return Raw output payload or empty. + */ @Override - public Mono apply(HandleRequest r) { - return listener.process(r.payload, r.metadata); + public Mono apply(Message r) { + return listener.process(r.data, r.metadata); } } - private static final class TopicHandler implements Function> { + /** + * Internal class to handle a topic message delivery. + */ + private static final class TopicHandler implements Function> { + /** + * User-provided listener. + */ private final TopicListener listener; + /** + * Instantiates a new topic handler. + * @param listener Callback to be executed on a given message. + */ private TopicHandler(TopicListener listener) { this.listener = listener; } + /** + * Executes the listener's method. + * @param r Internal request object. + * @return Always empty response. + */ @Override - public Mono apply(HandleRequest r) { - return listener.process(r.messageId, r.dataType, r.payload, r.metadata).then(Mono.empty()); + public Mono apply(Message r) { + return listener.process(r.id, r.datacontenttype, r.data, 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 index d7303a6fe2..829739e5fa 100644 --- a/sdk/src/main/java/io/dapr/runtime/DaprRuntime.java +++ b/sdk/src/main/java/io/dapr/runtime/DaprRuntime.java @@ -10,13 +10,37 @@ import java.util.Collection; import java.util.Map; +/** + * Common interface to configure and process callback API. + */ public interface DaprRuntime { + /** + * Subscribes to a topic. + * @param topic Topic name to be subscribed to. + * @param listener Callback to be executed on a given message. + */ void subscribeToTopic(String topic, TopicListener listener); + /** + * Returns the collection of subscribed topics. + * @return Collection of subscribed topics. + */ Collection getSubscribedTopics(); + /** + * Registers a service method to be executed on an API call. + * @param name Name of the service API's method. + * @param listener Method to be executed on a given API call. + */ void registerServiceMethod(String name, MethodListener listener); + /** + * Handles a given topic message or method API call. + * @param name Name of topic or method. + * @param payload Input body. + * @param metadata Headers (or metadata) for the call. + * @return Response payload or empty. + */ 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 index ff77937b24..4b1b793ca5 100644 --- a/sdk/src/main/java/io/dapr/runtime/MethodListener.java +++ b/sdk/src/main/java/io/dapr/runtime/MethodListener.java @@ -9,8 +9,17 @@ import java.util.Map; +/** + * Processes a given API's method call. + */ public interface MethodListener { + /** + * Processes a given API's method call. + * @param data Raw input payload. + * @param metadata Header (or metadata). + * @return Raw response or empty. + */ 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 index 7ba34fc581..8325ab8fe2 100644 --- a/sdk/src/main/java/io/dapr/runtime/TopicListener.java +++ b/sdk/src/main/java/io/dapr/runtime/TopicListener.java @@ -9,8 +9,19 @@ import java.util.Map; +/** + * Processes a given topic's message delivery. + */ public interface TopicListener { + /** + * Processes a given topic's message delivery. + * @param messageId Message's identifier. + * @param dataType Type of the input data. + * @param data Input data. + * @param metadata Headers (or metadata). + * @return Empty response. + */ 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 f75b486774..32aea9f048 100644 --- a/sdk/src/main/java/io/dapr/utils/Constants.java +++ b/sdk/src/main/java/io/dapr/utils/Constants.java @@ -29,13 +29,6 @@ public final class Constants { */ public static final int DEFAULT_PORT = 3500; - public static enum defaultHttpMethodSupported { - GET, - PUT, - POST, - DELETE; - } - /** * Environment variable used to set Dapr's port. */ @@ -46,6 +39,9 @@ public static enum defaultHttpMethodSupported { */ public static final String HEADER_DAPR_REQUEST_ID = "X-DaprRequestId"; + /** + * Header for the conditional operation. + */ public static final String HEADER_HTTP_ETAG_ID = "If-Match"; /** diff --git a/sdk/src/main/java/io/dapr/utils/ObjectSerializer.java b/sdk/src/main/java/io/dapr/utils/ObjectSerializer.java index 9ee36707c4..e35b0c30a8 100644 --- a/sdk/src/main/java/io/dapr/utils/ObjectSerializer.java +++ b/sdk/src/main/java/io/dapr/utils/ObjectSerializer.java @@ -4,12 +4,20 @@ */ package io.dapr.utils; +import com.fasterxml.jackson.annotation.JsonInclude; import com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.DeserializationFeature; +import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.MapperFeature; import com.fasterxml.jackson.databind.ObjectMapper; +import io.dapr.runtime.Dapr; import java.io.IOException; import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; /** * Serializes and deserializes an object. @@ -24,7 +32,14 @@ public class ObjectSerializer { /** * Shared Json serializer/deserializer as per Jackson's documentation. */ - protected static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + protected static final ObjectMapper OBJECT_MAPPER = new ObjectMapper() + .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false) + .setSerializationInclusion(JsonInclude.Include.NON_NULL); + + /** + * Constant result for empty maps. + */ + private static final Map EMPTY_MAP = Collections.unmodifiableMap(new HashMap<>()); /** * Serializes a given state object into byte array. @@ -98,6 +113,10 @@ public T deserialize(Object value, Class clazz) throws IOException { return (T) null; } + if (clazz == Dapr.Message.class) { + return (T) this.deserializeTopicMessage(value); + } + if (clazz == String.class) { return (value instanceof byte[]) ? (T) new String((byte[])value, StandardCharsets.UTF_8) : (T) value.toString(); @@ -120,6 +139,50 @@ public T deserialize(Object value, Class clazz) throws IOException { return OBJECT_MAPPER.readValue(value.toString(), clazz); } + /** + * Deserialized a message topic from Dapr. + * @param payload Payload sent from Dapr. + * @return Message (can be null if input is null) + * @throws IOException If cannot parse. + */ + private Dapr.Message deserializeTopicMessage(Object payload) throws IOException { + if (payload == null) { + return null; + } + + JsonNode node = null; + if (payload instanceof byte[]) { + node = OBJECT_MAPPER.readTree((byte[])payload); + } else { + node = OBJECT_MAPPER.readTree(payload.toString()); + } + + if (node== null) { + return null; + } + + String id = null; + if (node.has("id") && !node.get("id").isNull()) { + id = node.get("id").asText(); + } + + String dataType = null; + if (node.has("datacontenttype") && !node.get("datacontenttype").isNull()) { + dataType = node.get("datacontenttype").asText(); + } + + byte[] data = null; + if (node.has("data") && !node.get("data").isNull()) { + try { + data = node.get("data").binaryValue(); + } catch (IOException e) { + data = node.get("data").textValue().getBytes(StandardCharsets.UTF_8); + } + } + + return new Dapr.Message(id, dataType, data, null); + } + /** * Checks if the class is a primitive or equivalent. * @@ -188,30 +251,65 @@ private static T parse(Object value, Class clazz) { return null; } + /** + * Determines if this is boolean type. + * @param clazz Class to be checked. + * @return True if is boolean. + */ private static boolean isBooleanOrPrimitive(Class clazz) { return (Boolean.class == clazz) || (boolean.class == clazz); } + /** + * Determines if this is byte type. + * @param clazz Class to be checked. + * @return True if is byte. + */ private static boolean isByteOrPrimitive(Class clazz) { return (Byte.class == clazz) || (byte.class == clazz); } + /** + * Determines if this is short type. + * @param clazz Class to be checked. + * @return True if is short. + */ private static boolean isShortOrPrimitive(Class clazz) { return (Short.class == clazz) || (short.class == clazz); } + /** + * Determines if this is integer type. + * @param clazz Class to be checked. + * @return True if is integer. + */ private static boolean isIntegerOrPrimitive(Class clazz) { return (Integer.class == clazz) || (int.class == clazz); } + /** + * Determines if this is long type. + * @param clazz Class to be checked. + * @return True if is long. + */ private static boolean isLongOrPrimitive(Class clazz) { return (Long.class == clazz) || (long.class == clazz); } + /** + * Determines if this is float type. + * @param clazz Class to be checked. + * @return True if is float. + */ private static boolean isFloatOrPrimitive(Class clazz) { return (Float.class == clazz) || (float.class == clazz); } + /** + * Determines if this is double type. + * @param clazz Class to be checked. + * @return True if is double. + */ private static boolean isDoubleOrPrimitive(Class clazz) { return (Double.class == clazz) || (double.class == clazz); } diff --git a/sdk/src/test/java/io/dapr/client/DaprClientTestBuilder.java b/sdk/src/test/java/io/dapr/client/DaprClientTestBuilder.java new file mode 100644 index 0000000000..65333e2e1b --- /dev/null +++ b/sdk/src/test/java/io/dapr/client/DaprClientTestBuilder.java @@ -0,0 +1,21 @@ +/* + * Copyright (c) Microsoft Corporation. + * Licensed under the MIT License. + */ + +package io.dapr.client; + +/** + * Builder for DaprClient used in tests only. + */ +public class DaprClientTestBuilder { + + /** + * Builds a DaprClient. + * @param client DaprHttp used for http calls (can be mocked or stubbed) + * @return New instance of DaprClient. + */ + public static DaprClient buildHttpClient(DaprHttp client) { + return new DaprClientHttpAdapter(client); + } +} diff --git a/sdk/src/test/java/io/dapr/client/DaprHttpStub.java b/sdk/src/test/java/io/dapr/client/DaprHttpStub.java new file mode 100644 index 0000000000..0ede2041ba --- /dev/null +++ b/sdk/src/test/java/io/dapr/client/DaprHttpStub.java @@ -0,0 +1,48 @@ +/* + * Copyright (c) Microsoft Corporation. + * Licensed under the MIT License. + */ + +package io.dapr.client; + +import reactor.core.publisher.Mono; + +import java.util.Map; + +/** + * Stub class for DaprHttp. + * Useful to mock as well since it provides a default constructor. + */ +public class DaprHttpStub extends DaprHttp { + + /** + * Instantiates a stub for DaprHttp + */ + public DaprHttpStub() { + super("http://localhost", 3000, null); + } + + /** + * {@inheritDoc} + */ + @Override + public Mono invokeAPI(String method, String urlString, Map headers) { + return Mono.empty(); + } + + /** + * {@inheritDoc} + */ + @Override + public Mono invokeAPI(String method, String urlString, String content, Map headers) { + return Mono.empty(); + } + + /** + * {@inheritDoc} + */ + @Override + public Mono invokeAPI(String method, String urlString, byte[] content, Map headers) { + return Mono.empty(); + } +} diff --git a/sdk/src/test/java/io/dapr/runtime/DaprRuntimeTest.java b/sdk/src/test/java/io/dapr/runtime/DaprRuntimeTest.java new file mode 100644 index 0000000000..7c0c943fd9 --- /dev/null +++ b/sdk/src/test/java/io/dapr/runtime/DaprRuntimeTest.java @@ -0,0 +1,192 @@ +/* + * Copyright (c) Microsoft Corporation. + * Licensed under the MIT License. + */ + +package io.dapr.runtime; + +import com.fasterxml.jackson.databind.ObjectMapper; +import io.dapr.client.*; +import io.dapr.runtime.Dapr; +import io.dapr.runtime.DaprRuntime; +import io.dapr.runtime.TopicListener; +import io.dapr.utils.Constants; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import reactor.core.publisher.Mono; + +import java.lang.reflect.Field; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.Map; +import java.util.UUID; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.*; + +public class DaprRuntimeTest { + + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + + private static final String TYPE_PLAIN_TEXT = "plain/text"; + + private static final String TOPIC_NAME = "mytopic"; + + private final DaprRuntime daprRuntime = Dapr.getInstance(); + + @Before + public void setup() throws Exception { + // Only for unit tests to simulate a new start of the app. + Field field = this.daprRuntime.getClass().getDeclaredField("instance"); + field.setAccessible(true); + field.set(null, null); + } + + @Test + public void pubSubHappyCase() throws Exception { + Assert.assertNotNull(this.daprRuntime.getSubscribedTopics()); + Assert.assertTrue(this.daprRuntime.getSubscribedTopics().isEmpty()); + + TopicListener listener = mock(TopicListener.class); + this.daprRuntime.subscribeToTopic(TOPIC_NAME, listener); + + verify(listener, never()).process(any(), any(), any(), any()); + + Dapr.Message[] messages = new Dapr.Message[]{ + new Dapr.Message( + generateMessageId(), + TYPE_PLAIN_TEXT, + generatePayload(), + generateSingleMetadata()), + new Dapr.Message( + generateMessageId(), + TYPE_PLAIN_TEXT, + new byte[0], + generateSingleMetadata()), + new Dapr.Message( + generateMessageId(), + TYPE_PLAIN_TEXT, + null, + generateSingleMetadata()), + new Dapr.Message( + generateMessageId(), + TYPE_PLAIN_TEXT, + generatePayload(), + null), + new Dapr.Message( + "", + TYPE_PLAIN_TEXT, + generatePayload(), + generateSingleMetadata()), + new Dapr.Message( + null, + TYPE_PLAIN_TEXT, + generatePayload(), + generateSingleMetadata()), + new Dapr.Message( + generateMessageId(), + "", + generatePayload(), + generateSingleMetadata()), + new Dapr.Message( + generateMessageId(), + null, + generatePayload(), + generateSingleMetadata()) + }; + + DaprHttpStub daprHttp = mock(DaprHttpStub.class); + DaprClient client = DaprClientTestBuilder.buildHttpClient(daprHttp); + + for (Dapr.Message message : messages) { + when(daprHttp.invokeAPI( + eq("POST"), + eq(Constants.PUBLISH_PATH + "/" + TOPIC_NAME), + eq(message.getData()), + eq(null))) + .thenAnswer(invocationOnMock -> { + this.daprRuntime.handleInvocation( + TOPIC_NAME, OBJECT_MAPPER.writeValueAsBytes(message), message.getMetadata()); + return Mono.empty(); + }); + + client.publishEvent(TOPIC_NAME, message.getData()).block(); + + verify(listener, times(1)) + .process(eq(message.getId()), eq(message.getDatacontenttype()), eq(message.getData()), eq(message.getMetadata())); + } + + verify(listener, times(messages.length)).process(any(), any(), any(), any()); + } + + @Test(expected = RuntimeException.class) + public void subscribeCallbackException() throws Exception { + Assert.assertNotNull(this.daprRuntime.getSubscribedTopics()); + Assert.assertTrue(this.daprRuntime.getSubscribedTopics().isEmpty()); + + TopicListener listener = mock(TopicListener.class); + when(listener.process(any(), any(), any(), any())) + .thenReturn(Mono.error(new RuntimeException())); + + this.daprRuntime.subscribeToTopic(TOPIC_NAME, listener); + + Dapr.Message message = new Dapr.Message( + generateMessageId(), + TYPE_PLAIN_TEXT, + generatePayload(), + generateSingleMetadata()); + + Mono result = this.daprRuntime + .handleInvocation(TOPIC_NAME, OBJECT_MAPPER.writeValueAsBytes(message), message.getMetadata()); + + verify(listener, times(1)) + .process( + eq(message.getId()), + eq(message.getDatacontenttype()), + eq(message.getData()), + eq(message.getMetadata())); + + result.block(); + } + + @Test(expected = RuntimeException.class) + public void subscribeUnknownTopic() throws Exception { + Assert.assertNotNull(this.daprRuntime.getSubscribedTopics()); + Assert.assertTrue(this.daprRuntime.getSubscribedTopics().isEmpty()); + + TopicListener listener = mock(TopicListener.class); + + this.daprRuntime.subscribeToTopic(TOPIC_NAME, listener); + + Dapr.Message message = new Dapr.Message( + generateMessageId(), + TYPE_PLAIN_TEXT, + generatePayload(), + generateSingleMetadata()); + + Mono result = this.daprRuntime + .handleInvocation("UNKNOWN", OBJECT_MAPPER.writeValueAsBytes(message), message.getMetadata()); + + verify(listener, never()) + .process( + eq(message.getId()), + eq(message.getDatacontenttype()), + eq(message.getData()), + eq(message.getMetadata())); + + result.block(); + } + + private static final String generateMessageId() { + return UUID.randomUUID().toString(); + } + + private static final byte[] generatePayload() { + return UUID.randomUUID().toString().getBytes(StandardCharsets.UTF_8); + } + + private static final Map generateSingleMetadata() { + return Collections.singletonMap(UUID.randomUUID().toString(), UUID.randomUUID().toString()); + } +}