Skip to content

Receiving the StateOptions on save as part of the State, to comply with the DAPR API #105

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
artursouza merged 3 commits into from
Jan 16, 2020
Merged

Receiving the StateOptions on save as part of the State, to comply with the DAPR API #105

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f130992
Receiving the StateOptions on save as part of the State, to comply wi…
Jan 16, 2020
f327231
#26 Add integration test for Concurrency funtionality in the states m…
JuanJose-Herrera Jan 16, 2020
2c30072
Merge branch 'java_sdk_wip' into java_sdk_wip
artursouza 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
3 changes: 1 addition & 2 deletions sdk/src/main/java/io/dapr/client/DaprClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -129,11 +129,10 @@ public interface DaprClient {
* Save/Update a list of states.
*
* @param states the States to be saved.
* @param options the Options to use for each state.
* @param <T> the Type of the State.
* @return a Mono plan of type Void.
*/
<T> Mono<Void> saveStates(List<StateKeyValue<T>> states, StateOptions options);
<T> Mono<Void> saveStates(List<StateKeyValue<T>> states);

/**
* Save/Update a state.
Expand Down
96 changes: 50 additions & 46 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 @@ -183,7 +183,7 @@ public <T> Mono<StateKeyValue<T>> getState(StateKeyValue<T> state, StateOptions
ListenableFuture<DaprProtos.GetStateResponseEnvelope> futureResponse = client.getState(envelope);
return Mono.just(futureResponse).flatMap(f -> {
try {
return Mono.just(buildStateKeyValue(f.get(), state.getKey(), clazz));
return Mono.just(buildStateKeyValue(f.get(), state.getKey(), stateOptions, clazz));
} catch (Exception ex) {
return Mono.error(ex);
}
Expand All @@ -193,61 +193,22 @@ public <T> Mono<StateKeyValue<T>> getState(StateKeyValue<T> state, StateOptions
}
}

private <T> StateKeyValue<T> buildStateKeyValue(DaprProtos.GetStateResponseEnvelope resonse, String requestedKey, Class<T> clazz) throws IOException {
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();
String key = requestedKey;

return new StateKeyValue<>(value, key, etag);
return new StateKeyValue<>(value, key, etag, stateOptions);
}

/**
* {@inheritDoc}
*/
@Override
public <T> Mono<Void> saveStates(List<StateKeyValue<T>> states, StateOptions options) {
public <T> Mono<Void> saveStates(List<StateKeyValue<T>> states) {
try {
DaprProtos.StateRequestOptions.Builder optionBuilder = null;
if (options != null) {
DaprProtos.StateRetryPolicy.Builder retryPolicyBuilder = null;
if (options.getRetryPolicy() != null) {
retryPolicyBuilder = DaprProtos.StateRetryPolicy.newBuilder();
StateOptions.RetryPolicy retryPolicy = options.getRetryPolicy();
if (options.getRetryPolicy().getInterval() != null) {
Duration.Builder durationBuilder = Duration.newBuilder()
.setNanos(retryPolicy.getInterval().getNano())
.setSeconds(retryPolicy.getInterval().getSeconds());
retryPolicyBuilder.setInterval(durationBuilder.build());
}
retryPolicyBuilder.setThreshold(objectSerializer.deserialize(retryPolicy.getThreshold(), int.class));
if (retryPolicy.getPattern() != null) {
retryPolicyBuilder.setPattern(retryPolicy.getPattern().getValue());
}
}

optionBuilder = DaprProtos.StateRequestOptions.newBuilder();
if (options.getConcurrency() != null) {
optionBuilder.setConcurrency(options.getConcurrency().getValue());
}
if (options.getConsistency() != null) {
optionBuilder.setConsistency(options.getConsistency().getValue());
}
if (retryPolicyBuilder != null) {
optionBuilder.setRetryPolicy(retryPolicyBuilder.build());
}
}
DaprProtos.SaveStateEnvelope.Builder builder = DaprProtos.SaveStateEnvelope.newBuilder();
for (StateKeyValue state : states) {
byte[] byteState = objectSerializer.serialize(state.getValue());
Any data = Any.newBuilder().setValue(ByteString.copyFrom(byteState)).build();
DaprProtos.StateRequest.Builder stateBuilder = DaprProtos.StateRequest.newBuilder()
.setEtag(state.getEtag())
.setKey(state.getKey())
.setValue(data);
if(optionBuilder != null) {
stateBuilder.setOptions(optionBuilder.build());
}
builder.addRequests(stateBuilder.build());
builder.addRequests(buildStateRequest(state).build());
}
DaprProtos.SaveStateEnvelope envelope = builder.build();

Expand All @@ -265,10 +226,53 @@ public <T> Mono<Void> saveStates(List<StateKeyValue<T>> states, StateOptions opt
}
}

private <T> DaprProtos.StateRequest.Builder buildStateRequest(StateKeyValue<T> state) throws IOException {
byte[] byteState = objectSerializer.serialize(state.getValue());
Any data = Any.newBuilder().setValue(ByteString.copyFrom(byteState)).build();
DaprProtos.StateRequest.Builder stateBuilder = DaprProtos.StateRequest.newBuilder()
.setEtag(state.getEtag())
.setKey(state.getKey())
.setValue(data);
DaprProtos.StateRequestOptions.Builder optionBuilder = null;
if (state.getOptions() != null) {
StateOptions options = state.getOptions();
DaprProtos.StateRetryPolicy.Builder retryPolicyBuilder = null;
if (options.getRetryPolicy() != null) {
retryPolicyBuilder = DaprProtos.StateRetryPolicy.newBuilder();
StateOptions.RetryPolicy retryPolicy = options.getRetryPolicy();
if (options.getRetryPolicy().getInterval() != null) {
Duration.Builder durationBuilder = Duration.newBuilder()
.setNanos(retryPolicy.getInterval().getNano())
.setSeconds(retryPolicy.getInterval().getSeconds());
retryPolicyBuilder.setInterval(durationBuilder.build());
}
retryPolicyBuilder.setThreshold(objectSerializer.deserialize(retryPolicy.getThreshold(), int.class));
if (retryPolicy.getPattern() != null) {
retryPolicyBuilder.setPattern(retryPolicy.getPattern().getValue());
}
}

optionBuilder = DaprProtos.StateRequestOptions.newBuilder();
if (options.getConcurrency() != null) {
optionBuilder.setConcurrency(options.getConcurrency().getValue());
}
if (options.getConsistency() != null) {
optionBuilder.setConsistency(options.getConsistency().getValue());
}
if (retryPolicyBuilder != null) {
optionBuilder.setRetryPolicy(retryPolicyBuilder.build());
}
}
if(optionBuilder != null) {
stateBuilder.setOptions(optionBuilder.build());
}
return stateBuilder;
}

@Override
public <T> Mono<Void> saveState(String key, String etag, T value, StateOptions options) {
StateKeyValue<T> state = new StateKeyValue<>(value, key, etag);
return saveStates(Arrays.asList(state), options);
StateKeyValue<T> state = new StateKeyValue<>(value, key, etag, options);
return saveStates(Arrays.asList(state));
}

/**
Expand Down
21 changes: 10 additions & 11 deletions sdk/src/main/java/io/dapr/client/DaprClientHttpAdapter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -185,7 +185,7 @@ public <T> Mono<StateKeyValue<T>> getState(StateKeyValue<T> state, StateOptions
.invokeAPI(DaprHttp.HttpMethods.GET.name(), url.toString(), urlParameters, headers)
.flatMap(s -> {
try {
return Mono.just(buildStateKeyValue(s, state.getKey(), clazz));
return Mono.just(buildStateKeyValue(s, state.getKey(), stateOptions, clazz));
}catch (Exception ex){
return Mono.error(ex);
}
Expand All @@ -199,22 +199,21 @@ public <T> Mono<StateKeyValue<T>> getState(StateKeyValue<T> state, StateOptions
* {@inheritDoc}
*/
@Override
public <T> Mono<Void> saveStates(List<StateKeyValue<T>> states, StateOptions options) {
public <T> Mono<Void> saveStates(List<StateKeyValue<T>> states) {
try {
if (states == null || states.isEmpty()) {
return Mono.empty();
}
final Map<String, String> headers = new HashMap<>();
final String etag = states.stream().filter(state -> null != state.getEtag() && !state.getEtag().trim().isEmpty())
.findFirst().orElse(new StateKeyValue<>(null, null, null)).getEtag();
.findFirst().orElse(new StateKeyValue<>(null, null, null, null)).getEtag();
if (etag != null && !etag.trim().isEmpty()) {
headers.put(Constants.HEADER_HTTP_ETAG_ID, etag);
}
final String url = Constants.STATE_PATH;
Map<String, String> urlParameter = Optional.ofNullable(options).map(stateOptions -> stateOptions.getStateOptionsAsMap() ).orElse( new HashMap<>());
byte[] serializedStateBody = objectSerializer.serialize(states);
return this.client.invokeAPI(
DaprHttp.HttpMethods.POST.name(), url, urlParameter, serializedStateBody, headers).then();
DaprHttp.HttpMethods.POST.name(), url, null, serializedStateBody, headers).then();
} catch (Exception ex) {
return Mono.error(ex);
}
Expand All @@ -225,8 +224,8 @@ public <T> Mono<Void> saveStates(List<StateKeyValue<T>> states, StateOptions opt
*/
@Override
public <T> Mono<Void> saveState(String key, String etag, T value, StateOptions options) {
StateKeyValue<T> state = new StateKeyValue<>(value, key, etag);
return saveStates(Arrays.asList(state), options);
StateKeyValue<T> state = new StateKeyValue<>(value, key, etag, options);
return saveStates(Arrays.asList(state));
}

/**
Expand Down Expand Up @@ -339,14 +338,14 @@ public Mono<Void> unregisterActorTimer(String actorType, String actorId, String
* @return A StateKeyValue instance
* @throws IOException If there's a issue deserialzing the response.
*/
private <T> StateKeyValue<T> buildStateKeyValue(DaprHttp.Response resonse, String requestedKey, Class<T> clazz) throws IOException {
private <T> StateKeyValue<T> buildStateKeyValue(DaprHttp.Response resonse, String requestedKey, StateOptions stateOptions, Class<T> clazz) throws IOException {
T value = objectSerializer.deserialize(resonse.getBody(), clazz);
String key = requestedKey;
String etag = null;
if (resonse.getHeaders() != null && resonse.getHeaders().containsKey("ETag")) {
etag = objectSerializer.deserialize(resonse.getHeaders().get("ETag"), String.class);
if (resonse.getHeaders() != null && resonse.getHeaders().containsKey("Etag")) {
etag = objectSerializer.deserialize(resonse.getHeaders().get("Etag"), String.class);
}
return new StateKeyValue<>(value, key, etag);
return new StateKeyValue<>(value, key, etag, stateOptions);
}

}
40 changes: 29 additions & 11 deletions sdk/src/main/java/io/dapr/client/domain/StateKeyValue.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@

/**
* This class reprent what a State is
* @param <T>
* @param <T> The type of the value of the sate
*/
public class StateKeyValue<T> {
/**
Expand All @@ -23,42 +23,57 @@ public class StateKeyValue<T> {
*/
private final String etag;

/**
* The options used for saving the state
*/
private final StateOptions options;

/**
* Create an inmutable state
* @param value
* @param key
* @param etag
* @param value - The value of the state
* @param key - The key of the state
* @param etag - The etag of the state - Keep in mind that for some state stores (like reids) only numbers are supported.
* @param options - REQUIRED when saving a state.
*/
public StateKeyValue(T value, String key, String etag) {
public StateKeyValue(T value, String key, String etag, StateOptions options) {
this.value = value;
this.key = key;
this.etag = etag;
this.options = options;
}

/**
* Retrieves the Value of the state
* @return
* @return The value of the state
*/
public T getValue() {
return value;
}

/**
* Retrieves the Key of the state
* @return
* @return The key of the state
*/
public String getKey() {
return key;
}

/**
* Retrieve the ETag of this state
* @return
* @return The etag of the state
*/
public String getEtag() {
return etag;
}

/**
* Retrieve the Options used for saving the state
* @return The options to save the state
*/
public StateOptions getOptions() {
return options;
}

@Override
public boolean equals(Object o) {
if (this == o) return true;
Expand All @@ -69,6 +84,7 @@ public boolean equals(Object o) {
if (getValue() != null ? !getValue().equals(that.getValue()) : that.getValue() != null) return false;
if (getKey() != null ? !getKey().equals(that.getKey()) : that.getKey() != null) return false;
if (getEtag() != null ? !getEtag().equals(that.getEtag()) : that.getEtag() != null) return false;
if (getOptions() != null ? !getOptions().equals(that.getOptions()) : that.getOptions() != null) return false;

return true;
}
Expand All @@ -78,15 +94,17 @@ public int hashCode() {
int result = getValue() != null ? getValue().hashCode() : 0;
result = 31 * result + (getKey() != null ? getKey().hashCode() : 0);
result = 31 * result + (getEtag() != null ? getEtag().hashCode() : 0);
result = 31 * result + (getOptions() != null ? options.hashCode() : 0);
return result;
}

@Override
public String toString() {
return "StateKeyValue{" +
"value=" + value +
", key='" + key + '\'' +
", etag='" + etag + '\'' +
'}';
", key='" + key + "'" +
", etag='" + etag + "'" +
", options={'" + options.toString() + "}" +
"}";
}
}
5 changes: 5 additions & 0 deletions sdk/src/main/java/io/dapr/client/domain/StateOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
*/
package io.dapr.client.domain;

import com.fasterxml.jackson.annotation.JsonValue;
import io.dapr.utils.DurationUtils;

import java.time.Duration;
Expand Down Expand Up @@ -70,9 +71,11 @@ private Consistency(String value) {
this.value = value;
}

@JsonValue
public String getValue() {
return this.value;
}

}

public static enum Concurrency {
Expand All @@ -85,9 +88,11 @@ private Concurrency(String value) {
this.value = value;
}

@JsonValue
public String getValue() {
return this.value;
}

}

public static class RetryPolicy {
Expand Down
Loading