diff --git a/sdk/src/main/java/io/dapr/client/DaprClientGrpcAdapter.java b/sdk/src/main/java/io/dapr/client/DaprClientGrpcAdapter.java index 8e2d27a99e..13326eeac5 100644 --- a/sdk/src/main/java/io/dapr/client/DaprClientGrpcAdapter.java +++ b/sdk/src/main/java/io/dapr/client/DaprClientGrpcAdapter.java @@ -71,14 +71,11 @@ public Mono publishEvent(String topic, T event, Map me DaprProtos.PublishEventEnvelope envelope = DaprProtos.PublishEventEnvelope.newBuilder() .setTopic(topic).setData(data).build(); - ListenableFuture futureEmpty = client.publishEvent(envelope); - return Mono.just(futureEmpty).flatMap(f -> { - try { - f.get(); - } catch (Exception ex) { - return Mono.error(ex); - } - return Mono.empty(); + + return Mono.fromCallable(() -> { + ListenableFuture futureEmpty = client.publishEvent(envelope); + futureEmpty.get(); + return null; }); } catch (Exception ex) { return Mono.error(ex); @@ -92,16 +89,11 @@ public Mono publishEvent(String topic, T event, Map me public Mono invokeService(Verb verb, String appId, String method, R request, Map metadata, Class clazz) { try { DaprProtos.InvokeServiceEnvelope envelope = buildInvokeServiceEnvelope(verb.toString(), appId, method, request); - ListenableFuture futureResponse = - client.invokeService(envelope); - return Mono.just(futureResponse).flatMap(f -> { - try { - return Mono.just(objectSerializer.deserialize(f.get().getData().getValue().toByteArray(), clazz)); - } catch (Exception ex) { - return Mono.error(ex); - } + return Mono.fromCallable(() -> { + ListenableFuture futureResponse = + client.invokeService(envelope); + return objectSerializer.deserialize(futureResponse.get().getData().getValue().toByteArray(), clazz); }); - } catch (Exception ex) { return Mono.error(ex); } @@ -151,14 +143,10 @@ public Mono invokeBinding(String name, T request) { .setName(name) .setData(data); DaprProtos.InvokeBindingEnvelope envelope = builder.build(); - ListenableFuture futureEmpty = client.invokeBinding(envelope); - return Mono.just(futureEmpty).flatMap(f -> { - try { - f.get(); - } catch (Exception ex) { - return Mono.error(ex); - } - return Mono.empty(); + return Mono.fromCallable(() -> { + ListenableFuture futureEmpty = client.invokeBinding(envelope); + futureEmpty.get(); + return null; }); } catch (Exception ex) { return Mono.error(ex); @@ -167,7 +155,7 @@ public Mono invokeBinding(String name, T request) { /** * @return Returns an io.dapr.client.domain.StateKeyValue - * + *

* {@inheritDoc} */ @Override @@ -180,22 +168,23 @@ public Mono> getState(StateKeyValue state, StateOptions } DaprProtos.GetStateEnvelope envelope = builder.build(); - ListenableFuture futureResponse = client.getState(envelope); - return Mono.just(futureResponse).flatMap(f -> { + return Mono.fromCallable(() -> { + ListenableFuture futureResponse = client.getState(envelope); + DaprProtos.GetStateResponseEnvelope response = null; try { - return Mono.just(buildStateKeyValue(f.get(), state.getKey(), stateOptions, clazz)); - } catch (Exception ex) { - return Mono.error(ex); + response = futureResponse.get(); + } catch (NullPointerException npe) { + return null; } - }); - } catch (Exception ex) { + return buildStateKeyValue(response, state.getKey(), stateOptions, clazz); + }); } catch (Exception ex) { return Mono.error(ex); } } - private StateKeyValue buildStateKeyValue(DaprProtos.GetStateResponseEnvelope resonse, String requestedKey, StateOptions stateOptions, Class clazz) throws IOException { - T value = objectSerializer.deserialize(resonse.getData().getValue().toByteArray(), clazz); - String etag = resonse.getEtag(); + private StateKeyValue buildStateKeyValue(DaprProtos.GetStateResponseEnvelope response, String requestedKey, StateOptions stateOptions, Class clazz) throws IOException { + T value = objectSerializer.deserialize(Optional.ofNullable(response.getData().getValue().toByteArray()).orElse(null), clazz); + String etag = response.getEtag(); String key = requestedKey; return new StateKeyValue<>(value, key, etag, stateOptions); } @@ -208,8 +197,7 @@ public Mono saveStates(List> states) { try { DaprProtos.SaveStateEnvelope.Builder builder = DaprProtos.SaveStateEnvelope.newBuilder(); for (StateKeyValue state : states) { - builder.addRequests(buildStateRequest(state).build()); - } + builder.addRequests(buildStateRequest(state).build()); } DaprProtos.SaveStateEnvelope envelope = builder.build(); ListenableFuture futureEmpty = client.saveState(envelope); @@ -282,7 +270,7 @@ public Mono saveState(String key, String etag, T value, StateOptions o @Override public Mono deleteState(StateKeyValue state, StateOptions options) { try { - DaprProtos.StateOptions.Builder optionBuilder = null; + DaprProtos.StateOptions.Builder optionBuilder = null; if (options != null) { optionBuilder = DaprProtos.StateOptions.newBuilder(); @@ -407,12 +395,13 @@ public Mono unregisterActorTimer(String actorType, String actorId, String /** * Builds the object io.dapr.{@link DaprProtos.InvokeServiceEnvelope} to be send based on the parameters. - * @param verb String that must match HTTP Methods - * @param appId The application id to be invoked - * @param method The application method to be invoked - * @param request The body of the request to be send as part of the invokation - * @param The Type of the Body - * @return The object to be sent as part of the invokation. + * + * @param verb String that must match HTTP Methods + * @param appId The application id to be invoked + * @param method The application method to be invoked + * @param request The body of the request to be send as part of the invokation + * @param The Type of the Body + * @return The object to be sent as part of the invokation. * @throws IOException If there's an issue serializing the request. */ private DaprProtos.InvokeServiceEnvelope buildInvokeServiceEnvelope( diff --git a/sdk/src/main/java/io/dapr/client/DaprHttp.java b/sdk/src/main/java/io/dapr/client/DaprHttp.java index 82736f64f0..aa49a743cf 100644 --- a/sdk/src/main/java/io/dapr/client/DaprHttp.java +++ b/sdk/src/main/java/io/dapr/client/DaprHttp.java @@ -83,11 +83,6 @@ public int getStatusCode() { */ private final OkHttpClient httpClient; - /** - * Thread-pool for HTTP calls. - */ - private final ExecutorService pool; - /** * Creates a new instance of {@link DaprHttp}. * @@ -97,7 +92,6 @@ public int getStatusCode() { DaprHttp(int port, OkHttpClient httpClient) { this.port = port; this.httpClient = httpClient; - this.pool = Executors.newWorkStealingPool(); } /** @@ -132,7 +126,7 @@ public Mono invokeAPI(String method, String urlString, Map invokeAPI(String method, String urlString, Map urlParameters, byte[] content, Map headers) { - return Mono.fromFuture(CompletableFuture.supplyAsync( + return Mono.fromCallable( () -> { try { String requestId = UUID.randomUUID().toString(); @@ -185,12 +179,12 @@ public Mono invokeAPI(String method, String urlString, Map { mapHeaders.put(pair.getFirst(), pair.getSecond()); }); - return new Response(result, mapHeaders, response.code()); + return new Response(result.length > 0 ? result : null, mapHeaders, response.code()); } } catch (Exception e) { throw new RuntimeException(e); } - }, this.pool)); + }); } /** diff --git a/sdk/src/test/java/io/dapr/client/DaprClientGrpcAdapterTest.java b/sdk/src/test/java/io/dapr/client/DaprClientGrpcAdapterTest.java index dd83cb318b..e36d8442f2 100644 --- a/sdk/src/test/java/io/dapr/client/DaprClientGrpcAdapterTest.java +++ b/sdk/src/test/java/io/dapr/client/DaprClientGrpcAdapterTest.java @@ -14,12 +14,15 @@ import org.checkerframework.checker.nullness.compatqual.NullableDecl; import org.junit.Before; import org.junit.Test; +import org.mockito.ArgumentMatcher; import reactor.core.publisher.Mono; import javax.annotation.Nullable; import java.io.IOException; import java.time.Duration; +import java.util.HashMap; +import java.util.Map; import static com.google.common.util.concurrent.Futures.addCallback; import static com.google.common.util.concurrent.MoreExecutors.directExecutor; @@ -42,13 +45,13 @@ public void setup() { @Test(expected = UnsupportedOperationException.class) public void unregisterActorTimerTest() { - Mono result = adater.unregisterActorTimer("actorType", "actorId", "timerName"); + Mono result = adater.unregisterActorTimer("actorType", "actorId", "timerName"); result.block(); } @Test(expected = UnsupportedOperationException.class) public void registerActorTimerTest() { - Mono result = adater.registerActorTimer("actorType", "actorId", "timerName" , "DATA"); + Mono result = adater.registerActorTimer("actorType", "actorId", "timerName", "DATA"); result.block(); } @@ -273,7 +276,7 @@ public void invokeServiceTest() throws Exception { } @Test - public void invokeServiceObjectTest() throws Exception { + public void invokeServiceObjectTest() throws Exception { MyObject resultObj = new MyObject(1, "Value"); SettableFuture settableFuture = SettableFuture.create(); MockCallback callback = @@ -322,13 +325,13 @@ public void invokeServiceNoRequestBodyTest() throws Exception { settableFuture.set(DaprProtos.InvokeServiceResponseEnvelope.newBuilder().setData(getAny(expected)).build()); when(client.invokeService(any(DaprProtos.InvokeServiceEnvelope.class))) .thenReturn(settableFuture); - Mono result = adater.invokeService(Verb.GET, "appId", "method",null, String.class); + Mono result = adater.invokeService(Verb.GET, "appId", "method", null, String.class); String strOutput = result.block(); assertEquals(expected, strOutput); } @Test - public void invokeServiceNoRequestBodyObjectTest() throws Exception { + public void invokeServiceNoRequestBodyObjectTest() throws Exception { MyObject resultObj = new MyObject(1, "Value"); SettableFuture settableFuture = SettableFuture.create(); @@ -339,7 +342,7 @@ public void invokeServiceNoRequestBodyObjectTest() throws Exception { settableFuture.set(DaprProtos.InvokeServiceResponseEnvelope.newBuilder().setData(getAny(resultObj)).build()); when(client.invokeService(any(DaprProtos.InvokeServiceEnvelope.class))) .thenReturn(settableFuture); - Mono result = adater.invokeService(Verb.GET, "appId", "method",null, String.class); + Mono result = adater.invokeService(Verb.GET, "appId", "method", null, String.class); String strOutput = result.block(); assertEquals(serializer.serializeString(resultObj), strOutput); } @@ -390,7 +393,7 @@ public void invokeByteRequestServiceTest() throws Exception { } @Test - public void invokeServiceByteRequestObjectTest() throws Exception { + public void invokeServiceByteRequestObjectTest() throws Exception { MyObject resultObj = new MyObject(1, "Value"); SettableFuture settableFuture = SettableFuture.create(); MockCallback callback = @@ -446,7 +449,7 @@ public void invokeServiceNoRequestNoClassBodyTest() throws Exception { } @Test - public void invokeServiceNoRequestNoClassBodyObjectTest() throws Exception { + public void invokeServiceNoRequestNoClassBodyObjectTest() throws Exception { MyObject resultObj = new MyObject(1, "Value"); SettableFuture settableFuture = SettableFuture.create(); @@ -491,10 +494,7 @@ public void getStateStringValueNoOptionsTest() throws IOException { String key = "key1"; String expectedValue = "Expected state"; StateKeyValue expectedState = buildStateKey(expectedValue, key, etag, null); - DaprProtos.GetStateResponseEnvelope responseEnvelope = DaprProtos.GetStateResponseEnvelope.newBuilder() - .setData(getAny(expectedValue)) - .setEtag(etag) - .build(); + DaprProtos.GetStateResponseEnvelope responseEnvelope = buildGetStateResponseEnvelope(expectedValue, etag); SettableFuture settableFuture = SettableFuture.create(); MockCallback callback = new MockCallback<>(responseEnvelope); addCallback(settableFuture, callback, directExecutor()); @@ -880,6 +880,76 @@ private StateKeyValue buildStateKey(T value, String key, String etag, Sta return new StateKeyValue(value, key, etag, options); } + /** + * The purpose of this test is to show that it doesn't matter when the client is called, the actual coll to DAPR + * will be done when the output Mono response call the Mono.block method. + * Like for instanche if you call getState, withouth blocking for the response, and then call delete for the same state + * you just retrived but block for the delete response, when later you block for the response of the getState, you will + * not found the state. + *

This test will execute the following flow:

+ *
    + *
  1. Exeucte client getState for Key=key1
    2. + *
  2. Block for result to the the state
    3. + *
  3. Assert the Returned State is the expected to key1
    4. + *
  4. Execute client getState for Key=key2
    5. + *
  5. Execute client deleteState for Key=key2
    6. + *
  6. Block for deleteState call.
    7. + *
  7. Block for getState for Key=key2 and Assert they 2 was not found.
    8. + *
+ * @throws Exception + */ + + @Test + public void getStateDeleteStateThenBlockDeleteThenBlockGet() throws Exception { + String etag = "ETag1"; + String key1 = "key1"; + String expectedValue1 = "Expected state 1"; + String key2 = "key2"; + String expectedValue2 = "Expected state 2"; + StateKeyValue expectedState1 = buildStateKey(expectedValue1, key1, etag, null); + Map> futuresMap = new HashMap<>(); + futuresMap.put(key1, buildFutureGetStateEnvelop(expectedValue1, etag)); + futuresMap.put(key2, buildFutureGetStateEnvelop(expectedValue2, etag)); + when(client.getState(argThat(new GetStateEnvelopeKeyMatcher(key1)))).thenReturn(futuresMap.get(key1)); + StateKeyValue keyRequest1 = buildStateKey(null, key1, etag, null); + Mono> resultGet1 = adater.getState(keyRequest1, null, String.class); + assertEquals(expectedState1, resultGet1.block()); + StateKeyValue keyRequest2 = buildStateKey(null, key2, etag, null); + Mono> resultGet2 = adater.getState(keyRequest2, null, String.class); + + SettableFuture settableFutureDelete = SettableFuture.create(); + MockCallback callbackDelete = new MockCallback<>(Empty.newBuilder().build()); + addCallback(settableFutureDelete, callbackDelete, directExecutor()); + when(client.deleteState(any(io.dapr.DaprProtos.DeleteStateEnvelope.class))) + .thenReturn(settableFutureDelete); + Mono resultDelete = adater.deleteState(keyRequest2, null); + settableFutureDelete.set(Empty.newBuilder().build()); + resultDelete.block(); + assertTrue(callbackDelete.wasCalled); + futuresMap.replace(key2, null); + when(client.getState(argThat(new GetStateEnvelopeKeyMatcher(key2)))).thenReturn(futuresMap.get(key2)); + + StateKeyValue state2 = resultGet2.block(); + assertNull(state2); + } + + private SettableFuture buildFutureGetStateEnvelop(T value, String etag) throws IOException { + DaprProtos.GetStateResponseEnvelope envelope = buildGetStateResponseEnvelope(value, etag); + SettableFuture settableFuture = SettableFuture.create(); + MockCallback callback = new MockCallback<>(envelope); + addCallback(settableFuture, callback, directExecutor()); + settableFuture.set(envelope); + + return settableFuture; + } + + private DaprProtos.GetStateResponseEnvelope buildGetStateResponseEnvelope(T value, String etag) throws IOException { + return DaprProtos.GetStateResponseEnvelope.newBuilder() + .setData(getAny(value)) + .setEtag(etag) + .build(); + } + private StateOptions buildStateOptions(StateOptions.Consistency consistency, StateOptions.Concurrency concurrency, Duration interval, Integer threshold, StateOptions.RetryPolicy.Pattern pattern) { @@ -977,4 +1047,32 @@ public int hashCode() { return result; } } + + private static class GetStateEnvelopeKeyMatcher implements ArgumentMatcher { + + private final String propValue; + + GetStateEnvelopeKeyMatcher(String propValue) { + this.propValue = propValue; + } + + @Override + public boolean matches(DaprProtos.GetStateEnvelope argument) { + if (argument == null) { + return false; + } + if (propValue == null && argument.getKey() != null) { + return false; + } + if (propValue == null && argument.getKey() == null) { + return true; + } + return propValue.equals(argument.getKey()); + } + + @Override + public String toString() { + return ""; + } + } } diff --git a/sdk/src/test/java/io/dapr/client/DaprHttpTest.java b/sdk/src/test/java/io/dapr/client/DaprHttpTest.java index 041c08a11e..5b2a7d3235 100644 --- a/sdk/src/test/java/io/dapr/client/DaprHttpTest.java +++ b/sdk/src/test/java/io/dapr/client/DaprHttpTest.java @@ -4,6 +4,8 @@ */ package io.dapr.client; +import io.dapr.exceptions.DaprException; +import io.dapr.utils.Constants; import io.dapr.utils.ObjectSerializer; import okhttp3.*; import okhttp3.mock.Behavior; @@ -16,7 +18,7 @@ import java.util.HashMap; import java.util.Map; -import static org.junit.Assert.assertEquals; +import static org.junit.Assert.*; public class DaprHttpTest { @@ -177,16 +179,66 @@ public void invokePostMethodUnknownError() throws IOException { @Test public void getHeadersAndStatus(){ - mockInterceptor.addRule() .post("http://localhost:3500/v1.0/state") .respond(500, ResponseBody.create(MediaType.parse("application/json"), "{\"errorCode\":\"null\",\"message\":\"null\"}")); - DaprHttp daprHttp = new DaprHttp(3500, okHttpClient); - System.out.println(daprHttp); + } + /** + * The purpose of this test is to show that it doesn't matter when the client is called, the actual coll to DAPR + * will be done when the output Mono response call the Mono.block method. + * Like for instanche if you call getState, withouth blocking for the response, and then call delete for the same state + * you just retrived but block for the delete response, when later you block for the response of the getState, you will + * not found the state. + *

This test will execute the following flow:

+ *
    + *
  1. Exeucte client getState for Key=key1
    2. + *
  2. Block for result to the the state
    3. + *
  3. Assert the Returned State is the expected to key1
    4. + *
  4. Execute client getState for Key=key2
    5. + *
  5. Execute client deleteState for Key=key2
    6. + *
  6. Block for deleteState call.
    7. + *
  7. Block for getState for Key=key2 and Assert they 2 was not found.
    8. + *
+ * @throws Exception + */ + @Test () + public void testCallbackCalledAtTheExpectedTimeTest() throws IOException { + String deletedStateKey = "deletedKey"; + String existingState = "existingState"; + String urlDeleteState = Constants.STATE_PATH + "/" + deletedStateKey; + String urlExistingState = Constants.STATE_PATH + "/" + existingState; + mockInterceptor.addRule() + .get("http://localhost:3500/" + urlDeleteState) + .respond(200, ResponseBody.create(MediaType.parse("application/json"), + deletedStateKey)); + mockInterceptor.addRule() + .delete("http://localhost:3500/" + urlDeleteState) + .respond(204); + mockInterceptor.addRule() + .get("http://localhost:3500/" +urlExistingState) + .respond(200, ResponseBody.create(MediaType.parse("application/json"), + existingState)); + DaprHttp daprHttp = new DaprHttp(3500, okHttpClient); + Mono response = daprHttp.invokeAPI("GET", urlExistingState, null, null); + assertEquals(existingState, serializer.deserialize(response.block().getBody(), String.class)); + Mono responseDeleted = daprHttp.invokeAPI("GET", urlDeleteState, null, null); + Mono responseDeleteKey = daprHttp.invokeAPI("DELETE", urlDeleteState, null, null); + assertNull(serializer.deserialize(responseDeleteKey.block().getBody(), String.class)); + mockInterceptor.reset(); + mockInterceptor.addRule() + .get("http://localhost:3500/" +urlDeleteState) + .respond(404, ResponseBody.create(MediaType.parse("application/json"), + "{\"errorCode\":\"404\",\"message\":\"State Not Fuund\"}")); + try { + responseDeleted.block(); + fail("Expected DaprException"); + } catch (Exception ex) { + assertEquals(DaprException.class, ex.getCause().getCause().getClass()); + } } }