Skip to content

Fixing bug, where call should be deferred instead of async #102

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
LMWF merged 6 commits into from
Jan 16, 2020
+203 −70
Merged

Fixing bug, where call should be deferred instead of async #102

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
1a0debd
Fixing bug, where call should be deferred instead of async
Jan 15, 2020
feceb1a
Removing unused pool property
Jan 15, 2020
4da1c94
Fixing assertion based on previous changes
Jan 15, 2020
329fb3d
Adding test case to verify that the call to grpc only happens when th…
Jan 16, 2020
4dc4f3c
Fixing merge conflicts
Jan 16, 2020
d1b4585
Documenting complex test case
Jan 16, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
79 changes: 34 additions & 45 deletions sdk/src/main/java/io/dapr/client/DaprClientGrpcAdapter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,14 +71,11 @@ public <T> Mono<Void> publishEvent(String topic, T event, Map<String, String> me

DaprProtos.PublishEventEnvelope envelope = DaprProtos.PublishEventEnvelope.newBuilder()
.setTopic(topic).setData(data).build();
ListenableFuture<Empty> 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<Empty> futureEmpty = client.publishEvent(envelope);
futureEmpty.get();
return null;
});
} catch (Exception ex) {
return Mono.error(ex);
Expand All @@ -92,16 +89,11 @@ public <T> Mono<Void> publishEvent(String topic, T event, Map<String, String> me
public <T, R> Mono<T> invokeService(Verb verb, String appId, String method, R request, Map<String, String> metadata, Class<T> clazz) {
try {
DaprProtos.InvokeServiceEnvelope envelope = buildInvokeServiceEnvelope(verb.toString(), appId, method, request);
ListenableFuture<DaprProtos.InvokeServiceResponseEnvelope> 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<DaprProtos.InvokeServiceResponseEnvelope> futureResponse =
client.invokeService(envelope);
return objectSerializer.deserialize(futureResponse.get().getData().getValue().toByteArray(), clazz);
});

} catch (Exception ex) {
return Mono.error(ex);
}
Expand Down Expand Up @@ -151,14 +143,10 @@ public <T> Mono<Void> invokeBinding(String name, T request) {
.setName(name)
.setData(data);
DaprProtos.InvokeBindingEnvelope envelope = builder.build();
ListenableFuture<Empty> 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<Empty> futureEmpty = client.invokeBinding(envelope);
futureEmpty.get();
return null;
});
} catch (Exception ex) {
return Mono.error(ex);
Expand All @@ -167,7 +155,7 @@ public <T> Mono<Void> invokeBinding(String name, T request) {

/**
* @return Returns an io.dapr.client.domain.StateKeyValue
*
* <p>
* {@inheritDoc}
*/
@Override
Expand All @@ -180,22 +168,23 @@ public <T> Mono<StateKeyValue<T>> getState(StateKeyValue<T> state, StateOptions
}

DaprProtos.GetStateEnvelope envelope = builder.build();
ListenableFuture<DaprProtos.GetStateResponseEnvelope> futureResponse = client.getState(envelope);
return Mono.just(futureResponse).flatMap(f -> {
return Mono.fromCallable(() -> {
ListenableFuture<DaprProtos.GetStateResponseEnvelope> 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 <T> StateKeyValue<T> buildStateKeyValue(DaprProtos.GetStateResponseEnvelope resonse, String requestedKey, StateOptions stateOptions, Class<T> clazz) throws IOException {
T value = objectSerializer.deserialize(resonse.getData().getValue().toByteArray(), clazz);
String etag = resonse.getEtag();
private <T> StateKeyValue<T> buildStateKeyValue(DaprProtos.GetStateResponseEnvelope response, String requestedKey, StateOptions stateOptions, Class<T> 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);
}
Expand All @@ -208,8 +197,7 @@ public <T> Mono<Void> saveStates(List<StateKeyValue<T>> 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<Empty> futureEmpty = client.saveState(envelope);
Expand Down Expand Up @@ -282,7 +270,7 @@ public <T> Mono<Void> saveState(String key, String etag, T value, StateOptions o
@Override
public <T> Mono<Void> deleteState(StateKeyValue<T> state, StateOptions options) {
try {
DaprProtos.StateOptions.Builder optionBuilder = null;
DaprProtos.StateOptions.Builder optionBuilder = null;

if (options != null) {
optionBuilder = DaprProtos.StateOptions.newBuilder();
Expand Down Expand Up @@ -407,12 +395,13 @@ public Mono<Void> 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 <K> 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 <K> 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 <K> DaprProtos.InvokeServiceEnvelope buildInvokeServiceEnvelope(
Expand Down
12 changes: 3 additions & 9 deletions sdk/src/main/java/io/dapr/client/DaprHttp.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}.
*
Expand All @@ -97,7 +92,6 @@ public int getStatusCode() {
DaprHttp(int port, OkHttpClient httpClient) {
this.port = port;
this.httpClient = httpClient;
this.pool = Executors.newWorkStealingPool();
}

/**
Expand Down Expand Up @@ -132,7 +126,7 @@ public Mono<Response> invokeAPI(String method, String urlString, Map<String, Str
* @return Asynchronous text
*/
public Mono<Response> invokeAPI(String method, String urlString, Map<String, String> urlParameters, byte[] content, Map<String, String> headers) {
return Mono.fromFuture(CompletableFuture.supplyAsync(
Comment thread
LMWF marked this conversation as resolved.
return Mono.fromCallable(
() -> {
try {
String requestId = UUID.randomUUID().toString();
Expand Down Expand Up @@ -185,12 +179,12 @@ public Mono<Response> invokeAPI(String method, String urlString, Map<String, Str
response.headers().forEach(pair -> {
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));
});
Comment thread
LMWF marked this conversation as resolved.
}

/**
Expand Down
122 changes: 110 additions & 12 deletions sdk/src/test/java/io/dapr/client/DaprClientGrpcAdapterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -42,13 +45,13 @@ public void setup() {

@Test(expected = UnsupportedOperationException.class)
public void unregisterActorTimerTest() {
Mono<Void> result = adater.unregisterActorTimer("actorType", "actorId", "timerName");
Mono<Void> result = adater.unregisterActorTimer("actorType", "actorId", "timerName");
result.block();
}

@Test(expected = UnsupportedOperationException.class)
public void registerActorTimerTest() {
Mono<Void> result = adater.registerActorTimer("actorType", "actorId", "timerName" , "DATA");
Mono<Void> result = adater.registerActorTimer("actorType", "actorId", "timerName", "DATA");
result.block();
}

Expand Down Expand Up @@ -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<DaprProtos.InvokeServiceResponseEnvelope> settableFuture = SettableFuture.create();
MockCallback<DaprProtos.InvokeServiceResponseEnvelope> callback =
Expand Down Expand Up @@ -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<String> result = adater.invokeService(Verb.GET, "appId", "method",null, String.class);
Mono<String> 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<DaprProtos.InvokeServiceResponseEnvelope> settableFuture = SettableFuture.create();

Expand All @@ -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<String> result = adater.invokeService(Verb.GET, "appId", "method",null, String.class);
Mono<String> result = adater.invokeService(Verb.GET, "appId", "method", null, String.class);
String strOutput = result.block();
assertEquals(serializer.serializeString(resultObj), strOutput);
}
Expand Down Expand Up @@ -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<DaprProtos.InvokeServiceResponseEnvelope> settableFuture = SettableFuture.create();
MockCallback<DaprProtos.InvokeServiceResponseEnvelope> callback =
Expand Down Expand Up @@ -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<DaprProtos.InvokeServiceResponseEnvelope> settableFuture = SettableFuture.create();

Expand Down Expand Up @@ -491,10 +494,7 @@ public void getStateStringValueNoOptionsTest() throws IOException {
String key = "key1";
String expectedValue = "Expected state";
StateKeyValue<String> 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<DaprProtos.GetStateResponseEnvelope> settableFuture = SettableFuture.create();
MockCallback<DaprProtos.GetStateResponseEnvelope> callback = new MockCallback<>(responseEnvelope);
addCallback(settableFuture, callback, directExecutor());
Expand Down Expand Up @@ -880,6 +880,76 @@ private <T> StateKeyValue<T> 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.
* <p>This test will execute the following flow:</p>
* <ol>
* <li>Exeucte client getState for Key=key1</li>
* <li>Block for result to the the state</li>
* <li>Assert the Returned State is the expected to key1</li>
* <li>Execute client getState for Key=key2</li>
* <li>Execute client deleteState for Key=key2</li>
* <li>Block for deleteState call.</li>
* <li>Block for getState for Key=key2 and Assert they 2 was not found.</li>
* </ol>
* @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<String> expectedState1 = buildStateKey(expectedValue1, key1, etag, null);
Map<String, SettableFuture<DaprProtos.GetStateResponseEnvelope>> 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<String> keyRequest1 = buildStateKey(null, key1, etag, null);
Mono<StateKeyValue<String>> resultGet1 = adater.getState(keyRequest1, null, String.class);
assertEquals(expectedState1, resultGet1.block());
StateKeyValue<String> keyRequest2 = buildStateKey(null, key2, etag, null);
Mono<StateKeyValue<String>> resultGet2 = adater.getState(keyRequest2, null, String.class);

SettableFuture<Empty> settableFutureDelete = SettableFuture.create();
MockCallback<Empty> callbackDelete = new MockCallback<>(Empty.newBuilder().build());
addCallback(settableFutureDelete, callbackDelete, directExecutor());
when(client.deleteState(any(io.dapr.DaprProtos.DeleteStateEnvelope.class)))
.thenReturn(settableFutureDelete);
Mono<Void> 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<String> state2 = resultGet2.block();
assertNull(state2);
}

private <T> SettableFuture<DaprProtos.GetStateResponseEnvelope> buildFutureGetStateEnvelop(T value, String etag) throws IOException {
DaprProtos.GetStateResponseEnvelope envelope = buildGetStateResponseEnvelope(value, etag);
SettableFuture<DaprProtos.GetStateResponseEnvelope> settableFuture = SettableFuture.create();
MockCallback<DaprProtos.GetStateResponseEnvelope> callback = new MockCallback<>(envelope);
addCallback(settableFuture, callback, directExecutor());
settableFuture.set(envelope);

return settableFuture;
}

private <T> 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) {

Expand Down Expand Up @@ -977,4 +1047,32 @@ public int hashCode() {
return result;
}
}

private static class GetStateEnvelopeKeyMatcher implements ArgumentMatcher<DaprProtos.GetStateEnvelope> {

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 "<Has property of certain value (propName: " + propValue + ") matcher>";
}
}
}
Loading