dapr
/
java-sdk
mirror of https://github.com/dapr/java-sdk.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Support W3C tracing + example. (#320)

This commit is contained in:
Artur Souza 2020-08-25 11:20:33 -07:00 committed by GitHub
parent 811728133d
commit eede480403
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
57 changed files with 2573 additions and 662 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
README.md
View File

@ -140,6 +140,7 @@ Try the following examples to learn more about Dapr's Java SDK:
* [PubSub with subscriber over Http](./examples/src/main/java/io/dapr/examples/pubsub/http)
* [Binding with input over Http](./examples/src/main/java/io/dapr/examples/bindings/http)
* [Actors over Http](./examples/src/main/java/io/dapr/examples/actors/http)
* [Distributed tracing with OpenTelemetry SDK](./examples/src/main/java/io/dapr/examples/tracing)
#### API Documentation

15
examples/pom.xml
View File

@ -73,6 +73,21 @@
<artifactId>spring-boot-autoconfigure</artifactId>
<version>2.2.2.RELEASE</version>
</dependency>
<dependency>
<groupId>io.opentelemetry</groupId>
<artifactId>opentelemetry-sdk</artifactId>
<version>0.7.0</version>
</dependency>
<dependency>
<groupId>io.opentelemetry</groupId>
<artifactId>opentelemetry-exporters-logging</artifactId>
<version>0.7.0</version>
</dependency>
<dependency>
<groupId>io.opentelemetry</groupId>
<artifactId>opentelemetry-exporters-zipkin</artifactId>
<version>0.7.0</version>
</dependency>
<dependency>
<groupId>io.dapr</groupId>
<artifactId>dapr-sdk-springboot</artifactId>

2
examples/src/main/java/io/dapr/examples/invoke/grpc/HelloWorldClient.java
View File

@ -7,7 +7,7 @@ package io.dapr.examples.invoke.grpc;
import io.dapr.client.DaprClient;
import io.dapr.client.DaprClientBuilder;
import io.dapr.client.HttpExtension;
import io.dapr.client.domain.HttpExtension;
import java.io.IOException;

2
examples/src/main/java/io/dapr/examples/invoke/http/InvokeClient.java
View File

@ -7,7 +7,7 @@ package io.dapr.examples.invoke.http;
import io.dapr.client.DaprClient;
import io.dapr.client.DaprClientBuilder;
import io.dapr.client.HttpExtension;
import io.dapr.client.domain.HttpExtension;
import java.io.IOException;

2
examples/src/main/java/io/dapr/examples/invoke/http/README.md
View File

@ -131,7 +131,7 @@ The class knows the app id for the remote application. It uses the the static `D
Execute the follow script in order to run the InvokeClient example, passing two messages for the remote method:
```sh
dapr run --components-path ./components --dapr-http-port 3006 -- java -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.invoke.http.InvokeClient 'message one' 'message two'
dapr run --components-path ./components --dapr-http-port 3006 -- java -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.invoke.http.InvokeClient "message one" "message two"
```
Once running, the output should display the messages sent from invoker in the demo service output as follows:

79
examples/src/main/java/io/dapr/examples/tracing/InvokeClient.java Normal file
View File

@ -0,0 +1,79 @@
/*
* Copyright (c) Microsoft Corporation.
* Licensed under the MIT License.
*/
package io.dapr.examples.tracing;
import io.dapr.client.DaprClient;
import io.dapr.client.DaprClientBuilder;
import io.dapr.client.domain.HttpExtension;
import io.dapr.client.domain.InvokeServiceRequest;
import io.dapr.client.domain.InvokeServiceRequestBuilder;
import io.dapr.springboot.OpenTelemetryConfig;
import io.dapr.utils.TypeRef;
import io.grpc.Context;
import io.opentelemetry.context.Scope;
import io.opentelemetry.sdk.OpenTelemetrySdk;
import io.opentelemetry.trace.Span;
import io.opentelemetry.trace.Tracer;
import java.io.IOException;
/**
* 1. Build and install jars:
* mvn clean install
* 2. Send messages to the server:
* dapr run --components-path ./examples/components \
* --port 3006 -- java -jar examples/target/dapr-java-sdk-examples-exec.jar \
* io.dapr.examples.tracing.InvokeClient 'message one' 'message two'
*/
public class InvokeClient {
/**
* Identifier in Dapr for the service this client will invoke.
*/
private static final String SERVICE_APP_ID = "tracingdemo";
/**
* Starts the invoke client.
*
* @param args Messages to be sent as request for the invoke API.
*/
public static void main(String[] args) throws IOException {
Tracer tracer = OpenTelemetryConfig.createTracer(InvokeClient.class.getCanonicalName());
Span span = tracer.spanBuilder("Example's Main").setSpanKind(Span.Kind.CLIENT).startSpan();
try (DaprClient client = (new DaprClientBuilder()).build()) {
for (String message : args) {
try (Scope scope = tracer.withSpan(span)) {
InvokeServiceRequestBuilder builder = new InvokeServiceRequestBuilder(SERVICE_APP_ID, "echo");
InvokeServiceRequest request
= builder.withBody(message).withHttpExtension(HttpExtension.POST).withContext(Context.current()).build();
client.invokeService(request, TypeRef.get(byte[].class))
.map(r -> {
System.out.println(new String(r.getObject()));
return r;
})
.flatMap(r -> {
InvokeServiceRequest sleepRequest = new InvokeServiceRequestBuilder(SERVICE_APP_ID, "sleep")
.withHttpExtension(HttpExtension.POST)
.withContext(r.getContext()).build();
return client.invokeService(sleepRequest, TypeRef.get(Void.class));
}).block();
}
}
// This is an example, so for simplicity we are just exiting here.
// Normally a dapr app would be a web service and not exit main.
System.out.println("Done");
}
span.end();
shutdown();
}
private static void shutdown() {
OpenTelemetrySdk.getTracerProvider().shutdown();
}
}

200
examples/src/main/java/io/dapr/examples/tracing/README.md Normal file
View File

@ -0,0 +1,200 @@
# Distributed Tracing Sample
In this sample, we'll create two java applications: a service application, which exposes two methods, and a client application which will invoke the methods from the service using Dapr.
This sample includes:
* TracingDemoService (Exposes the method to be remotely accessed)
* InvokeClient (Invokes the exposed methods from TracingDemoService)
Also consider [getting started with observability in Dapr](https://github.com/dapr/quickstarts/tree/master/observability).
## Remote invocation using the Java-SDK
This sample uses the Client provided in Dapr Java SDK invoking a remote method and Zipkin to collect and display tracing data.
## Pre-requisites
* [Dapr and Dapr CLI](https://github.com/dapr/docs/blob/master/getting-started/environment-setup.md#environment-setup).
* Java JDK 11 (or greater): [Oracle JDK](https://www.oracle.com/technetwork/java/javase/downloads/index.html#JDK11) or [OpenJDK](https://jdk.java.net/13/).
* [Apache Maven](https://maven.apache.org/install.html) version 3.x.
* [Configure Redis](https://github.com/dapr/docs/tree/master/howto/configure-redis) as a state store for Dapr.
### Checking out the code
Clone this repository:
```sh
git clone https://github.com/dapr/java-sdk.git
cd java-sdk
```
Then build the Maven project:
```sh
# make sure you are in the `java-sdk` directory.
mvn install
```
Then get into the examples directory:
```sh
cd examples
```
### Verify Zipkin is running
Run `docker ps` to see if the container `dapr_zipkin` is running locally:
```bash
CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES
24d043379da2 daprio/dapr "./placement" 2 days ago Up 32 hours 0.0.0.0:6050->50005/tcp dapr_placement
5779a0268159 openzipkin/zipkin "/busybox/sh run.sh" 2 days ago Up 32 hours 9410/tcp, 0.0.0.0:9411->9411/tcp dapr_zipkin
317fef6a8297 redis "docker-entrypoint.s…" 2 days ago Up 32 hours 0.0.0.0:6379->6379/tcp dapr_redis
```
If Zipkin is not working, [install the newest version of Dapr Cli and initialize it](https://github.com/dapr/cli#install-dapr-on-your-local-machine-self-hosted).
### Running the Demo service sample
The Demo service application exposes two methods that can be remotely invoked. In this example, the service code has two parts:
In the `TracingDemoService.java` file, you will find the `TracingDemoService` class, containing the main method. The main method uses the Spring Boot´s DaprApplication class for initializing the `TracingDemoServiceController`. See the code snippet below:
```java
public class TracingDemoService {
///...
public static void main(String[] args) throws Exception {
///...
// If port string is not valid, it will throw an exception.
int port = Integer.parseInt(cmd.getOptionValue("port"));
DaprApplication.start(port);
}
}
```
`DaprApplication.start()` Method will run a Spring Boot application that registers the `TracingDemoServiceController`, which exposes the invoking actions as POST requests. The Dapr's sidecar is the one that performs the actual call to the controller, triggered by client invocations or [bindings](https://github.com/dapr/docs/blob/master/concepts/bindings/README.md).
This Rest Controller exposes the `echo` and `sleep` methods. The `echo` method retrieves metadata from the headers and prints them along with the current date in console. The actual response from method is the formatted current date. See the code snippet below:
```java
@RestController
public class TracingDemoServiceController {
///...
@PostMapping(path = "/echo")
public Mono<String> handleMethod(@RequestBody(required = false) byte[] body,
@RequestHeader Map<String, String> headers) {
return Mono.fromSupplier(() -> {
try {
String message = body == null ? "" : new String(body, StandardCharsets.UTF_8);
Calendar utcNow = Calendar.getInstance(TimeZone.getTimeZone("GMT"));
String utcNowAsString = DATE_FORMAT.format(utcNow.getTime());
String metadataString = headers == null ? "" : OBJECT_MAPPER.writeValueAsString(headers);
// Handles the request by printing message.
System.out.println(
"Server: " + message + " @ " + utcNowAsString + " and metadata: " + metadataString);
return utcNowAsString;
} catch (Exception e) {
throw new RuntimeException(e);
}
});
}
}
```
The `sleep` methods simply waits for one second to simulate a slow operation.
```java
@RestController
public class TracingDemoServiceController {
///...
@PostMapping(path = "/sleep")
public void sleep() throws Exception {
// Simulate slow processing for metrics.
Thread.sleep(1000);
}
//...
}
```
The instrumentation for the service happens via the `OpenTelemetryIterceptor` class. This class uses the [OpenTelemetrySDK](https://github.com/open-telemetry/opentelemetry-java) for Java.
Use the follow command to execute the service:
```sh
dapr run --components-path ./components --app-id tracingdemo --app-port 3000 -- java -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.tracing.TracingDemoService -p 3000
```
Once running, the TracingDemoService is now ready to be invoked by Dapr.
### Running the InvokeClient sample
This sample code uses the Dapr SDK for invoking two remote methods (`echo` and `sleep`). It is also instrumented with OpenTelemetry. See the code snippet below:
```java
public class InvokeClient {
private static final String SERVICE_APP_ID = "invokedemo";
///...
public static void main(String[] args) throws IOException {
Tracer tracer = OpenTelemetryConfig.createTracer(InvokeClient.class.getCanonicalName());
Span span = tracer.spanBuilder("Example's Main").setSpanKind(Span.Kind.CLIENT).startSpan();
try (DaprClient client = (new DaprClientBuilder()).build()) {
for (String message : args) {
try (Scope scope = tracer.withSpan(span)) {
InvokeServiceRequestBuilder builder = new InvokeServiceRequestBuilder(SERVICE_APP_ID, "echo");
InvokeServiceRequest request
= builder.withBody(message).withHttpExtension(HttpExtension.POST).withContext(Context.current()).build();
client.invokeService(request, TypeRef.get(byte[].class))
.map(r -> {
System.out.println(new String(r.getObject()));
return r;
})
.flatMap(r -> {
InvokeServiceRequest sleepRequest = new InvokeServiceRequestBuilder(SERVICE_APP_ID, "sleep")
.withHttpExtension(HttpExtension.POST)
.withContext(r.getContext()).build();
return client.invokeService(sleepRequest, TypeRef.get(Void.class));
}).block();
}
}
// This is an example, so for simplicity we are just exiting here.
// Normally a dapr app would be a web service and not exit main.
System.out.println("Done");
}
span.end();
shutdown();
}
///...
}
```
The class knows the app id for the remote application. It uses `invokeService` method to invoke API calls on the service endpoint. The request object includes an instance of `io.grpc.Context` for the proper tracing headers to be propagated.
Execute the follow script in order to run the InvokeClient example, passing two messages for the remote method:
```sh
dapr run --components-path ./components -- java -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.tracing.InvokeClient "message one" "message two"
```
Once running, the output should display the messages sent from invoker in the demo service output as follows:
![exposeroutput](../../../../../../resources/img/exposer-service.png)
Method have been remotely invoked and displaying the remote messages.
Now, open Zipkin on [http://localhost:9411/zipkin](http://localhost:9411/zipkin). You should see a screen like the one below:
![zipking-landing](../../../../../../resources/img/zipkin-landing.png)
Click on the search icon to see the latest query results. You should see a tracing diagram similar to the one below:
![zipking-landing](../../../../../../resources/img/zipkin-tracing.png)
Once you click on the tracing event, you will see the details of the call stack starting in the client and then showing the service API calls right below.
![zipking-details](../../../../../../resources/img/zipkin-details.png)

45
examples/src/main/java/io/dapr/examples/tracing/TracingDemoService.java Normal file
View File

@ -0,0 +1,45 @@
/*
* Copyright (c) Microsoft Corporation.
* Licensed under the MIT License.
*/
package io.dapr.examples.tracing;
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;
/**
* Main method to invoke DemoService to test tracing.
*
* <p>Instrumentation is handled in {@link io.dapr.springboot.OpenTelemetryInterceptor}.
*
* <p>1. Build and install jars:
* mvn clean install
* 2. Run in server mode:
* dapr run --components-path ./examples/components --app-id tracingdemo --app-port 3000 --port 3005 \
* -- java -jar examples/target/dapr-java-sdk-examples-exec.jar \
* io.dapr.examples.tracing.TracingDemoService -p 3000
*/
public class TracingDemoService {
/**
* Starts the service.
* @param args Expects the port: -p PORT
* @throws Exception If cannot start service.
*/
public static void main(String[] args) throws Exception {
Options options = new Options();
options.addRequiredOption("p", "port", true, "Port to 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"));
DaprApplication.start(port);
}
}

87
examples/src/main/java/io/dapr/examples/tracing/TracingDemoServiceController.java Normal file
View File

@ -0,0 +1,87 @@
/*
* Copyright (c) Microsoft Corporation.
* Licensed under the MIT License.
*/
package io.dapr.examples.tracing;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.opentelemetry.trace.Tracer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestHeader;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Mono;
import java.nio.charset.StandardCharsets;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Map;
import java.util.TimeZone;
/**
* SpringBoot Controller to handle service invocation.
*
* <p>Instrumentation is handled in {@link io.dapr.springboot.OpenTelemetryInterceptor}.
*/
@RestController
public class TracingDemoServiceController {
/**
* Json serializer/deserializer.
*/
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
/**
* Format to output date and time.
*/
private static final DateFormat DATE_FORMAT = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
/**
* OpenTelemetry's tracer.
*/
@Autowired
private Tracer tracer;
/**
* Handles the 'echo' method invocation.
*
* @param body The body of the http message.
* @param headers The headers of the http message.
* @return A message containing the time.
*/
@PostMapping(path = "/echo")
public Mono<String> handleMethod(@RequestBody(required = false) byte[] body,
@RequestHeader Map<String, String> headers) {
return Mono.fromSupplier(() -> {
try {
String message = body == null ? "" : new String(body, StandardCharsets.UTF_8);
Calendar utcNow = Calendar.getInstance(TimeZone.getTimeZone("GMT"));
String utcNowAsString = DATE_FORMAT.format(utcNow.getTime());
String metadataString = headers == null ? "" : OBJECT_MAPPER.writeValueAsString(headers);
// Handles the request by printing message.
System.out.println(
"Server: " + message + " @ " + utcNowAsString + " and metadata: " + metadataString);
return utcNowAsString;
} catch (Exception e) {
throw new RuntimeException(e);
}
});
}
/**
* Handles the 'sleep' method invocation.
*/
@PostMapping(path = "/sleep")
public void sleep() throws Exception {
// Simulate slow processing for metrics.
Thread.sleep(1000);
}
}

1
examples/src/main/java/io/dapr/springboot/DaprMainApplication.java
View File

@ -27,7 +27,6 @@ public class DaprMainApplication {
Object[] methodArgs = new Object[1];
methodArgs[0] = arguments;
mainMethod.invoke(mainClass, methodArgs);
}
}
}

75
examples/src/main/java/io/dapr/springboot/OpenTelemetryConfig.java Normal file
View File

@ -0,0 +1,75 @@
/*
* Copyright (c) Microsoft Corporation.
* Licensed under the MIT License.
*/
package io.dapr.springboot;
import io.dapr.examples.invoke.http.InvokeClient;
import io.opentelemetry.OpenTelemetry;
import io.opentelemetry.exporters.logging.LoggingSpanExporter;
import io.opentelemetry.exporters.zipkin.ZipkinSpanExporter;
import io.opentelemetry.sdk.OpenTelemetrySdk;
import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
import io.opentelemetry.trace.Tracer;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.boot.autoconfigure.orm.jpa.HibernateJpaAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.io.IOException;
import java.net.Socket;
@Configuration
@EnableAutoConfiguration(exclude = {DataSourceAutoConfiguration.class, HibernateJpaAutoConfiguration.class})
public class OpenTelemetryConfig {
private static final int ZIPKIN_PORT = 9411;
private static final String ENDPOINT_V2_SPANS = "/api/v2/spans";
@Bean
public Tracer initTracer() throws Exception {
return createTracer("io.dapr.examples");
}
/**
* Creates an OpenTelemetry's tracer.
* @param instrumentationName Name of the instrumentation.
* @return New tracer's instance.
*/
public static Tracer createTracer(String instrumentationName) {
final Tracer tracer = OpenTelemetry.getTracer(instrumentationName);
// Only exports to Zipkin if it is up. Otherwise, ignore it.
// This is helpful to avoid exceptions for examples that do not require Zipkin.
if (isZipkinUp()) {
String httpUrl = String.format("http://localhost:%d", ZIPKIN_PORT);
ZipkinSpanExporter zipkinExporter =
ZipkinSpanExporter.newBuilder()
.setEndpoint(httpUrl + ENDPOINT_V2_SPANS)
.setServiceName(InvokeClient.class.getName())
.build();
OpenTelemetrySdk.getTracerProvider()
.addSpanProcessor(SimpleSpanProcessor.newBuilder(zipkinExporter).build());
} else {
System.out.println("WARNING: Zipkin is not available.");
}
final LoggingSpanExporter loggingExporter = new LoggingSpanExporter();
OpenTelemetrySdk.getTracerProvider()
.addSpanProcessor(SimpleSpanProcessor.newBuilder(loggingExporter).build());
return tracer;
}
private static boolean isZipkinUp() {
try (Socket ignored = new Socket("localhost", ZIPKIN_PORT)) {
return true;
} catch (IOException ignored) {
return false;
}
}
}

91
examples/src/main/java/io/dapr/springboot/OpenTelemetryInterceptor.java Normal file
View File

@ -0,0 +1,91 @@
/*
* Copyright (c) Microsoft Corporation.
* Licensed under the MIT License.
*/
package io.dapr.springboot;
import io.grpc.Context;
import io.opentelemetry.OpenTelemetry;
import io.opentelemetry.context.propagation.HttpTextFormat;
import io.opentelemetry.trace.Span;
import io.opentelemetry.trace.Tracer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.HandlerInterceptor;
import org.springframework.web.servlet.ModelAndView;
import javax.servlet.DispatcherType;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
@Component
public class OpenTelemetryInterceptor implements HandlerInterceptor {
@Autowired
Tracer tracer;
@Override
public boolean preHandle(
HttpServletRequest request, HttpServletResponse response, Object handler) {
final HttpTextFormat textFormat = OpenTelemetry.getPropagators().getHttpTextFormat();
// preHandle is called twice for asynchronous request. For more information, read:
// https://docs.spring.io/spring-framework/docs/current/javadoc-api/org/springframework/web/servlet/AsyncHandlerInterceptor.html
if (request.getDispatcherType() == DispatcherType.ASYNC) {
return true;
}
Span span;
try {
Context context = textFormat.extract(
Context.current(),
request,
new HttpTextFormat.Getter<HttpServletRequest>() {
@Override
public String get(HttpServletRequest req, String key) {
return req.getHeader(key);
}
});
request.setAttribute("opentelemetry-context", context);
span = tracer.spanBuilder(request.getRequestURI()).setParent(context).startSpan();
span.setAttribute("handler", "pre");
} catch (Exception e) {
span = tracer.spanBuilder(request.getRequestURI()).startSpan();
span.setAttribute("handler", "pre");
span.addEvent(e.toString());
span.setAttribute("error", true);
}
request.setAttribute("opentelemetry-span", span);
return true;
}
@Override
public void postHandle(
HttpServletRequest request, HttpServletResponse response, Object handler,
ModelAndView modelAndView) throws Exception {
}
@Override
public void afterCompletion(HttpServletRequest request, HttpServletResponse response,
Object handler, Exception exception) {
Object contextObject = request.getAttribute("opentelemetry-context");
Object spanObject = request.getAttribute("opentelemetry-span");
if ((contextObject == null) || (spanObject == null)) {
return;
}
Context context = (Context) contextObject;
Span span = (Span) spanObject;
span.setAttribute("handler", "afterCompletion");
final HttpTextFormat textFormat = OpenTelemetry.getPropagators().getHttpTextFormat();
textFormat.inject(context, response,
new HttpTextFormat.Setter<HttpServletResponse>() {
@Override
public void set(HttpServletResponse response, String key, String value) {
response.addHeader(key, value);
}
});
span.end();
}
}

23
examples/src/main/java/io/dapr/springboot/OpenTelemetryInterceptorConfig.java Normal file
View File

@ -0,0 +1,23 @@
/*
* Copyright (c) Microsoft Corporation.
* Licensed under the MIT License.
*/
package io.dapr.springboot;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.config.annotation.InterceptorRegistry;
import org.springframework.web.servlet.config.annotation.WebMvcConfigurationSupport;
@Component
public class OpenTelemetryInterceptorConfig extends WebMvcConfigurationSupport {
@Autowired
OpenTelemetryInterceptor interceptor;
@Override
public void addInterceptors(InterceptorRegistry registry) {
registry.addInterceptor(interceptor);
}
}

BIN
examples/src/main/resources/img/zipkin-details.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 98 KiB

BIN
examples/src/main/resources/img/zipkin-landing.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 29 KiB

BIN
examples/src/main/resources/img/zipkin-result.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 61 KiB

2
sdk-actors/src/main/java/io/dapr/actors/client/DaprHttpClient.java
View File

@ -39,7 +39,7 @@ class DaprHttpClient implements DaprClient {
public Mono<byte[]> invokeActorMethod(String actorType, String actorId, String methodName, byte[] jsonPayload) {
String url = String.format(Constants.ACTOR_METHOD_RELATIVE_URL_FORMAT, actorType, actorId, methodName);
Mono<DaprHttp.Response> responseMono =
this.client.invokeApi(DaprHttp.HttpMethods.POST.name(), url, null, jsonPayload, null);
this.client.invokeApi(DaprHttp.HttpMethods.POST.name(), url, null, jsonPayload, null, null);
return responseMono.map(r -> r.getBody());
}

13
sdk-actors/src/main/java/io/dapr/actors/runtime/DaprHttpClient.java
View File

@ -36,7 +36,8 @@ class DaprHttpClient implements DaprClient {
@Override
public Mono<byte[]> getActorState(String actorType, String actorId, String keyName) {
String url = String.format(Constants.ACTOR_STATE_KEY_RELATIVE_URL_FORMAT, actorType, actorId, keyName);
Mono<DaprHttp.Response> responseMono = this.client.invokeApi(DaprHttp.HttpMethods.GET.name(), url, null, "", null);
Mono<DaprHttp.Response> responseMono =
this.client.invokeApi(DaprHttp.HttpMethods.GET.name(), url, null, "", null, null);
return responseMono.map(r -> {
if ((r.getStatusCode() != 200) && (r.getStatusCode() != 204)) {
throw new IllegalStateException(
@ -52,7 +53,7 @@ class DaprHttpClient implements DaprClient {
@Override
public Mono<Void> saveActorStateTransactionally(String actorType, String actorId, byte[] data) {
String url = String.format(Constants.ACTOR_STATE_RELATIVE_URL_FORMAT, actorType, actorId);
return this.client.invokeApi(DaprHttp.HttpMethods.PUT.name(), url, null, data, null).then();
return this.client.invokeApi(DaprHttp.HttpMethods.PUT.name(), url, null, data, null, null).then();
}
/**
@ -61,7 +62,7 @@ class DaprHttpClient implements DaprClient {
@Override
public Mono<Void> registerActorReminder(String actorType, String actorId, String reminderName, byte[] data) {
String url = String.format(Constants.ACTOR_REMINDER_RELATIVE_URL_FORMAT, actorType, actorId, reminderName);
return this.client.invokeApi(DaprHttp.HttpMethods.PUT.name(), url, null, data, null).then();
return this.client.invokeApi(DaprHttp.HttpMethods.PUT.name(), url, null, data, null, null).then();
}
/**
@ -70,7 +71,7 @@ class DaprHttpClient implements DaprClient {
@Override
public Mono<Void> unregisterActorReminder(String actorType, String actorId, String reminderName) {
String url = String.format(Constants.ACTOR_REMINDER_RELATIVE_URL_FORMAT, actorType, actorId, reminderName);
return this.client.invokeApi(DaprHttp.HttpMethods.DELETE.name(), url, null, null).then();
return this.client.invokeApi(DaprHttp.HttpMethods.DELETE.name(), url, null, null, null).then();
}
/**
@ -79,7 +80,7 @@ class DaprHttpClient implements DaprClient {
@Override
public Mono<Void> registerActorTimer(String actorType, String actorId, String timerName, byte[] data) {
String url = String.format(Constants.ACTOR_TIMER_RELATIVE_URL_FORMAT, actorType, actorId, timerName);
return this.client.invokeApi(DaprHttp.HttpMethods.PUT.name(), url, null, data, null).then();
return this.client.invokeApi(DaprHttp.HttpMethods.PUT.name(), url, null, data, null, null).then();
}
/**
@ -88,7 +89,7 @@ class DaprHttpClient implements DaprClient {
@Override
public Mono<Void> unregisterActorTimer(String actorType, String actorId, String timerName) {
String url = String.format(Constants.ACTOR_TIMER_RELATIVE_URL_FORMAT, actorType, actorId, timerName);
return this.client.invokeApi(DaprHttp.HttpMethods.DELETE.name(), url, null, null).then();
return this.client.invokeApi(DaprHttp.HttpMethods.DELETE.name(), url, null, null, null).then();
}
}

2
sdk-tests/src/test/java/io/dapr/it/actors/ActorTurnBasedConcurrencyIT.java
View File

@ -51,7 +51,7 @@ public class ActorTurnBasedConcurrencyIT extends BaseIT {
System.out.println("Invoking during cleanup");
try {
client.invokeApi(DaprHttp.HttpMethods.DELETE.name(), url, null, null).block();
client.invokeApi(DaprHttp.HttpMethods.DELETE.name(), url, null, null, null).block();
} catch(Exception e) {
// informational only
System.out.println("Caught " + e.toString());

2
sdk-tests/src/test/java/io/dapr/it/binding/http/BindingIT.java
View File

@ -8,7 +8,7 @@ package io.dapr.it.binding.http;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.dapr.client.DaprClient;
import io.dapr.client.DaprClientBuilder;
import io.dapr.client.HttpExtension;
import io.dapr.client.domain.HttpExtension;
import io.dapr.it.BaseIT;
import io.dapr.it.DaprRun;
import org.junit.Test;

2
sdk-tests/src/test/java/io/dapr/it/methodinvoke/http/MethodInvokeIT.java
View File

@ -3,7 +3,7 @@ package io.dapr.it.methodinvoke.http;
import io.dapr.client.DaprClient;
import io.dapr.client.DaprClientBuilder;
import io.dapr.client.DaprHttp;
import io.dapr.client.HttpExtension;
import io.dapr.client.domain.HttpExtension;
import io.dapr.it.BaseIT;
import io.dapr.it.DaprRun;
import org.junit.After;

2
sdk-tests/src/test/java/io/dapr/it/pubsub/http/PubSubIT.java
View File

@ -8,7 +8,7 @@ package io.dapr.it.pubsub.http;
import io.dapr.client.DaprClient;
import io.dapr.client.DaprClientBuilder;
import io.dapr.client.DaprHttp;
import io.dapr.client.HttpExtension;
import io.dapr.client.domain.HttpExtension;
import io.dapr.it.BaseIT;
import io.dapr.it.DaprRun;
import org.junit.Test;

12
sdk/pom.xml
View File

@ -50,7 +50,17 @@
<dependency>
<groupId>com.squareup.okhttp3</groupId>
<artifactId>okhttp</artifactId>
<version>4.2.1</version>
<version>4.8.1</version>
</dependency>
<dependency>
<groupId>io.opentelemetry</groupId>
<artifactId>opentelemetry-api</artifactId>
<version>0.7.0</version>
</dependency>
<dependency>
<groupId>io.opencensus</groupId>
<artifactId>opencensus-impl-core</artifactId>
<version>0.26.0</version>
</dependency>
<dependency>
<groupId>junit</groupId>

420
sdk/src/main/java/io/dapr/client/AbstractDaprClient.java Normal file
View File

@ -0,0 +1,420 @@
/*
* Copyright (c) Microsoft Corporation.
* Licensed under the MIT License.
*/
package io.dapr.client;
import com.google.protobuf.Any;
import com.google.protobuf.ByteString;
import io.dapr.client.domain.DeleteStateRequest;
import io.dapr.client.domain.DeleteStateRequestBuilder;
import io.dapr.client.domain.GetSecretRequest;
import io.dapr.client.domain.GetSecretRequestBuilder;
import io.dapr.client.domain.GetStateRequest;
import io.dapr.client.domain.GetStateRequestBuilder;
import io.dapr.client.domain.HttpExtension;
import io.dapr.client.domain.InvokeBindingRequest;
import io.dapr.client.domain.InvokeBindingRequestBuilder;
import io.dapr.client.domain.InvokeServiceRequest;
import io.dapr.client.domain.InvokeServiceRequestBuilder;
import io.dapr.client.domain.PublishEventRequest;
import io.dapr.client.domain.PublishEventRequestBuilder;
import io.dapr.client.domain.SaveStateRequest;
import io.dapr.client.domain.SaveStateRequestBuilder;
import io.dapr.client.domain.State;
import io.dapr.client.domain.StateOptions;
import io.dapr.serializer.DaprObjectSerializer;
import io.dapr.utils.TypeRef;
import io.dapr.v1.CommonProtos;
import io.dapr.v1.DaprProtos;
import reactor.core.publisher.Mono;
import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* Abstract class with convenient methods common between client implementations.
*
* @see io.dapr.client.DaprClient
* @see io.dapr.client.DaprClientGrpc
* @see io.dapr.client.DaprClientHttp
*/
abstract class AbstractDaprClient implements DaprClient {
/**
* A utility class for serialize and deserialize the transient objects.
*/
protected DaprObjectSerializer objectSerializer;
/**
* A utility class for serialize and deserialize state objects.
*/
protected DaprObjectSerializer stateSerializer;
/**
* Common constructor for implementations of this class.
*
* @param objectSerializer Serializer for transient request/response objects.
* @param stateSerializer Serializer for state objects.
* @see DaprClientBuilder
*/
AbstractDaprClient(
DaprObjectSerializer objectSerializer,
DaprObjectSerializer stateSerializer) {
this.objectSerializer = objectSerializer;
this.stateSerializer = stateSerializer;
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> publishEvent(String pubsubName, String topic, Object data) {
return this.publishEvent(pubsubName, topic, data, null);
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> publishEvent(String pubsubName, String topic, Object data, Map<String, String> metadata) {
PublishEventRequest req = new PublishEventRequestBuilder(pubsubName, topic, data).withMetadata(metadata).build();
return this.publishEvent(req).then();
}
/**
* {@inheritDoc}
*/
public <T> Mono<T> invokeService(
String appId,
String method,
Object request,
HttpExtension httpExtension,
Map<String, String> metadata,
TypeRef<T> type) {
InvokeServiceRequestBuilder builder = new InvokeServiceRequestBuilder(appId, method);
InvokeServiceRequest req = builder
.withBody(request)
.withHttpExtension(httpExtension)
.withMetadata(metadata)
.build();
return this.invokeService(req, type).map(r -> r.getObject());
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeService(
String appId,
String method,
Object request,
HttpExtension httpExtension,
Map<String, String> metadata,
Class<T> clazz) {
return this.invokeService(appId, method, request, httpExtension, metadata, TypeRef.get(clazz));
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeService(
String appId, String method, HttpExtension httpExtension, Map<String, String> metadata, TypeRef<T> type) {
return this.invokeService(appId, method, null, httpExtension, metadata, type);
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeService(
String appId, String method, HttpExtension httpExtension, Map<String, String> metadata, Class<T> clazz) {
return this.invokeService(appId, method, null, httpExtension, metadata, TypeRef.get(clazz));
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeService(String appId, String method, Object request, HttpExtension httpExtension,
TypeRef<T> type) {
return this.invokeService(appId, method, request, httpExtension, null, type);
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeService(String appId, String method, Object request, HttpExtension httpExtension,
Class<T> clazz) {
return this.invokeService(appId, method, request, httpExtension, null, TypeRef.get(clazz));
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> invokeService(String appId, String method, Object request, HttpExtension httpExtension) {
return this.invokeService(appId, method, request, httpExtension, null, TypeRef.BYTE_ARRAY).then();
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> invokeService(
String appId, String method, Object request, HttpExtension httpExtension, Map<String, String> metadata) {
return this.invokeService(appId, method, request, httpExtension, metadata, TypeRef.BYTE_ARRAY).then();
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> invokeService(
String appId, String method, HttpExtension httpExtension, Map<String, String> metadata) {
return this.invokeService(appId, method, null, httpExtension, metadata, TypeRef.BYTE_ARRAY).then();
}
/**
* {@inheritDoc}
*/
@Override
public Mono<byte[]> invokeService(
String appId, String method, byte[] request, HttpExtension httpExtension, Map<String, String> metadata) {
return this.invokeService(appId, method, request, httpExtension, metadata, TypeRef.BYTE_ARRAY);
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> invokeBinding(String name, String operation, Object data) {
return this.invokeBinding(name, operation, data, null, TypeRef.BYTE_ARRAY).then();
}
/**
* {@inheritDoc}
*/
@Override
public Mono<byte[]> invokeBinding(String name, String operation, byte[] data, Map<String, String> metadata) {
return this.invokeBinding(name, operation, data, metadata, TypeRef.BYTE_ARRAY);
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeBinding(String name, String operation, Object data, TypeRef<T> type) {
return this.invokeBinding(name, operation, data, null, type);
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeBinding(String name, String operation, Object data, Class<T> clazz) {
return this.invokeBinding(name, operation, data, null, TypeRef.get(clazz));
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeBinding(
String name, String operation, Object data, Map<String, String> metadata, TypeRef<T> type) {
InvokeBindingRequest request = new InvokeBindingRequestBuilder(name, operation)
.withData(data)
.withMetadata(metadata)
.build();
return this.invokeBinding(request, type).map(r -> r.getObject());
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeBinding(
String name, String operation, Object data, Map<String, String> metadata, Class<T> clazz) {
return this.invokeBinding(name, operation, data, metadata, TypeRef.get(clazz));
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<State<T>> getState(String stateStoreName, State<T> state, TypeRef<T> type) {
return this.getState(stateStoreName, state.getKey(), state.getEtag(), state.getOptions(), type);
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<State<T>> getState(String stateStoreName, State<T> state, Class<T> clazz) {
return this.getState(stateStoreName, state.getKey(), state.getEtag(), state.getOptions(), TypeRef.get(clazz));
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<State<T>> getState(String stateStoreName, String key, TypeRef<T> type) {
return this.getState(stateStoreName, key, null, null, type);
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<State<T>> getState(String stateStoreName, String key, Class<T> clazz) {
return this.getState(stateStoreName, key, null, null, TypeRef.get(clazz));
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<State<T>> getState(
String stateStoreName, String key, String etag, StateOptions options, TypeRef<T> type) {
GetStateRequest request = new GetStateRequestBuilder(stateStoreName, key)
.withEtag(etag)
.withStateOptions(options)
.build();
return this.getState(request, type).map(r -> r.getObject());
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<State<T>> getState(
String stateStoreName, String key, String etag, StateOptions options, Class<T> clazz) {
return this.getState(stateStoreName, key, etag, options, TypeRef.get(clazz));
}
private <T> State<T> buildStateKeyValue(
DaprProtos.GetStateResponse response,
String requestedKey,
StateOptions stateOptions,
TypeRef<T> type) throws IOException {
ByteString payload = response.getData();
byte[] data = payload == null ? null : payload.toByteArray();
T value = stateSerializer.deserialize(data, type);
String etag = response.getEtag();
String key = requestedKey;
return new State<>(value, key, etag, stateOptions);
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> saveStates(String stateStoreName, List<State<?>> states) {
SaveStateRequest request = new SaveStateRequestBuilder(stateStoreName)
.withStates(states)
.build();
return this.saveStates(request).then();
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> saveState(String stateStoreName, String key, Object value) {
return this.saveState(stateStoreName, key, null, value, null);
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> saveState(String stateStoreName, String key, String etag, Object value, StateOptions options) {
State<?> state = new State<>(value, key, etag, options);
return this.saveStates(stateStoreName, Arrays.asList(state));
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> deleteState(String stateStoreName, String key) {
return this.deleteState(stateStoreName, key, null, null);
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> deleteState(String stateStoreName, String key, String etag, StateOptions options) {
DeleteStateRequest request = new DeleteStateRequestBuilder(stateStoreName, key)
.withEtag(etag)
.withStateOptions(options)
.build();
return deleteState(request).then();
}
/**
* Builds the object io.dapr.{@link io.dapr.v1.DaprProtos.InvokeServiceRequest} to be send based on the parameters.
*
* @param httpExtension Object for HttpExtension
* @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 invocation
* @param <K> The Type of the Body
* @return The object to be sent as part of the invocation.
* @throws java.io.IOException If there's an issue serializing the request.
*/
private <K> DaprProtos.InvokeServiceRequest buildInvokeServiceRequest(
HttpExtension httpExtension, String appId, String method, K request) throws IOException {
if (httpExtension == null) {
throw new IllegalArgumentException("HttpExtension cannot be null. Use HttpExtension.NONE instead.");
}
CommonProtos.InvokeRequest.Builder requestBuilder = CommonProtos.InvokeRequest.newBuilder();
requestBuilder.setMethod(method);
if (request != null) {
byte[] byteRequest = objectSerializer.serialize(request);
Any data = Any.newBuilder().setValue(ByteString.copyFrom(byteRequest)).build();
requestBuilder.setData(data);
} else {
requestBuilder.setData(Any.newBuilder().build());
}
CommonProtos.HTTPExtension.Builder httpExtensionBuilder = CommonProtos.HTTPExtension.newBuilder();
httpExtensionBuilder.setVerb(CommonProtos.HTTPExtension.Verb.valueOf(httpExtension.getMethod().toString()))
.putAllQuerystring(httpExtension.getQueryString());
requestBuilder.setHttpExtension(httpExtensionBuilder.build());
DaprProtos.InvokeServiceRequest.Builder envelopeBuilder = DaprProtos.InvokeServiceRequest.newBuilder()
.setId(appId)
.setMessage(requestBuilder.build());
return envelopeBuilder.build();
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Map<String, String>> getSecret(String secretStoreName, String key, Map<String, String> metadata) {
GetSecretRequest request = new GetSecretRequestBuilder(secretStoreName, key)
.withMetadata(metadata)
.build();
return getSecret(request).map(r -> r.getObject() == null ? new HashMap<>() : r.getObject());
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Map<String, String>> getSecret(String secretStoreName, String secretName) {
return this.getSecret(secretStoreName, secretName, null);
}
}

113
sdk/src/main/java/io/dapr/client/DaprClient.java
View File

@ -5,6 +5,15 @@
package io.dapr.client;
import io.dapr.client.domain.DeleteStateRequest;
import io.dapr.client.domain.GetSecretRequest;
import io.dapr.client.domain.GetStateRequest;
import io.dapr.client.domain.HttpExtension;
import io.dapr.client.domain.InvokeBindingRequest;
import io.dapr.client.domain.InvokeServiceRequest;
import io.dapr.client.domain.PublishEventRequest;
import io.dapr.client.domain.Response;
import io.dapr.client.domain.SaveStateRequest;
import io.dapr.client.domain.State;
import io.dapr.client.domain.StateOptions;
import io.dapr.utils.TypeRef;
@ -43,23 +52,31 @@ public interface DaprClient extends Closeable {
Mono<Void> publishEvent(String pubsubName, String topic, Object data, Map<String, String> metadata);
/**
* Invoke a service with all possible parameters, using serialization.
* Publish an event.
*
* @param request the request for the publish event.
* @return a Mono plan of a Dapr's void response.
*/
Mono<Response<Void>> publishEvent(PublishEventRequest request);
/**
* Invoke a service method, using serialization.
*
* @param appId The Application ID where the service is.
* @param method The actual Method to be call in the application.
* @param request The request to be sent to invoke the service, use byte[] to skip serialization.
* @param httpExtension Additional fields that are needed if the receiving app is listening on
* HTTP, {@link HttpExtension#NONE} otherwise.
* HTTP, {@link io.dapr.client.domain.HttpExtension#NONE} otherwise.
* @param metadata Metadata (in GRPC) or headers (in HTTP) to be sent in request.
* @param type The Type needed as return for the call.
* @param <T> The Type of the return, use byte[] to skip serialization.
* @return A Mono Plan of type type.
* @return A Mono Plan of type T.
*/
<T> Mono<T> invokeService(String appId, String method, Object request, HttpExtension httpExtension,
Map<String, String> metadata, TypeRef<T> type);
/**
* Invoke a service with all possible parameters, using serialization.
* Invoke a service method, using serialization.
*
* @param appId The Application ID where the service is.
* @param method The actual Method to be call in the application.
@ -69,13 +86,13 @@ public interface DaprClient extends Closeable {
* @param metadata Metadata (in GRPC) or headers (in HTTP) to be sent in request.
* @param clazz The type needed as return for the call.
* @param <T> The Type of the return, use byte[] to skip serialization.
* @return A Mono Plan of type type.
* @return A Mono Plan of type T.
*/
<T> Mono<T> invokeService(String appId, String method, Object request, HttpExtension httpExtension,
Map<String, String> metadata, Class<T> clazz);
/**
* Invoke a service with all possible parameters, using serialization.
* Invoke a service method, using serialization.
*
* @param appId The Application ID where the service is.
* @param method The actual Method to be call in the application.
@ -84,12 +101,12 @@ public interface DaprClient extends Closeable {
* HTTP, {@link HttpExtension#NONE} otherwise.
* @param type The Type needed as return for the call.
* @param <T> The Type of the return, use byte[] to skip serialization.
* @return A Mono Plan of type type.
* @return A Mono Plan of type T.
*/
<T> Mono<T> invokeService(String appId, String method, Object request, HttpExtension httpExtension, TypeRef<T> type);
/**
* Invoke a service with all possible parameters, using serialization.
* Invoke a service method, using serialization.
*
* @param appId The Application ID where the service is.
* @param method The actual Method to be call in the application.
@ -98,12 +115,12 @@ public interface DaprClient extends Closeable {
* HTTP, {@link HttpExtension#NONE} otherwise.
* @param clazz The type needed as return for the call.
* @param <T> The Type of the return, use byte[] to skip serialization.
* @return A Mono Plan of type type.
* @return A Mono Plan of type T.
*/
<T> Mono<T> invokeService(String appId, String method, Object request, HttpExtension httpExtension, Class<T> clazz);
/**
* Invoke a service with all possible parameters, using serialization.
* Invoke a service method, using serialization.
*
* @param appId The Application ID where the service is.
* @param method The actual Method to be call in the application.
@ -112,13 +129,13 @@ public interface DaprClient extends Closeable {
* @param metadata Metadata (in GRPC) or headers (in HTTP) to be sent in request.
* @param type The Type needed as return for the call.
* @param <T> The Type of the return, use byte[] to skip serialization.
* @return A Mono Plan of type type.
* @return A Mono Plan of type T.
*/
<T> Mono<T> invokeService(String appId, String method, HttpExtension httpExtension, Map<String, String> metadata,
TypeRef<T> type);
/**
* Invoke a service with all possible parameters, using serialization.
* Invoke a service method, using serialization.
*
* @param appId The Application ID where the service is.
* @param method The actual Method to be call in the application.
@ -127,13 +144,13 @@ public interface DaprClient extends Closeable {
* @param metadata Metadata (in GRPC) or headers (in HTTP) to be sent in request.
* @param clazz The type needed as return for the call.
* @param <T> The Type of the return, use byte[] to skip serialization.
* @return A Mono Plan of type type.
* @return A Mono Plan of type T.
*/
<T> Mono<T> invokeService(String appId, String method, HttpExtension httpExtension, Map<String, String> metadata,
Class<T> clazz);
/**
* Invoke a service with all possible parameters, using serialization.
* Invoke a service method, using serialization.
*
* @param appId The Application ID where the service is.
* @param method The actual Method to be call in the application.
@ -141,37 +158,37 @@ public interface DaprClient extends Closeable {
* @param httpExtension Additional fields that are needed if the receiving app is listening on
* HTTP, {@link HttpExtension#NONE} otherwise.
* @param metadata Metadata (in GRPC) or headers (in HTTP) to be sent in request.
* @return A Mono Plan of type type.
* @return A Mono Plan of type Void.
*/
Mono<Void> invokeService(String appId, String method, Object request, HttpExtension httpExtension,
Map<String, String> metadata);
/**
* Invoke a service with all possible parameters, using serialization.
* Invoke a service method, using serialization.
*
* @param appId The Application ID where the service is.
* @param method The actual Method to be call in the application.
* @param request The request to be sent to invoke the service, use byte[] to skip serialization.
* @param httpExtension Additional fields that are needed if the receiving app is listening on
* HTTP, {@link HttpExtension#NONE} otherwise.
* @return A Mono Plan of type type.
* @return A Mono Plan of type Void.
*/
Mono<Void> invokeService(String appId, String method, Object request, HttpExtension httpExtension);
/**
* Invoke a service with all possible parameters, using serialization.
* Invoke a service method, using serialization.
*
* @param appId The Application ID where the service is.
* @param method The actual Method to be call in the application.
* @param httpExtension Additional fields that are needed if the receiving app is listening on
* HTTP, {@link HttpExtension#NONE} otherwise.
* @param metadata Metadata (in GRPC) or headers (in HTTP) to be sent in request.
* @return A Mono Plan of type type.
* @return A Mono Plan of type Void.
*/
Mono<Void> invokeService(String appId, String method, HttpExtension httpExtension, Map<String, String> metadata);
/**
* Invoke a service with all possible parameters, using serialization.
* Invoke a service method, without using serialization.
*
* @param appId The Application ID where the service is.
* @param method The actual Method to be call in the application.
@ -179,11 +196,21 @@ public interface DaprClient extends Closeable {
* @param httpExtension Additional fields that are needed if the receiving app is listening on
* HTTP, {@link HttpExtension#NONE} otherwise.
* @param metadata Metadata (in GRPC) or headers (in HTTP) to be sent in request.
* @return A Mono Plan of type type.
* @return A Mono Plan of type byte[].
*/
Mono<byte[]> invokeService(String appId, String method, byte[] request, HttpExtension httpExtension,
Map<String, String> metadata);
/**
* Invoke a service method.
*
* @param invokeServiceRequest Request object.
* @param type The Type needed as return for the call.
* @param <T> The Type of the return, use byte[] to skip serialization.
* @return A Mono Plan of type T.
*/
<T> Mono<Response<T>> invokeService(InvokeServiceRequest invokeServiceRequest, TypeRef<T> type);
/**
* Invokes a Binding operation.
*
@ -255,6 +282,16 @@ public interface DaprClient extends Closeable {
*/
<T> Mono<T> invokeBinding(String name, String operation, Object data, Map<String, String> metadata, Class<T> clazz);
/**
* Invokes a Binding operation.
*
* @param request The binding invocation request.
* @param type The type being returned.
* @param <T> The type of the return
* @return a Mono plan of type T.
*/
<T> Mono<Response<T>> invokeBinding(InvokeBindingRequest request, TypeRef<T> type);
/**
* Retrieve a State based on their key.
*
@ -312,6 +349,16 @@ public interface DaprClient extends Closeable {
*/
<T> Mono<State<T>> getState(String stateStoreName, String key, String etag, StateOptions options, TypeRef<T> type);
/**
* Retrieve a State based on their key.
*
* @param request The request to get state.
* @param type The Type of State needed as return.
* @param <T> The Type of the return.
* @return A Mono Plan for the requested State.
*/
<T> Mono<Response<State<T>>> getState(GetStateRequest request, TypeRef<T> type);
/**
* Retrieve a State based on their key.
*
@ -334,6 +381,14 @@ public interface DaprClient extends Closeable {
*/
Mono<Void> saveStates(String stateStoreName, List<State<?>> states);
/**
* Save/Update a list of states.
*
* @param request Request to save states.
* @return a Mono plan of type Void.
*/
Mono<Response<Void>> saveStates(SaveStateRequest request);
/**
* Save/Update a state.
*
@ -376,6 +431,14 @@ public interface DaprClient extends Closeable {
*/
Mono<Void> deleteState(String stateStoreName, String key, String etag, StateOptions options);
/**
* Delete a state.
*
* @param request Request to delete a state.
* @return a Mono plan of type Void.
*/
Mono<Response<Void>> deleteState(DeleteStateRequest request);
/**
* Fetches a secret from the configured vault.
*
@ -394,4 +457,12 @@ public interface DaprClient extends Closeable {
* @return Key-value pairs for the secret.
*/
Mono<Map<String, String>> getSecret(String secretStoreName, String secretName);
/**
* Fetches a secret from the configured vault.
*
* @param request Request to fetch secret.
* @return Key-value pairs for the secret.
*/
Mono<Response<Map<String, String>>> getSecret(GetSecretRequest request);
}

443
sdk/src/main/java/io/dapr/client/DaprClientGrpc.java
View File

@ -9,6 +9,15 @@ import com.google.common.util.concurrent.ListenableFuture;
import com.google.protobuf.Any;
import com.google.protobuf.ByteString;
import com.google.protobuf.Empty;
import io.dapr.client.domain.DeleteStateRequest;
import io.dapr.client.domain.GetSecretRequest;
import io.dapr.client.domain.GetStateRequest;
import io.dapr.client.domain.HttpExtension;
import io.dapr.client.domain.InvokeBindingRequest;
import io.dapr.client.domain.InvokeServiceRequest;
import io.dapr.client.domain.PublishEventRequest;
import io.dapr.client.domain.Response;
import io.dapr.client.domain.SaveStateRequest;
import io.dapr.client.domain.State;
import io.dapr.client.domain.StateOptions;
import io.dapr.serializer.DaprObjectSerializer;
@ -16,13 +25,31 @@ import io.dapr.utils.TypeRef;
import io.dapr.v1.CommonProtos;
import io.dapr.v1.DaprGrpc;
import io.dapr.v1.DaprProtos;
import io.grpc.CallOptions;
import io.grpc.Channel;
import io.grpc.ClientCall;
import io.grpc.ClientInterceptor;
import io.grpc.Context;
import io.grpc.ForwardingClientCall;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;
import io.opencensus.implcore.trace.propagation.PropagationComponentImpl;
import io.opencensus.implcore.trace.propagation.TraceContextFormat;
import io.opencensus.trace.SpanContext;
import io.opencensus.trace.propagation.BinaryFormat;
import io.opencensus.trace.propagation.SpanContextParseException;
import io.opencensus.trace.propagation.TextFormat;
import io.opentelemetry.OpenTelemetry;
import io.opentelemetry.context.propagation.HttpTextFormat;
import org.jetbrains.annotations.Nullable;
import reactor.core.publisher.Mono;
import java.io.Closeable;
import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
/**
* An adapter for the GRPC Client.
@ -30,7 +57,12 @@ import java.util.Map;
* @see io.dapr.v1.DaprGrpc
* @see io.dapr.client.DaprClient
*/
public class DaprClientGrpc implements DaprClient {
public class DaprClientGrpc extends AbstractDaprClient {
/**
* Binary formatter to generate grpc-trace-bin.
*/
private static final BinaryFormat OPENCENSUS_BINARY_FORMAT = new PropagationComponentImpl().getBinaryFormat();
/**
* The GRPC managed channel to be used.
@ -44,16 +76,6 @@ public class DaprClientGrpc implements DaprClient {
*/
private DaprGrpc.DaprFutureStub client;
/**
* A utitlity class for serialize and deserialize the transient objects.
*/
private DaprObjectSerializer objectSerializer;
/**
* A utitlity class for serialize and deserialize state objects.
*/
private DaprObjectSerializer stateSerializer;
/**
* Default access level constructor, in order to create an instance of this class use io.dapr.client.DaprClientBuilder
*
@ -68,10 +90,9 @@ public class DaprClientGrpc implements DaprClient {
DaprGrpc.DaprFutureStub futureClient,
DaprObjectSerializer objectSerializer,
DaprObjectSerializer stateSerializer) {
super(objectSerializer, stateSerializer);
this.channel = closeableChannel;
this.client = futureClient;
this.objectSerializer = objectSerializer;
this.stateSerializer = stateSerializer;
this.client = populateWithInterceptors(futureClient);
}
private CommonProtos.StateOptions.StateConsistency getGrpcStateConsistency(StateOptions options) {
@ -100,43 +121,48 @@ public class DaprClientGrpc implements DaprClient {
* {@inheritDoc}
*/
@Override
public Mono<Void> publishEvent(String pubsubName, String topic, Object data) {
return this.publishEvent(pubsubName, topic, data, null);
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> publishEvent(String pubsubName, String topic, Object data, Map<String, String> metadata) {
public Mono<Response<Void>> publishEvent(PublishEventRequest request) {
try {
// TODO: handle metadata.
String pubsubName = request.getPubsubName();
String topic = request.getTopic();
Object data = request.getData();
// TODO(artursouza): handle metadata.
// Map<String, String> metadata = request.getMetadata();
Context context = request.getContext();
DaprProtos.PublishEventRequest envelope = DaprProtos.PublishEventRequest.newBuilder()
.setTopic(topic)
.setPubsubName(pubsubName)
.setData(ByteString.copyFrom(objectSerializer.serialize(data))).build();
return Mono.fromCallable(() -> {
return Mono.fromCallable(wrap(context, () -> {
ListenableFuture<Empty> futureEmpty = client.publishEvent(envelope);
futureEmpty.get();
return null;
});
}));
} catch (Exception ex) {
return Mono.error(ex);
}
}
@Override
public <T> Mono<T> invokeService(String appId, String method, Object request, HttpExtension httpExtension,
Map<String, String> metadata, TypeRef<T> type) {
/**
* {@inheritDoc}
*/
public <T> Mono<Response<T>> invokeService(InvokeServiceRequest invokeServiceRequest, TypeRef<T> type) {
try {
String appId = invokeServiceRequest.getAppId();
String method = invokeServiceRequest.getMethod();
Object request = invokeServiceRequest.getBody();
HttpExtension httpExtension = invokeServiceRequest.getHttpExtension();
// TODO(artursouza): handle metadata once available in GRPC proto.
// Map<String, String> metadata = invokeServiceRequest.getMetadata();
Context context = invokeServiceRequest.getContext();
DaprProtos.InvokeServiceRequest envelope = buildInvokeServiceRequest(httpExtension, appId, method, request);
return Mono.fromCallable(() -> {
return Mono.fromCallable(wrap(context, () -> {
ListenableFuture<CommonProtos.InvokeResponse> futureResponse =
client.invokeService(envelope);
return objectSerializer.deserialize(futureResponse.get().getData().getValue().toByteArray(), type);
});
})).map(r -> new Response<>(context, r));
} catch (Exception ex) {
return Mono.error(ex);
}
@ -146,126 +172,13 @@ public class DaprClientGrpc implements DaprClient {
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeService(
String appId,
String method,
Object request,
HttpExtension httpExtension,
Map<String, String> metadata,
Class<T> clazz) {
return this.invokeService(appId, method, request, httpExtension, metadata, TypeRef.get(clazz));
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeService(
String appId, String method, HttpExtension httpExtension, Map<String, String> metadata, TypeRef<T> type) {
return this.invokeService(appId, method, null, httpExtension, metadata, type);
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeService(
String appId, String method, HttpExtension httpExtension, Map<String, String> metadata, Class<T> clazz) {
return this.invokeService(appId, method, null, httpExtension, metadata, TypeRef.get(clazz));
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeService(String appId, String method, Object request, HttpExtension httpExtension,
TypeRef<T> type) {
return this.invokeService(appId, method, request, httpExtension, null, type);
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeService(String appId, String method, Object request, HttpExtension httpExtension,
Class<T> clazz) {
return this.invokeService(appId, method, request, httpExtension, null, TypeRef.get(clazz));
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> invokeService(String appId, String method, Object request, HttpExtension httpExtension) {
return this.invokeService(appId, method, request, httpExtension, null, TypeRef.BYTE_ARRAY).then();
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> invokeService(
String appId, String method, Object request, HttpExtension httpExtension, Map<String, String> metadata) {
return this.invokeService(appId, method, request, httpExtension, metadata, TypeRef.BYTE_ARRAY).then();
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> invokeService(
String appId, String method, HttpExtension httpExtension, Map<String, String> metadata) {
return this.invokeService(appId, method, null, httpExtension, metadata, TypeRef.BYTE_ARRAY).then();
}
/**
* {@inheritDoc}
*/
@Override
public Mono<byte[]> invokeService(
String appId, String method, byte[] request, HttpExtension httpExtension, Map<String, String> metadata) {
return this.invokeService(appId, method, request, httpExtension, metadata, TypeRef.BYTE_ARRAY);
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> invokeBinding(String name, String operation, Object data) {
return this.invokeBinding(name, operation, data, null, TypeRef.BYTE_ARRAY).then();
}
/**
* {@inheritDoc}
*/
@Override
public Mono<byte[]> invokeBinding(String name, String operation, byte[] data, Map<String, String> metadata) {
return this.invokeBinding(name, operation, data, metadata, TypeRef.BYTE_ARRAY);
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeBinding(String name, String operation, Object data, TypeRef<T> type) {
return this.invokeBinding(name, operation, data, null, type);
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeBinding(String name, String operation, Object data, Class<T> clazz) {
return this.invokeBinding(name, operation, data, null, TypeRef.get(clazz));
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeBinding(
String name, String operation, Object data, Map<String, String> metadata, TypeRef<T> type) {
public <T> Mono<Response<T>> invokeBinding(InvokeBindingRequest request, TypeRef<T> type) {
try {
final String name = request.getName();
final String operation = request.getOperation();
final Object data = request.getData();
final Map<String, String> metadata = request.getMetadata();
final Context context = request.getContext();
if (name == null || name.trim().isEmpty()) {
throw new IllegalArgumentException("Binding name cannot be null or empty.");
}
@ -284,10 +197,10 @@ public class DaprClientGrpc implements DaprClient {
builder.putAllMetadata(metadata);
}
DaprProtos.InvokeBindingRequest envelope = builder.build();
return Mono.fromCallable(() -> {
return Mono.fromCallable(wrap(context, () -> {
ListenableFuture<DaprProtos.InvokeBindingResponse> futureResponse = client.invokeBinding(envelope);
return objectSerializer.deserialize(futureResponse.get().getData().toByteArray(), type);
});
})).map(r -> new Response<>(context, r));
} catch (Exception ex) {
return Mono.error(ex);
}
@ -297,50 +210,15 @@ public class DaprClientGrpc implements DaprClient {
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeBinding(
String name, String operation, Object data, Map<String, String> metadata, Class<T> clazz) {
return this.invokeBinding(name, operation, data, metadata, TypeRef.get(clazz));
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<State<T>> getState(String stateStoreName, State<T> state, TypeRef<T> type) {
return this.getState(stateStoreName, state.getKey(), state.getEtag(), state.getOptions(), type);
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<State<T>> getState(String stateStoreName, State<T> state, Class<T> clazz) {
return this.getState(stateStoreName, state.getKey(), state.getEtag(), state.getOptions(), TypeRef.get(clazz));
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<State<T>> getState(String stateStoreName, String key, TypeRef<T> type) {
return this.getState(stateStoreName, key, null, null, type);
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<State<T>> getState(String stateStoreName, String key, Class<T> clazz) {
return this.getState(stateStoreName, key, null, null, TypeRef.get(clazz));
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<State<T>> getState(
String stateStoreName, String key, String etag, StateOptions options, TypeRef<T> type) {
public <T> Mono<Response<State<T>>> getState(GetStateRequest request, TypeRef<T> type) {
try {
final String stateStoreName = request.getStateStoreName();
final String key = request.getKey();
final StateOptions options = request.getStateOptions();
// TODO(artursouza): handle etag once available in proto.
// String etag = request.getEtag();
final Context context = request.getContext();
if ((stateStoreName == null) || (stateStoreName.trim().isEmpty())) {
throw new IllegalArgumentException("State store name cannot be null or empty.");
}
@ -355,7 +233,7 @@ public class DaprClientGrpc implements DaprClient {
}
DaprProtos.GetStateRequest envelope = builder.build();
return Mono.fromCallable(() -> {
return Mono.fromCallable(wrap(context, () -> {
ListenableFuture<DaprProtos.GetStateResponse> futureResponse = client.getState(envelope);
DaprProtos.GetStateResponse response = null;
try {
@ -364,21 +242,12 @@ public class DaprClientGrpc implements DaprClient {
return null;
}
return buildStateKeyValue(response, key, options, type);
});
})).map(s -> new Response(context, s));
} catch (Exception ex) {
return Mono.error(ex);
}
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<State<T>> getState(
String stateStoreName, String key, String etag, StateOptions options, Class<T> clazz) {
return this.getState(stateStoreName, key, etag, options, TypeRef.get(clazz));
}
private <T> State<T> buildStateKeyValue(
DaprProtos.GetStateResponse response,
String requestedKey,
@ -396,8 +265,11 @@ public class DaprClientGrpc implements DaprClient {
* {@inheritDoc}
*/
@Override
public Mono<Void> saveStates(String stateStoreName, List<State<?>> states) {
public Mono<Response<Void>> saveStates(SaveStateRequest request) {
try {
final String stateStoreName = request.getStateStoreName();
final List<State<?>> states = request.getStates();
final Context context = request.getContext();
if ((stateStoreName == null) || (stateStoreName.trim().isEmpty())) {
throw new IllegalArgumentException("State store name cannot be null or empty.");
}
@ -406,16 +278,16 @@ public class DaprClientGrpc implements DaprClient {
for (State state : states) {
builder.addStates(buildStateRequest(state).build());
}
DaprProtos.SaveStateRequest request = builder.build();
DaprProtos.SaveStateRequest req = builder.build();
return Mono.fromCallable(() -> client.saveState(request)).flatMap(f -> {
return Mono.fromCallable(wrap(context, () -> client.saveState(req))).flatMap(f -> {
try {
f.get();
} catch (Exception ex) {
return Mono.error(ex);
}
return Mono.empty();
});
}).thenReturn(new Response<Void>(context, null));
} catch (Exception ex) {
return Mono.error(ex);
}
@ -453,33 +325,14 @@ public class DaprClientGrpc implements DaprClient {
* {@inheritDoc}
*/
@Override
public Mono<Void> saveState(String stateStoreName, String key, Object value) {
return this.saveState(stateStoreName, key, null, value, null);
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> saveState(String stateStoreName, String key, String etag, Object value, StateOptions options) {
State<?> state = new State<>(value, key, etag, options);
return this.saveStates(stateStoreName, Arrays.asList(state));
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> deleteState(String stateStoreName, String key) {
return this.deleteState(stateStoreName, key, null, null);
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> deleteState(String stateStoreName, String key, String etag, StateOptions options) {
public Mono<Response<Void>> deleteState(DeleteStateRequest request) {
try {
final String stateStoreName = request.getStateStoreName();
final String key = request.getKey();
final StateOptions options = request.getStateOptions();
final String etag = request.getEtag();
final Context context = request.getContext();
if ((stateStoreName == null) || (stateStoreName.trim().isEmpty())) {
throw new IllegalArgumentException("State store name cannot be null or empty.");
}
@ -510,15 +363,15 @@ public class DaprClientGrpc implements DaprClient {
builder.setOptions(optionBuilder.build());
}
DaprProtos.DeleteStateRequest request = builder.build();
return Mono.fromCallable(() -> client.deleteState(request)).flatMap(f -> {
DaprProtos.DeleteStateRequest req = builder.build();
return Mono.fromCallable(wrap(context, () -> client.deleteState(req))).flatMap(f -> {
try {
f.get();
} catch (Exception ex) {
return Mono.error(ex);
}
return Mono.empty();
});
}).thenReturn(new Response<>(context, null));
} catch (Exception ex) {
return Mono.error(ex);
}
@ -564,13 +417,17 @@ public class DaprClientGrpc implements DaprClient {
* {@inheritDoc}
*/
@Override
public Mono<Map<String, String>> getSecret(String secretStoreName, String secretName, Map<String, String> metadata) {
public Mono<Response<Map<String, String>>> getSecret(GetSecretRequest request) {
String secretStoreName = request.getSecretStoreName();
String key = request.getKey();
Map<String, String> metadata = request.getMetadata();
Context context = request.getContext();
try {
if ((secretStoreName == null) || (secretStoreName.trim().isEmpty())) {
throw new IllegalArgumentException("Secret store name cannot be null or empty.");
}
if ((secretName == null) || (secretName.trim().isEmpty())) {
throw new IllegalArgumentException("Secret name cannot be null or empty.");
if ((key == null) || (key.trim().isEmpty())) {
throw new IllegalArgumentException("Secret key cannot be null or empty.");
}
} catch (Exception e) {
return Mono.error(e);
@ -578,24 +435,16 @@ public class DaprClientGrpc implements DaprClient {
DaprProtos.GetSecretRequest.Builder requestBuilder = DaprProtos.GetSecretRequest.newBuilder()
.setStoreName(secretStoreName)
.setKey(secretName);
.setKey(key);
if (metadata != null) {
requestBuilder.putAllMetadata(metadata);
}
return Mono.fromCallable(() -> {
DaprProtos.GetSecretRequest request = requestBuilder.build();
ListenableFuture<DaprProtos.GetSecretResponse> future = client.getSecret(request);
return Mono.fromCallable(wrap(context, () -> {
DaprProtos.GetSecretRequest req = requestBuilder.build();
ListenableFuture<DaprProtos.GetSecretResponse> future = client.getSecret(req);
return future.get();
}).map(future -> future.getDataMap());
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Map<String, String>> getSecret(String secretStoreName, String secretName) {
return this.getSecret(secretStoreName, secretName, null);
})).map(future -> new Response<>(context, future.getDataMap()));
}
/**
@ -609,4 +458,86 @@ public class DaprClientGrpc implements DaprClient {
channel.close();
}
}
/**
* Populates GRPC client with interceptors.
*
* @param client GRPC client for Dapr.
* @return Client after adding interceptors.
*/
private static DaprGrpc.DaprFutureStub populateWithInterceptors(DaprGrpc.DaprFutureStub client) {
ClientInterceptor interceptor = new ClientInterceptor() {
@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
MethodDescriptor<ReqT, RespT> methodDescriptor,
CallOptions callOptions,
Channel channel) {
ClientCall<ReqT, RespT> clientCall = channel.newCall(methodDescriptor, callOptions);
return new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(clientCall) {
@Override
public void start(final Listener<RespT> responseListener, final Metadata headers) {
// Dapr only supports "grpc-trace-bin" for GRPC and OpenTelemetry SDK does not support that yet:
// https://github.com/open-telemetry/opentelemetry-specification/issues/639
// This should be the only use of OpenCensus SDK: populate "grpc-trace-bin".
Context context = Context.current();
SpanContext opencensusSpanContext = extractOpenCensusSpanContext(context);
if (opencensusSpanContext != null) {
byte[] grpcTraceBin = OPENCENSUS_BINARY_FORMAT.toByteArray(opencensusSpanContext);
headers.put(Metadata.Key.of("grpc-trace-bin", Metadata.BINARY_BYTE_MARSHALLER), grpcTraceBin);
}
super.start(responseListener, headers);
}
};
}
};
return client.withInterceptors(interceptor);
}
/**
* Extracts the context from OpenTelemetry and creates a SpanContext for OpenCensus.
*
* @param openTelemetryContext Context from OpenTelemetry.
* @return SpanContext for OpenCensus.
*/
private static SpanContext extractOpenCensusSpanContext(Context openTelemetryContext) {
Map<String, String> map = new HashMap<>();
OpenTelemetry.getPropagators().getHttpTextFormat().inject(
openTelemetryContext,
map,
new HttpTextFormat.Setter<Map<String, String>>() {
@Override
public void set(Map<String, String> map, String key, String value) {
if (map != null) {
map.put(key, value);
}
}
});
if (!map.containsKey("traceparent")) {
// Trying to extract context without this key will throw an "expected" exception, so we avoid it here.
return null;
}
try {
return new TraceContextFormat()
.extract(map, new TextFormat.Getter<Map<String, String>>() {
@Nullable
@Override
public String get(Map<String, String> map, String key) {
return map.get(key);
}
});
} catch (SpanContextParseException e) {
throw new RuntimeException(e);
}
}
private static <V> Callable<V> wrap(Context context, Callable<V> callable) {
if (context == null) {
return callable;
}
return context.wrap(callable);
}
}

325
sdk/src/main/java/io/dapr/client/DaprClientHttp.java
View File

@ -5,18 +5,27 @@
package io.dapr.client;
import io.dapr.client.domain.DeleteStateRequest;
import io.dapr.client.domain.GetSecretRequest;
import io.dapr.client.domain.GetStateRequest;
import io.dapr.client.domain.HttpExtension;
import io.dapr.client.domain.InvokeBindingRequest;
import io.dapr.client.domain.InvokeServiceRequest;
import io.dapr.client.domain.PublishEventRequest;
import io.dapr.client.domain.Response;
import io.dapr.client.domain.SaveStateRequest;
import io.dapr.client.domain.State;
import io.dapr.client.domain.StateOptions;
import io.dapr.serializer.DaprObjectSerializer;
import io.dapr.serializer.DefaultObjectSerializer;
import io.dapr.utils.Constants;
import io.dapr.utils.TypeRef;
import io.grpc.Context;
import reactor.core.publisher.Mono;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@ -30,7 +39,7 @@ import java.util.Optional;
* @see io.dapr.client.DaprHttp
* @see io.dapr.client.DaprClient
*/
public class DaprClientHttp implements DaprClient {
public class DaprClientHttp extends AbstractDaprClient {
/**
* Serializer for internal objects.
@ -44,16 +53,6 @@ public class DaprClientHttp implements DaprClient {
*/
private final DaprHttp client;
/**
* A utility class for serialize and deserialize customer's transient objects.
*/
private final DaprObjectSerializer objectSerializer;
/**
* A utility class for serialize and deserialize customer's state objects.
*/
private final DaprObjectSerializer stateSerializer;
/**
* Flag determining if object serializer's input and output is Dapr's default instead of user provided.
*/
@ -74,9 +73,8 @@ public class DaprClientHttp implements DaprClient {
* @see DefaultObjectSerializer
*/
DaprClientHttp(DaprHttp client, DaprObjectSerializer objectSerializer, DaprObjectSerializer stateSerializer) {
super(objectSerializer, stateSerializer);
this.client = client;
this.objectSerializer = objectSerializer;
this.stateSerializer = stateSerializer;
this.isObjectSerializerDefault = objectSerializer.getClass() == DefaultObjectSerializer.class;
this.isStateSerializerDefault = stateSerializer.getClass() == DefaultObjectSerializer.class;
}
@ -96,16 +94,14 @@ public class DaprClientHttp implements DaprClient {
* {@inheritDoc}
*/
@Override
public Mono<Void> publishEvent(String pubsubName, String topic, Object data) {
return this.publishEvent(pubsubName, topic, data, null);
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> publishEvent(String pubsubName, String topic, Object data, Map<String, String> metadata) {
public Mono<Response<Void>> publishEvent(PublishEventRequest request) {
try {
String pubsubName = request.getPubsubName();
String topic = request.getTopic();
Object data = request.getData();
Map<String, String> metadata = request.getMetadata();
Context context = request.getContext();
if (topic == null || topic.trim().isEmpty()) {
throw new IllegalArgumentException("Topic name cannot be null or empty.");
}
@ -115,7 +111,8 @@ public class DaprClientHttp implements DaprClient {
.append("/").append(topic);
byte[] serializedEvent = objectSerializer.serialize(data);
return this.client.invokeApi(
DaprHttp.HttpMethods.POST.name(), url.toString(), null, serializedEvent, metadata).then();
DaprHttp.HttpMethods.POST.name(), url.toString(), null, serializedEvent, metadata, context)
.thenReturn(new Response<>(context, null));
} catch (Exception ex) {
return Mono.error(ex);
}
@ -124,14 +121,14 @@ public class DaprClientHttp implements DaprClient {
/**
* {@inheritDoc}
*/
public <T> Mono<T> invokeService(
String appId,
String method,
Object request,
HttpExtension httpExtension,
Map<String, String> metadata,
TypeRef<T> type) {
public <T> Mono<Response<T>> invokeService(InvokeServiceRequest invokeServiceRequest, TypeRef<T> type) {
try {
final String appId = invokeServiceRequest.getAppId();
final String method = invokeServiceRequest.getMethod();
final Object request = invokeServiceRequest.getBody();
final Map<String, String> metadata = invokeServiceRequest.getMetadata();
final HttpExtension httpExtension = invokeServiceRequest.getHttpExtension();
final Context context = invokeServiceRequest.getContext();
if (httpExtension == null) {
throw new IllegalArgumentException("HttpExtension cannot be null. Use HttpExtension.NONE instead.");
}
@ -146,7 +143,7 @@ public class DaprClientHttp implements DaprClient {
String path = String.format("%s/%s/method/%s", Constants.INVOKE_PATH, appId, method);
byte[] serializedRequestBody = objectSerializer.serialize(request);
Mono<DaprHttp.Response> response = this.client.invokeApi(httMethod, path,
httpExtension.getQueryString(), serializedRequestBody, metadata);
httpExtension.getQueryString(), serializedRequestBody, metadata, context);
return response.flatMap(r -> {
try {
T object = objectSerializer.deserialize(r.getBody(), type);
@ -158,7 +155,7 @@ public class DaprClientHttp implements DaprClient {
} catch (Exception ex) {
return Mono.error(ex);
}
});
}).map(r -> new Response(context, r));
} catch (Exception ex) {
return Mono.error(ex);
}
@ -168,126 +165,13 @@ public class DaprClientHttp implements DaprClient {
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeService(
String appId,
String method,
Object request,
HttpExtension httpExtension,
Map<String, String> metadata,
Class<T> clazz) {
return this.invokeService(appId, method, request, httpExtension, metadata, TypeRef.get(clazz));
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeService(
String appId, String method, HttpExtension httpExtension, Map<String, String> metadata, TypeRef<T> type) {
return this.invokeService(appId, method, null, httpExtension, metadata, type);
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeService(
String appId, String method, HttpExtension httpExtension, Map<String, String> metadata, Class<T> clazz) {
return this.invokeService(appId, method, null, httpExtension, metadata, TypeRef.get(clazz));
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeService(String appId, String method, Object request, HttpExtension httpExtension,
TypeRef<T> type) {
return this.invokeService(appId, method, request, httpExtension, null, type);
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeService(String appId, String method, Object request, HttpExtension httpExtension,
Class<T> clazz) {
return this.invokeService(appId, method, request, httpExtension,null, TypeRef.get(clazz));
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> invokeService(String appId, String method, Object request, HttpExtension httpExtension) {
return this.invokeService(appId, method, request, httpExtension, null, TypeRef.BYTE_ARRAY).then();
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> invokeService(
String appId, String method, Object request, HttpExtension httpExtension, Map<String, String> metadata) {
return this.invokeService(appId, method, request, httpExtension, metadata, TypeRef.BYTE_ARRAY).then();
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> invokeService(
String appId, String method, HttpExtension httpExtension, Map<String, String> metadata) {
return this.invokeService(appId, method, null, httpExtension, metadata, TypeRef.BYTE_ARRAY).then();
}
/**
* {@inheritDoc}
*/
@Override
public Mono<byte[]> invokeService(
String appId, String method, byte[] request, HttpExtension httpExtension, Map<String, String> metadata) {
return this.invokeService(appId, method, request, httpExtension, metadata, TypeRef.BYTE_ARRAY);
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> invokeBinding(String name, String operation, Object data) {
return this.invokeBinding(name, operation, data, null, TypeRef.BYTE_ARRAY).then();
}
/**
* {@inheritDoc}
*/
@Override
public Mono<byte[]> invokeBinding(String name, String operation, byte[] data, Map<String, String> metadata) {
return this.invokeBinding(name, operation, data, metadata, TypeRef.BYTE_ARRAY);
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeBinding(String name, String operation, Object data, TypeRef<T> type) {
return this.invokeBinding(name, operation, data, null, type);
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeBinding(String name, String operation, Object data, Class<T> clazz) {
return this.invokeBinding(name, operation, data, null, TypeRef.get(clazz));
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeBinding(
String name, String operation, Object data, Map<String, String> metadata, TypeRef<T> type) {
public <T> Mono<Response<T>> invokeBinding(InvokeBindingRequest request, TypeRef<T> type) {
try {
final String name = request.getName();
final String operation = request.getOperation();
final Object data = request.getData();
final Map<String, String> metadata = request.getMetadata();
final Context context = request.getContext();
if (name == null || name.trim().isEmpty()) {
throw new IllegalArgumentException("Binding name cannot be null or empty.");
}
@ -326,7 +210,8 @@ public class DaprClientHttp implements DaprClient {
byte[] payload = INTERNAL_SERIALIZER.serialize(jsonMap);
String httpMethod = DaprHttp.HttpMethods.POST.name();
Mono<DaprHttp.Response> response = this.client.invokeApi(httpMethod, url.toString(), null, payload, null);
Mono<DaprHttp.Response> response = this.client.invokeApi(
httpMethod, url.toString(), null, payload, null, context);
return response.flatMap(r -> {
try {
T object = objectSerializer.deserialize(r.getBody(), type);
@ -338,61 +223,24 @@ public class DaprClientHttp implements DaprClient {
} catch (Exception ex) {
return Mono.error(ex);
}
});
}).map(r -> new Response<T>(context, r));
} catch (Exception ex) {
return Mono.error(ex);
}
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<T> invokeBinding(
String name, String operation, Object data, Map<String, String> metadata, Class<T> clazz) {
return this.invokeBinding(name, operation, data, metadata, TypeRef.get(clazz));
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<State<T>> getState(String stateStoreName, State<T> state, TypeRef<T> type) {
return this.getState(stateStoreName, state.getKey(), state.getEtag(), state.getOptions(), type);
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<State<T>> getState(String stateStoreName, State<T> state, Class<T> clazz) {
return this.getState(stateStoreName, state.getKey(), state.getEtag(), state.getOptions(), TypeRef.get(clazz));
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<State<T>> getState(String stateStoreName, String key, TypeRef<T> type) {
return this.getState(stateStoreName, key, null, null, type);
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<State<T>> getState(String stateStoreName, String key, Class<T> clazz) {
return this.getState(stateStoreName, key, null, null, TypeRef.get(clazz));
}
/**
* {@inheritDoc}
*/
@Override
public <T> Mono<State<T>> getState(
String stateStoreName, String key, String etag, StateOptions options, TypeRef<T> type) {
public <T> Mono<Response<State<T>>> getState(GetStateRequest request, TypeRef<T> type) {
try {
final String stateStoreName = request.getStateStoreName();
final String key = request.getKey();
final StateOptions options = request.getStateOptions();
final String etag = request.getEtag();
final Context context = request.getContext();
if ((stateStoreName == null) || (stateStoreName.trim().isEmpty())) {
throw new IllegalArgumentException("State store name cannot be null or empty.");
}
@ -414,31 +262,29 @@ public class DaprClientHttp implements DaprClient {
.orElse(new HashMap<>());
return this.client
.invokeApi(DaprHttp.HttpMethods.GET.name(), url.toString(), urlParameters, headers)
.invokeApi(DaprHttp.HttpMethods.GET.name(), url.toString(), urlParameters, headers, context)
.flatMap(s -> {
try {
return Mono.just(buildStateKeyValue(s, key, options, type));
} catch (Exception ex) {
return Mono.error(ex);
}
});
})
.map(r -> new Response<>(context, r));
} catch (Exception ex) {
return Mono.error(ex);
}
}
@Override
public <T> Mono<State<T>> getState(
String stateStoreName, String key, String etag, StateOptions options, Class<T> clazz) {
return null;
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> saveStates(String stateStoreName, List<State<?>> states) {
public Mono<Response<Void>> saveStates(SaveStateRequest request) {
try {
final String stateStoreName = request.getStateStoreName();
final List<State<?>> states = request.getStates();
final Context context = request.getContext();
if ((stateStoreName == null) || (stateStoreName.trim().isEmpty())) {
throw new IllegalArgumentException("State store name cannot be null or empty.");
}
@ -472,7 +318,8 @@ public class DaprClientHttp implements DaprClient {
}
byte[] serializedStateBody = INTERNAL_SERIALIZER.serialize(internalStateObjects);
return this.client.invokeApi(
DaprHttp.HttpMethods.POST.name(), url, null, serializedStateBody, headers).then();
DaprHttp.HttpMethods.POST.name(), url, null, serializedStateBody, headers, context)
.thenReturn(new Response<>(context, null));
} catch (Exception ex) {
return Mono.error(ex);
}
@ -482,33 +329,14 @@ public class DaprClientHttp implements DaprClient {
* {@inheritDoc}
*/
@Override
public Mono<Void> saveState(String stateStoreName, String key, Object value) {
return this.saveState(stateStoreName, key, null, value, null);
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> saveState(String stateStoreName, String key, String etag, Object value, StateOptions options) {
return Mono.fromSupplier(() -> new State<>(value, key, etag, options))
.flatMap(state -> saveStates(stateStoreName, Arrays.asList(state)));
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> deleteState(String stateStoreName, String key) {
return this.deleteState(stateStoreName, key, null, null);
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Void> deleteState(String stateStoreName, String key, String etag, StateOptions options) {
public Mono<Response<Void>> deleteState(DeleteStateRequest request) {
try {
final String stateStoreName = request.getStateStoreName();
final String key = request.getKey();
final StateOptions options = request.getStateOptions();
final String etag = request.getEtag();
final Context context = request.getContext();
if ((stateStoreName == null) || (stateStoreName.trim().isEmpty())) {
throw new IllegalArgumentException("State store name cannot be null or empty.");
}
@ -524,7 +352,9 @@ public class DaprClientHttp implements DaprClient {
.map(stateOptions -> stateOptions.getStateOptionsAsMap())
.orElse(new HashMap<>());
return this.client.invokeApi(DaprHttp.HttpMethods.DELETE.name(), url, urlParameters, headers).then();
return this.client.invokeApi(
DaprHttp.HttpMethods.DELETE.name(), url, urlParameters, headers, context)
.thenReturn(new Response<>(context, null));
} catch (Exception ex) {
return Mono.error(ex);
}
@ -556,21 +386,25 @@ public class DaprClientHttp implements DaprClient {
* {@inheritDoc}
*/
@Override
public Mono<Map<String, String>> getSecret(String secretStoreName, String secretName, Map<String, String> metadata) {
public Mono<Response<Map<String, String>>> getSecret(GetSecretRequest request) {
String secretStoreName = request.getSecretStoreName();
String key = request.getKey();
Map<String, String> metadata = request.getMetadata();
Context context = request.getContext();
try {
if ((secretStoreName == null) || (secretStoreName.trim().isEmpty())) {
throw new IllegalArgumentException("Secret store name cannot be null or empty.");
}
if ((secretName == null) || (secretName.trim().isEmpty())) {
throw new IllegalArgumentException("Secret name cannot be null or empty.");
if ((key == null) || (key.trim().isEmpty())) {
throw new IllegalArgumentException("Secret key cannot be null or empty.");
}
} catch (Exception e) {
return Mono.error(e);
}
String url = Constants.SECRETS_PATH + "/" + secretStoreName + "/" + secretName;
String url = Constants.SECRETS_PATH + "/" + secretStoreName + "/" + key;
return this.client
.invokeApi(DaprHttp.HttpMethods.GET.name(), url, metadata, (String)null, null)
.invokeApi(DaprHttp.HttpMethods.GET.name(), url, metadata, (String)null, null, context)
.flatMap(response -> {
try {
Map m = INTERNAL_SERIALIZER.deserialize(response.getBody(), Map.class);
@ -583,15 +417,8 @@ public class DaprClientHttp implements DaprClient {
return Mono.error(e);
}
})
.map(m -> (Map<String, String>)m);
}
/**
* {@inheritDoc}
*/
@Override
public Mono<Map<String, String>> getSecret(String secretStoreName, String secretName) {
return this.getSecret(secretStoreName, secretName, null);
.map(m -> (Map<String, String>)m)
.map(m -> new Response<>(context, m));
}
@Override

189
sdk/src/main/java/io/dapr/client/DaprHttp.java
View File

@ -9,11 +9,16 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import io.dapr.exceptions.DaprError;
import io.dapr.exceptions.DaprException;
import io.dapr.utils.Constants;
import io.grpc.Context;
import io.opentelemetry.OpenTelemetry;
import io.opentelemetry.context.propagation.HttpTextFormat;
import okhttp3.HttpUrl;
import okhttp3.MediaType;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.RequestBody;
import okhttp3.ResponseBody;
import org.jetbrains.annotations.NotNull;
import reactor.core.publisher.Mono;
import java.io.Closeable;
@ -32,6 +37,14 @@ public class DaprHttp implements Closeable {
*/
private static final String DEFAULT_HTTP_SCHEME = "http";
private static final HttpTextFormat.Setter<Request.Builder> OPENTELEMETRY_SETTER =
new HttpTextFormat.Setter<Request.Builder>() {
@Override
public void set(Request.Builder requestBuilder, String key, String value) {
requestBuilder.addHeader(key, value);
}
};
/**
* HTTP Methods supported.
*/
@ -128,11 +141,16 @@ public class DaprHttp implements Closeable {
* @param urlString url as String.
* @param urlParameters URL parameters
* @param headers HTTP headers.
* @param context OpenTelemetry's Context.
* @return Asynchronous text
*/
public Mono<Response> invokeApi(
String method, String urlString, Map<String, String> urlParameters, Map<String, String> headers) {
return this.invokeApi(method, urlString, urlParameters, (byte[]) null, headers);
String method,
String urlString,
Map<String, String> urlParameters,
Map<String, String> headers,
Context context) {
return this.invokeApi(method, urlString, urlParameters, (byte[]) null, headers, context);
}
/**
@ -143,15 +161,21 @@ public class DaprHttp implements Closeable {
* @param urlParameters Parameters in the URL
* @param content payload to be posted.
* @param headers HTTP headers.
* @param context OpenTelemetry's Context.
* @return Asynchronous response
*/
public Mono<Response> invokeApi(
String method, String urlString, Map<String, String> urlParameters, String content, Map<String, String> headers) {
String method,
String urlString,
Map<String, String> urlParameters,
String content,
Map<String, String> headers,
Context context) {
return this.invokeApi(
method, urlString, urlParameters, content == null
? EMPTY_BYTES
: content.getBytes(StandardCharsets.UTF_8), headers);
: content.getBytes(StandardCharsets.UTF_8), headers, context);
}
/**
@ -162,69 +186,104 @@ public class DaprHttp implements Closeable {
* @param urlParameters Parameters in the URL
* @param content payload to be posted.
* @param headers HTTP headers.
* @param context OpenTelemetry's Context.
* @return Asynchronous response
*/
public Mono<Response> invokeApi(
String method, String urlString, Map<String, String> urlParameters, byte[] content, Map<String, String> headers) {
return Mono.fromCallable(
() -> {
final String requestId = UUID.randomUUID().toString();
RequestBody body;
String method,
String urlString,
Map<String, String> urlParameters,
byte[] content,
Map<String, String> headers,
Context context) {
return Mono.fromCallable(() -> doInvokeApi(method, urlString, urlParameters, content, headers, context));
}
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);
}
HttpUrl.Builder urlBuilder = new HttpUrl.Builder();
urlBuilder.scheme(DEFAULT_HTTP_SCHEME)
.host(Constants.DEFAULT_HOSTNAME)
.port(this.port)
.addPathSegments(urlString);
Optional.ofNullable(urlParameters).orElse(Collections.emptyMap()).entrySet().stream()
.forEach(urlParameter -> urlBuilder.addQueryParameter(urlParameter.getKey(), urlParameter.getValue()));
/**
* Shutdown call is not necessary for OkHttpClient.
* @see OkHttpClient
*/
@Override
public void close() throws IOException {
// No code needed
}
Request.Builder requestBuilder = new Request.Builder()
.url(urlBuilder.build())
.addHeader(Constants.HEADER_DAPR_REQUEST_ID, requestId);
if (HttpMethods.GET.name().equals(method)) {
requestBuilder.get();
} else if (HttpMethods.DELETE.name().equals(method)) {
requestBuilder.delete();
} else {
requestBuilder.method(method, body);
}
if (headers != null) {
Optional.ofNullable(headers.entrySet()).orElse(Collections.emptySet()).stream()
.forEach(header -> {
requestBuilder.addHeader(header.getKey(), header.getValue());
});
}
/**
* Invokes an API that returns a text payload.
*
* @param method HTTP method.
* @param urlString url as String.
* @param urlParameters Parameters in the URL
* @param content payload to be posted.
* @param headers HTTP headers.
* @param context OpenTelemetry's Context.
* @return Response
*/
@NotNull
private Response doInvokeApi(String method,
String urlString,
Map<String, String> urlParameters,
byte[] content, Map<String, String> headers,
Context context) throws IOException {
final String requestId = UUID.randomUUID().toString();
RequestBody body;
Request request = requestBuilder.build();
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);
}
HttpUrl.Builder urlBuilder = new HttpUrl.Builder();
urlBuilder.scheme(DEFAULT_HTTP_SCHEME)
.host(Constants.DEFAULT_HOSTNAME)
.port(this.port)
.addPathSegments(urlString);
Optional.ofNullable(urlParameters).orElse(Collections.emptyMap()).entrySet().stream()
.forEach(urlParameter -> urlBuilder.addQueryParameter(urlParameter.getKey(), urlParameter.getValue()));
try (okhttp3.Response response = this.httpClient.newCall(request).execute()) {
if (!response.isSuccessful()) {
DaprError error = parseDaprError(response.body().bytes());
if ((error != null) && (error.getErrorCode() != null) && (error.getMessage() != null)) {
throw new DaprException(error);
}
Request.Builder requestBuilder = new Request.Builder()
.url(urlBuilder.build())
.addHeader(Constants.HEADER_DAPR_REQUEST_ID, requestId);
if (context != null) {
OpenTelemetry.getPropagators().getHttpTextFormat().inject(context, requestBuilder, OPENTELEMETRY_SETTER);
}
if (HttpMethods.GET.name().equals(method)) {
requestBuilder.get();
} else if (HttpMethods.DELETE.name().equals(method)) {
requestBuilder.delete();
} else {
requestBuilder.method(method, body);
}
if (headers != null) {
Optional.ofNullable(headers.entrySet()).orElse(Collections.emptySet()).stream()
.forEach(header -> {
requestBuilder.addHeader(header.getKey(), header.getValue());
});
}
throw new IllegalStateException("Unknown Dapr error. HTTP status code: " + response.code());
}
Request request = requestBuilder.build();
Map<String, String> mapHeaders = new HashMap<>();
byte[] result = response.body().bytes();
response.headers().forEach(pair -> {
mapHeaders.put(pair.getFirst(), pair.getSecond());
});
return new Response(result, mapHeaders, response.code());
}
});
try (okhttp3.Response response = this.httpClient.newCall(request).execute()) {
if (!response.isSuccessful()) {
DaprError error = parseDaprError(getBodyBytesOrEmptyArray(response));
if ((error != null) && (error.getErrorCode() != null) && (error.getMessage() != null)) {
throw new DaprException(error);
}
throw new IllegalStateException("Unknown Dapr error. HTTP status code: " + response.code());
}
Map<String, String> mapHeaders = new HashMap<>();
byte[] result = getBodyBytesOrEmptyArray(response);
response.headers().forEach(pair -> {
mapHeaders.put(pair.getFirst(), pair.getSecond());
});
return new Response(result, mapHeaders, response.code());
}
}
/**
@ -240,13 +299,13 @@ public class DaprHttp implements Closeable {
return OBJECT_MAPPER.readValue(json, DaprError.class);
}
/**
* Shutdown call is not necessary for OkHttpClient.
* @see OkHttpClient
*/
@Override
public void close() throws IOException {
// No code needed
private static byte[] getBodyBytesOrEmptyArray(okhttp3.Response response) throws IOException {
ResponseBody body = response.body();
if (body != null) {
return body.bytes();
}
return EMPTY_BYTES;
}
}

1
sdk/src/main/java/io/dapr/client/ObjectSerializer.java
View File

@ -13,7 +13,6 @@ import io.dapr.client.domain.CloudEvent;
import io.dapr.utils.TypeRef;
import java.io.IOException;
import java.lang.reflect.Type;
/**
* Serializes and deserializes an internal object.

64
sdk/src/main/java/io/dapr/client/domain/DeleteStateRequest.java Normal file
View File

@ -0,0 +1,64 @@
/*
* Copyright (c) Microsoft Corporation.
* Licensed under the MIT License.
*/
package io.dapr.client.domain;
import io.grpc.Context;
/**
* A request to delete a state by key.
*/
public class DeleteStateRequest {
private String stateStoreName;
private String key;
private String etag;
private StateOptions stateOptions;
private Context context;
public String getStateStoreName() {
return stateStoreName;
}
void setStateStoreName(String stateStoreName) {
this.stateStoreName = stateStoreName;
}
public String getKey() {
return key;
}
void setKey(String key) {
this.key = key;
}
public String getEtag() {
return etag;
}
void setEtag(String etag) {
this.etag = etag;
}
public StateOptions getStateOptions() {
return stateOptions;
}
void setStateOptions(StateOptions stateOptions) {
this.stateOptions = stateOptions;
}
public Context getContext() {
return context;
}
void setContext(Context context) {
this.context = context;
}
}

59
sdk/src/main/java/io/dapr/client/domain/DeleteStateRequestBuilder.java Normal file
View File

@ -0,0 +1,59 @@
/*
* Copyright (c) Microsoft Corporation.
* Licensed under the MIT License.
*/
package io.dapr.client.domain;
import io.grpc.Context;
/**
* Builds a request to delete a state by key.
*/
public class DeleteStateRequestBuilder {
private final String stateStoreName;
private final String key;
private String etag;
private StateOptions stateOptions;
private Context context;
public DeleteStateRequestBuilder(String stateStoreName, String key) {
this.stateStoreName = stateStoreName;
this.key = key;
}
public DeleteStateRequestBuilder withEtag(String etag) {
this.etag = etag;
return this;
}
public DeleteStateRequestBuilder withStateOptions(StateOptions stateOptions) {
this.stateOptions = stateOptions;
return this;
}
public DeleteStateRequestBuilder withContext(Context context) {
this.context = context;
return this;
}
/**
* Builds a request object.
* @return Request object.
*/
public DeleteStateRequest build() {
DeleteStateRequest request = new DeleteStateRequest();
request.setStateStoreName(this.stateStoreName);
request.setKey(this.key);
request.setEtag(this.etag);
request.setStateOptions(this.stateOptions);
request.setContext(this.context);
return request;
}
}

56
sdk/src/main/java/io/dapr/client/domain/GetSecretRequest.java Normal file
View File

@ -0,0 +1,56 @@
/*
* Copyright (c) Microsoft Corporation.
* Licensed under the MIT License.
*/
package io.dapr.client.domain;
import io.grpc.Context;
import java.util.Map;
/**
* A request to get a secret by key.
*/
public class GetSecretRequest {
private String secretStoreName;
private String key;
private Map<String, String> metadata;
private Context context;
public String getSecretStoreName() {
return secretStoreName;
}
void setSecretStoreName(String secretStoreName) {
this.secretStoreName = secretStoreName;
}
public String getKey() {
return key;
}
void setKey(String key) {
this.key = key;
}
public Map<String, String> getMetadata() {
return metadata;
}
void setMetadata(Map<String, String> metadata) {
this.metadata = metadata;
}
public Context getContext() {
return context;
}
void setContext(Context context) {
this.context = context;
}
}

54
sdk/src/main/java/io/dapr/client/domain/GetSecretRequestBuilder.java Normal file
View File

@ -0,0 +1,54 @@
/*
* Copyright (c) Microsoft Corporation.
* Licensed under the MIT License.
*/
package io.dapr.client.domain;
import io.grpc.Context;
import java.util.Collections;
import java.util.Map;
/**
* Builds a request to publish an event.
*/
public class GetSecretRequestBuilder {
private final String secretStoreName;
private final String key;
private Map<String, String> metadata;
private Context context;
public GetSecretRequestBuilder(String secretStoreName, String key) {
this.secretStoreName = secretStoreName;
this.key = key;
}
public GetSecretRequestBuilder withMetadata(Map<String, String> metadata) {
this.metadata = metadata == null ? null : Collections.unmodifiableMap(metadata);
return this;
}
public GetSecretRequestBuilder withContext(Context context) {
this.context = context;
return this;
}
/**
* Builds a request object.
* @return Request object.
*/
public GetSecretRequest build() {
GetSecretRequest request = new GetSecretRequest();
request.setSecretStoreName(this.secretStoreName);
request.setKey(this.key);
request.setMetadata(this.metadata);
request.setContext(this.context);
return request;
}
}

64
sdk/src/main/java/io/dapr/client/domain/GetStateRequest.java Normal file
View File

@ -0,0 +1,64 @@
/*
* Copyright (c) Microsoft Corporation.
* Licensed under the MIT License.
*/
package io.dapr.client.domain;
import io.grpc.Context;
/**
* A request to get a state by key.
*/
public class GetStateRequest {
private String stateStoreName;
private String key;
private String etag;
private StateOptions stateOptions;
private Context context;
public String getStateStoreName() {
return stateStoreName;
}
void setStateStoreName(String stateStoreName) {
this.stateStoreName = stateStoreName;
}
public String getKey() {
return key;
}
void setKey(String key) {
this.key = key;
}
public String getEtag() {
return etag;
}
void setEtag(String etag) {
this.etag = etag;
}
public StateOptions getStateOptions() {
return stateOptions;
}
void setStateOptions(StateOptions stateOptions) {
this.stateOptions = stateOptions;
}
public Context getContext() {
return context;
}
void setContext(Context context) {
this.context = context;
}
}

59
sdk/src/main/java/io/dapr/client/domain/GetStateRequestBuilder.java Normal file
View File

@ -0,0 +1,59 @@
/*
* Copyright (c) Microsoft Corporation.
* Licensed under the MIT License.
*/
package io.dapr.client.domain;
import io.grpc.Context;
/**
* Builds a request to publish an event.
*/
public class GetStateRequestBuilder {
private final String stateStoreName;
private final String key;
private String etag;
private StateOptions stateOptions;
private Context context;
public GetStateRequestBuilder(String stateStoreName, String key) {
this.stateStoreName = stateStoreName;
this.key = key;
}
public GetStateRequestBuilder withEtag(String etag) {
this.etag = etag;
return this;
}
public GetStateRequestBuilder withStateOptions(StateOptions stateOptions) {
this.stateOptions = stateOptions;
return this;
}
public GetStateRequestBuilder withContext(Context context) {
this.context = context;
return this;
}
/**
* Builds a request object.
* @return Request object.
*/
public GetStateRequest build() {
GetStateRequest request = new GetStateRequest();
request.setStateStoreName(this.stateStoreName);
request.setKey(this.key);
request.setEtag(this.etag);
request.setStateOptions(this.stateOptions);
request.setContext(this.context);
return request;
}
}

6
sdk/src/main/java/io/dapr/client/HttpExtension.java → sdk/src/main/java/io/dapr/client/domain/HttpExtension.java
View File

@ -3,7 +3,9 @@
* Licensed under the MIT License.
*/
package io.dapr.client;
package io.dapr.client.domain;
import io.dapr.client.DaprHttp;
import java.util.Collections;
import java.util.HashMap;
@ -17,7 +19,7 @@ import java.util.Map;
public final class HttpExtension {
/**
* Convenience HttpExtension object for {@link DaprHttp.HttpMethods#NONE} with empty queryString.
* Convenience HttpExtension object for {@link io.dapr.client.DaprHttp.HttpMethods#NONE} with empty queryString.
*/
public static final HttpExtension NONE = new HttpExtension(DaprHttp.HttpMethods.NONE, new HashMap<>());
/**

66
sdk/src/main/java/io/dapr/client/domain/InvokeBindingRequest.java Normal file
View File

@ -0,0 +1,66 @@
/*
* Copyright (c) Microsoft Corporation.
* Licensed under the MIT License.
*/
package io.dapr.client.domain;
import io.grpc.Context;
import java.util.Map;
/**
* A request to invoke binding.
*/
public class InvokeBindingRequest {
private String name;
private String operation;
private Object data;
private Map<String, String> metadata;
private Context context;
public String getName() {
return name;
}
void setName(String name) {
this.name = name;
}
public String getOperation() {
return operation;
}
void setOperation(String operation) {
this.operation = operation;
}
public Object getData() {
return data;
}
void setData(Object data) {
this.data = data;
}
public Map<String, String> getMetadata() {
return metadata;
}
void setMetadata(Map<String, String> metadata) {
this.metadata = metadata;
}
public Context getContext() {
return context;
}
void setContext(Context context) {
this.context = context;
}
}

63
sdk/src/main/java/io/dapr/client/domain/InvokeBindingRequestBuilder.java Normal file
View File

@ -0,0 +1,63 @@
/*
* Copyright (c) Microsoft Corporation.
* Licensed under the MIT License.
*/
package io.dapr.client.domain;
import io.grpc.Context;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
/**
* Builds a request to publish an event.
*/
public class InvokeBindingRequestBuilder {
private final String name;
private final String operation;
private Object data;
private Map<String, String> metadata = new HashMap<>();
private Context context;
public InvokeBindingRequestBuilder(String name, String operation) {
this.name = name;
this.operation = operation;
}
public InvokeBindingRequestBuilder withData(Object data) {
this.data = data;
return this;
}
public InvokeBindingRequestBuilder withMetadata(Map<String, String> metadata) {
this.metadata = metadata == null ? null : Collections.unmodifiableMap(metadata);
return this;
}
public InvokeBindingRequestBuilder withContext(Context context) {
this.context = context;
return this;
}
/**
* Builds a request object.
* @return Request object.
*/
public InvokeBindingRequest build() {
InvokeBindingRequest request = new InvokeBindingRequest();
request.setName(this.name);
request.setOperation(this.operation);
request.setData(this.data);
request.setMetadata(this.metadata);
request.setContext(this.context);
return request;
}
}

76
sdk/src/main/java/io/dapr/client/domain/InvokeServiceRequest.java Normal file
View File

@ -0,0 +1,76 @@
/*
* Copyright (c) Microsoft Corporation.
* Licensed under the MIT License.
*/
package io.dapr.client.domain;
import io.grpc.Context;
import java.util.Map;
/**
* A request to invoke a service.
*/
public class InvokeServiceRequest {
private String appId;
private String method;
private Object body;
private Map<String, String> metadata;
private HttpExtension httpExtension;
private Context context;
public String getAppId() {
return appId;
}
void setAppId(String appId) {
this.appId = appId;
}
public String getMethod() {
return method;
}
void setMethod(String method) {
this.method = method;
}
public Object getBody() {
return body;
}
void setBody(Object body) {
this.body = body;
}
public Map<String, String> getMetadata() {
return metadata;
}
void setMetadata(Map<String, String> metadata) {
this.metadata = metadata;
}
public HttpExtension getHttpExtension() {
return httpExtension;
}
void setHttpExtension(HttpExtension httpExtension) {
this.httpExtension = httpExtension;
}
public Context getContext() {
return context;
}
void setContext(Context context) {
this.context = context;
}
}

71
sdk/src/main/java/io/dapr/client/domain/InvokeServiceRequestBuilder.java Normal file
View File

@ -0,0 +1,71 @@
/*
* Copyright (c) Microsoft Corporation.
* Licensed under the MIT License.
*/
package io.dapr.client.domain;
import io.grpc.Context;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
/**
* Builds a request to invoke a service.
*/
public class InvokeServiceRequestBuilder {
private final String appId;
private final String method;
private Object body;
private Map<String, String> metadata = new HashMap<>();
private HttpExtension httpExtension = HttpExtension.NONE;
private Context context;
public InvokeServiceRequestBuilder(String appId, String method) {
this.appId = appId;
this.method = method;
}
public InvokeServiceRequestBuilder withBody(Object body) {
this.body = body;
return this;
}
public InvokeServiceRequestBuilder withMetadata(Map<String, String> metadata) {
this.metadata = metadata == null ? null : Collections.unmodifiableMap(metadata);
return this;
}
public InvokeServiceRequestBuilder withHttpExtension(HttpExtension httpExtension) {
this.httpExtension = httpExtension;
return this;
}
public InvokeServiceRequestBuilder withContext(Context context) {
this.context = context;
return this;
}
/**
* Builds a request object.
* @return Request object.
*/
public InvokeServiceRequest build() {
InvokeServiceRequest request = new InvokeServiceRequest();
request.setAppId(this.appId);
request.setMethod(this.method);
request.setBody(this.body);
request.setHttpExtension(this.httpExtension);
request.setMetadata(this.metadata);
request.setContext(this.context);
return request;
}
}

66
sdk/src/main/java/io/dapr/client/domain/PublishEventRequest.java Normal file
View File

@ -0,0 +1,66 @@
/*
* Copyright (c) Microsoft Corporation.
* Licensed under the MIT License.
*/
package io.dapr.client.domain;
import io.grpc.Context;
import java.util.Map;
/**
* A request to publish an event.
*/
public class PublishEventRequest {
private String pubsubName;
private String topic;
private Object data;
private Map<String, String> metadata;
private Context context;
public String getPubsubName() {
return pubsubName;
}
void setPubsubName(String pubsubName) {
this.pubsubName = pubsubName;
}
public String getTopic() {
return topic;
}
void setTopic(String topic) {
this.topic = topic;
}
public Object getData() {
return data;
}
void setData(Object data) {
this.data = data;
}
public Map<String, String> getMetadata() {
return metadata;
}
void setMetadata(Map<String, String> metadata) {
this.metadata = metadata;
}
public Context getContext() {
return context;
}
void setContext(Context context) {
this.context = context;
}
}

65
sdk/src/main/java/io/dapr/client/domain/PublishEventRequestBuilder.java Normal file
View File

@ -0,0 +1,65 @@
/*
* Copyright (c) Microsoft Corporation.
* Licensed under the MIT License.
*/
package io.dapr.client.domain;
import io.grpc.Context;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
/**
* Builds a request to publish an event.
*/
public class PublishEventRequestBuilder {
private final String pubsubName;
private final String topic;
private final Object data;
private Map<String, String> metadata = new HashMap<>();
private Context context;
/**
* Instantiates a builder for a publish request.
* @param pubsubName Name of the Dapr PubSub.
* @param topic Topic name.
* @param data Data to be published.
*/
public PublishEventRequestBuilder(String pubsubName, String topic, Object data) {
this.pubsubName = pubsubName;
this.topic = topic;
this.data = data;
}
public PublishEventRequestBuilder withMetadata(Map<String, String> metadata) {
this.metadata = metadata == null ? null : Collections.unmodifiableMap(metadata);
return this;
}
public PublishEventRequestBuilder withContext(Context context) {
this.context = context;
return this;
}
/**
* Builds a request object.
* @return Request object.
*/
public PublishEventRequest build() {
PublishEventRequest request = new PublishEventRequest();
request.setPubsubName(this.pubsubName);
request.setTopic(this.topic);
request.setData(this.data);
request.setMetadata(this.metadata);
request.setContext(this.context);
return request;
}
}

32
sdk/src/main/java/io/dapr/client/domain/Response.java Normal file
View File

@ -0,0 +1,32 @@
/*
* Copyright (c) Microsoft Corporation.
* Licensed under the MIT License.
*/
package io.dapr.client.domain;
import io.grpc.Context;
/**
* A Dapr Response.
*/
public class Response<T> {
private final Context context;
private final T object;
public Response(Context context, T object) {
this.context = context;
this.object = object;
}
public Context getContext() {
return this.context;
}
public T getObject() {
return this.object;
}
}

46
sdk/src/main/java/io/dapr/client/domain/SaveStateRequest.java Normal file
View File

@ -0,0 +1,46 @@
/*
* Copyright (c) Microsoft Corporation.
* Licensed under the MIT License.
*/
package io.dapr.client.domain;
import io.grpc.Context;
import java.util.List;
/**
* A request to save states to state store.
*/
public class SaveStateRequest {
private String stateStoreName;
private List<State<?>> states;
private Context context;
public String getStateStoreName() {
return stateStoreName;
}
void setStateStoreName(String stateStoreName) {
this.stateStoreName = stateStoreName;
}
public List<State<?>> getStates() {
return states;
}
void setStates(List<State<?>> states) {
this.states = states;
}
public Context getContext() {
return context;
}
void setContext(Context context) {
this.context = context;
}
}

56
sdk/src/main/java/io/dapr/client/domain/SaveStateRequestBuilder.java Normal file
View File

@ -0,0 +1,56 @@
/*
* Copyright (c) Microsoft Corporation.
* Licensed under the MIT License.
*/
package io.dapr.client.domain;
import io.grpc.Context;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
/**
* A request to save states to state store.
*/
public class SaveStateRequestBuilder {
private final String stateStoreName;
private List<State<?>> states = new ArrayList<>();
private Context context;
public SaveStateRequestBuilder(String stateStoreName) {
this.stateStoreName = stateStoreName;
}
public SaveStateRequestBuilder withStates(State<?>... states) {
this.states = Collections.unmodifiableList(Arrays.asList(states));
return this;
}
public SaveStateRequestBuilder withStates(List<State<?>> states) {
this.states = states == null ? null : Collections.unmodifiableList(states);
return this;
}
public SaveStateRequestBuilder withContext(Context context) {
this.context = context;
return this;
}
/**
* Builds a request object.
* @return Request object.
*/
public SaveStateRequest build() {
SaveStateRequest request = new SaveStateRequest();
request.setStateStoreName(this.stateStoreName);
request.setStates(this.states);
request.setContext(this.context);
return request;
}
}

2
sdk/src/main/java/io/dapr/client/domain/StateOptions.java
View File

@ -12,8 +12,6 @@ import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.databind.DeserializationContext;
import com.fasterxml.jackson.databind.SerializerProvider;
import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
import com.fasterxml.jackson.databind.annotation.JsonSerialize;
import com.fasterxml.jackson.databind.deser.std.StdDeserializer;
import com.fasterxml.jackson.databind.exc.InvalidFormatException;
import com.fasterxml.jackson.databind.ser.std.StdSerializer;

1
sdk/src/main/java/io/dapr/serializer/DefaultObjectSerializer.java
View File

@ -9,7 +9,6 @@ import io.dapr.client.ObjectSerializer;
import io.dapr.utils.TypeRef;
import java.io.IOException;
import java.lang.reflect.Type;
/**
* Default serializer/deserializer for request/response objects and for state objects too.

3
sdk/src/test/java/io/dapr/client/DaprClientBuilderTest.java
View File

@ -7,7 +7,8 @@ package io.dapr.client;
import io.dapr.serializer.DaprObjectSerializer;
import org.junit.Test;
import static org.junit.jupiter.api.Assertions.*;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.mockito.Mockito.mock;
public class DaprClientBuilderTest {

68
sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java
View File

@ -10,6 +10,7 @@ import com.google.common.util.concurrent.SettableFuture;
import com.google.protobuf.Any;
import com.google.protobuf.ByteString;
import com.google.protobuf.Empty;
import io.dapr.client.domain.HttpExtension;
import io.dapr.client.domain.State;
import io.dapr.client.domain.StateOptions;
import io.dapr.serializer.DefaultObjectSerializer;
@ -26,17 +27,26 @@ import reactor.core.publisher.Mono;
import java.io.Closeable;
import java.io.IOException;
import java.time.Duration;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
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;
import static org.junit.Assert.*;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.*;
import static org.mockito.Mockito.argThat;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.when;
public class DaprClientGrpcTest {
@ -53,6 +63,7 @@ public class DaprClientGrpcTest {
public void setup() throws IOException {
closeable = mock(Closeable.class);
client = mock(DaprGrpc.DaprFutureStub.class);
when(client.withInterceptors(any())).thenReturn(client);
adapter = new DaprClientGrpc(closeable, client, new DefaultObjectSerializer(), new DefaultObjectSerializer());
serializer = new ObjectSerializer();
doNothing().when(closeable).close();
@ -154,7 +165,7 @@ public class DaprClientGrpcTest {
public void invokeBindingTest() throws IOException {
SettableFuture<DaprProtos.InvokeBindingResponse> settableFuture = SettableFuture.create();
DaprProtos.InvokeBindingResponse.Builder responseBuilder =
DaprProtos.InvokeBindingResponse.newBuilder().setData(getBytes("OK"));
DaprProtos.InvokeBindingResponse.newBuilder().setData(serialize("OK"));
MockCallback<DaprProtos.InvokeBindingResponse> callback = new MockCallback<>(responseBuilder.build());
addCallback(settableFuture, callback, directExecutor());
when(client.invokeBinding(any(DaprProtos.InvokeBindingRequest.class)))
@ -165,11 +176,26 @@ public class DaprClientGrpcTest {
assertTrue(callback.wasCalled);
}
@Test
public void invokeBindingByteArrayTest() throws IOException {
SettableFuture<DaprProtos.InvokeBindingResponse> settableFuture = SettableFuture.create();
DaprProtos.InvokeBindingResponse.Builder responseBuilder =
DaprProtos.InvokeBindingResponse.newBuilder().setData(ByteString.copyFrom("OK".getBytes()));
MockCallback<DaprProtos.InvokeBindingResponse> callback = new MockCallback<>(responseBuilder.build());
addCallback(settableFuture, callback, directExecutor());
when(client.invokeBinding(any(DaprProtos.InvokeBindingRequest.class)))
.thenReturn(settableFuture);
Mono<byte[]> result = adapter.invokeBinding("BindingName", "MyOperation", "request".getBytes(), Collections.EMPTY_MAP);
settableFuture.set(responseBuilder.build());
assertEquals("OK", new String(result.block(), StandardCharsets.UTF_8));
assertTrue(callback.wasCalled);
}
@Test
public void invokeBindingObjectTest() throws IOException {
SettableFuture<DaprProtos.InvokeBindingResponse> settableFuture = SettableFuture.create();
DaprProtos.InvokeBindingResponse.Builder responseBuilder =
DaprProtos.InvokeBindingResponse.newBuilder().setData(getBytes("OK"));
DaprProtos.InvokeBindingResponse.newBuilder().setData(serialize("OK"));
MockCallback<DaprProtos.InvokeBindingResponse> callback = new MockCallback<>(responseBuilder.build());
addCallback(settableFuture, callback, directExecutor());
when(client.invokeBinding(any(DaprProtos.InvokeBindingRequest.class)))
@ -185,13 +211,29 @@ public class DaprClientGrpcTest {
public void invokeBindingResponseObjectTest() throws IOException {
SettableFuture<DaprProtos.InvokeBindingResponse> settableFuture = SettableFuture.create();
DaprProtos.InvokeBindingResponse.Builder responseBuilder =
DaprProtos.InvokeBindingResponse.newBuilder().setData(getBytes("OK"));
DaprProtos.InvokeBindingResponse.newBuilder().setData(serialize("OK"));
MockCallback<DaprProtos.InvokeBindingResponse> callback = new MockCallback<>(responseBuilder.build());
addCallback(settableFuture, callback, directExecutor());
when(client.invokeBinding(any(DaprProtos.InvokeBindingRequest.class)))
.thenReturn(settableFuture);
.thenReturn(settableFuture);
MyObject event = new MyObject(1, "Event");
Mono<String> result = adapter.invokeBinding("BindingName", "MyOperation", event, null, String.class);
Mono<String> result = adapter.invokeBinding("BindingName", "MyOperation", event, String.class);
settableFuture.set(responseBuilder.build());
assertEquals("OK", result.block());
assertTrue(callback.wasCalled);
}
@Test
public void invokeBindingResponseObjectTypeRefTest() throws IOException {
SettableFuture<DaprProtos.InvokeBindingResponse> settableFuture = SettableFuture.create();
DaprProtos.InvokeBindingResponse.Builder responseBuilder =
DaprProtos.InvokeBindingResponse.newBuilder().setData(serialize("OK"));
MockCallback<DaprProtos.InvokeBindingResponse> callback = new MockCallback<>(responseBuilder.build());
addCallback(settableFuture, callback, directExecutor());
when(client.invokeBinding(any(DaprProtos.InvokeBindingRequest.class)))
.thenReturn(settableFuture);
MyObject event = new MyObject(1, "Event");
Mono<String> result = adapter.invokeBinding("BindingName", "MyOperation", event, TypeRef.get(String.class));
settableFuture.set(responseBuilder.build());
assertEquals("OK", result.block());
assertTrue(callback.wasCalled);
@ -642,7 +684,7 @@ public class DaprClientGrpcTest {
StateOptions options = buildStateOptions(StateOptions.Consistency.STRONG, StateOptions.Concurrency.FIRST_WRITE);
State<MyObject> expectedState = buildStateKey(expectedValue, key, etag, options);
DaprProtos.GetStateResponse responseEnvelope = DaprProtos.GetStateResponse.newBuilder()
.setData(getBytes(expectedValue))
.setData(serialize(expectedValue))
.setEtag(etag)
.build();
State<MyObject> keyRequest = buildStateKey(null, key, etag, options);
@ -664,7 +706,7 @@ public class DaprClientGrpcTest {
StateOptions options = new StateOptions(null, StateOptions.Concurrency.FIRST_WRITE);
State<MyObject> expectedState = buildStateKey(expectedValue, key, etag, options);
DaprProtos.GetStateResponse responseEnvelope = DaprProtos.GetStateResponse.newBuilder()
.setData(getBytes(expectedValue))
.setData(serialize(expectedValue))
.setEtag(etag)
.build();
State<MyObject> keyRequest = buildStateKey(null, key, etag, options);
@ -1125,7 +1167,7 @@ public class DaprClientGrpcTest {
private <T> DaprProtos.GetStateResponse buildGetStateResponse(T value, String etag) throws IOException {
return DaprProtos.GetStateResponse.newBuilder()
.setData(getBytes(value))
.setData(serialize(value))
.setEtag(etag)
.build();
}
@ -1149,10 +1191,10 @@ public class DaprClientGrpcTest {
}
private Any getAny(Object value) throws IOException {
return Any.newBuilder().setValue(getBytes(value)).build();
return Any.newBuilder().setValue(serialize(value)).build();
}
private ByteString getBytes(Object value) throws IOException {
private ByteString serialize(Object value) throws IOException {
byte[] byteValue = serializer.serialize(value);
return ByteString.copyFrom(byteValue);
}

12
sdk/src/test/java/io/dapr/client/DaprClientHttpTest.java
View File

@ -4,6 +4,7 @@
*/
package io.dapr.client;
import io.dapr.client.domain.HttpExtension;
import io.dapr.client.domain.State;
import io.dapr.client.domain.StateOptions;
import okhttp3.OkHttpClient;
@ -13,9 +14,16 @@ import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Mono;
import java.util.*;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import static org.junit.Assert.*;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.Mockito.mock;

1
sdk/src/test/java/io/dapr/client/DaprHttpBuilderTest.java
View File

@ -13,7 +13,6 @@ import java.time.Duration;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.mockito.Mockito.mock;
public class DaprHttpBuilderTest {

16
sdk/src/test/java/io/dapr/client/DaprHttpStub.java
View File

@ -5,8 +5,10 @@
package io.dapr.client;
import io.grpc.Context;
import reactor.core.publisher.Mono;
import java.io.IOException;
import java.util.Map;
/**
@ -31,7 +33,7 @@ public class DaprHttpStub extends DaprHttp {
* {@inheritDoc}
*/
@Override
public Mono<DaprHttp.Response> invokeApi(String method, String urlString, Map<String, String> urlParameters, Map<String, String> headers) {
public Mono<Response> invokeApi(String method, String urlString, Map<String, String> urlParameters, Map<String, String> headers, Context context) {
return Mono.empty();
}
@ -39,7 +41,7 @@ public class DaprHttpStub extends DaprHttp {
* {@inheritDoc}
*/
@Override
public Mono<DaprHttp.Response> invokeApi(String method, String urlString, Map<String, String> urlParameters, String content, Map<String, String> headers) {
public Mono<Response> invokeApi(String method, String urlString, Map<String, String> urlParameters, String content, Map<String, String> headers, Context context) {
return Mono.empty();
}
@ -47,7 +49,15 @@ public class DaprHttpStub extends DaprHttp {
* {@inheritDoc}
*/
@Override
public Mono<DaprHttp.Response> invokeApi(String method, String urlString, Map<String, String> urlParameters, byte[] content, Map<String, String> headers) {
public Mono<Response> invokeApi(String method, String urlString, Map<String, String> urlParameters, byte[] content, Map<String, String> headers, Context context) {
return Mono.empty();
}
/**
* {@inheritDoc}
*/
@Override
public void close() throws IOException {
return;
}
}

46
sdk/src/test/java/io/dapr/client/DaprHttpTest.java
View File

@ -6,7 +6,10 @@ package io.dapr.client;
import io.dapr.exceptions.DaprException;
import io.dapr.utils.Constants;
import okhttp3.*;
import io.grpc.Context;
import okhttp3.MediaType;
import okhttp3.OkHttpClient;
import okhttp3.ResponseBody;
import okhttp3.mock.Behavior;
import okhttp3.mock.MockInterceptor;
import org.junit.Before;
@ -17,19 +20,22 @@ import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import static org.junit.Assert.*;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.fail;
public class DaprHttpTest {
private static final String EXPECTED_RESULT =
"{\"data\":\"ewoJCSJwcm9wZXJ0eUEiOiAidmFsdWVBIiwKCQkicHJvcGVydHlCIjogInZhbHVlQiIKCX0=\"}";
private OkHttpClient okHttpClient;
private MockInterceptor mockInterceptor;
private ObjectSerializer serializer = new ObjectSerializer();
private final String EXPECTED_RESULT = "{\"data\":\"ewoJCSJwcm9wZXJ0eUEiOiAidmFsdWVBIiwKCQkicHJvcGVydHlCIjogInZhbHVlQiIKCX0=\"}";
@Before
public void setUp() throws Exception {
mockInterceptor = new MockInterceptor(Behavior.UNORDERED);
@ -42,10 +48,11 @@ public class DaprHttpTest {
headers.put("content-type", "text/html");
headers.put("header1", "value1");
mockInterceptor.addRule()
.post("http://127.0.0.1:3500/v1.0/state")
.respond(serializer.serialize(EXPECTED_RESULT));
.post("http://127.0.0.1:3500/v1.0/state")
.respond(serializer.serialize(EXPECTED_RESULT));
DaprHttp daprHttp = new DaprHttp(3500, okHttpClient);
Mono<DaprHttp.Response> mono = daprHttp.invokeApi("POST", "v1.0/state", null, (byte[]) null, headers);
Mono<DaprHttp.Response> mono =
daprHttp.invokeApi("POST", "v1.0/state", null, (byte[]) null, headers, Context.current());
DaprHttp.Response response = mono.block();
String body = serializer.deserialize(response.getBody(), String.class);
assertEquals(EXPECTED_RESULT, body);
@ -58,7 +65,8 @@ public class DaprHttpTest {
.respond(serializer.serialize(EXPECTED_RESULT))
.addHeader("Header", "Value");
DaprHttp daprHttp = new DaprHttp(3500, okHttpClient);
Mono<DaprHttp.Response> mono = daprHttp.invokeApi("POST", "v1.0/state", null, "", null);
Mono<DaprHttp.Response> mono =
daprHttp.invokeApi("POST", "v1.0/state", null, "", null, Context.current());
DaprHttp.Response response = mono.block();
String body = serializer.deserialize(response.getBody(), String.class);
assertEquals(EXPECTED_RESULT, body);
@ -70,7 +78,8 @@ public class DaprHttpTest {
.delete("http://127.0.0.1:3500/v1.0/state")
.respond(serializer.serialize(EXPECTED_RESULT));
DaprHttp daprHttp = new DaprHttp(3500, okHttpClient);
Mono<DaprHttp.Response> mono = daprHttp.invokeApi("DELETE", "v1.0/state", null, (String) null, null);
Mono<DaprHttp.Response> mono =
daprHttp.invokeApi("DELETE", "v1.0/state", null, (String) null, null, Context.current());
DaprHttp.Response response = mono.block();
String body = serializer.deserialize(response.getBody(), String.class);
assertEquals(EXPECTED_RESULT, body);
@ -82,7 +91,7 @@ public class DaprHttpTest {
.get("http://127.0.0.1:3500/v1.0/get")
.respond(serializer.serialize(EXPECTED_RESULT));
DaprHttp daprHttp = new DaprHttp(3500, okHttpClient);
Mono<DaprHttp.Response> mono = daprHttp.invokeApi("GET", "v1.0/get", null, null);
Mono<DaprHttp.Response> mono = daprHttp.invokeApi("GET", "v1.0/get", null, null, Context.current());
DaprHttp.Response response = mono.block();
String body = serializer.deserialize(response.getBody(), String.class);
assertEquals(EXPECTED_RESULT, body);
@ -99,7 +108,8 @@ public class DaprHttpTest {
.get("http://127.0.0.1:3500/v1.0/state/order?orderId=41")
.respond(serializer.serialize(EXPECTED_RESULT));
DaprHttp daprHttp = new DaprHttp(3500, okHttpClient);
Mono<DaprHttp.Response> mono = daprHttp.invokeApi("GET", "v1.0/state/order", urlParameters, headers);
Mono<DaprHttp.Response> mono =
daprHttp.invokeApi("GET", "v1.0/state/order", urlParameters, headers, Context.current());
DaprHttp.Response response = mono.block();
String body = serializer.deserialize(response.getBody(), String.class);
assertEquals(EXPECTED_RESULT, body);
@ -111,7 +121,8 @@ public class DaprHttpTest {
.post("http://127.0.0.1:3500/v1.0/state")
.respond(500);
DaprHttp daprHttp = new DaprHttp(3500, okHttpClient);
Mono<DaprHttp.Response> mono = daprHttp.invokeApi("POST", "v1.0/state", null, null);
Mono<DaprHttp.Response> mono =
daprHttp.invokeApi("POST", "v1.0/state", null, null, Context.current());
DaprHttp.Response response = mono.block();
String body = serializer.deserialize(response.getBody(), String.class);
assertEquals(EXPECTED_RESULT, body);
@ -124,7 +135,7 @@ public class DaprHttpTest {
.respond(500, ResponseBody.create(MediaType.parse("text"),
"{\"errorCode\":null,\"message\":null}"));
DaprHttp daprHttp = new DaprHttp(3500, okHttpClient);
Mono<DaprHttp.Response> mono = daprHttp.invokeApi("POST", "v1.0/state", null, null);
Mono<DaprHttp.Response> mono = daprHttp.invokeApi("POST", "v1.0/state", null, null, Context.current());
DaprHttp.Response response = mono.block();
String body = serializer.deserialize(response.getBody(), String.class);
assertEquals(EXPECTED_RESULT, body);
@ -137,7 +148,7 @@ public class DaprHttpTest {
.respond(500, ResponseBody.create(MediaType.parse("application/json"),
"{\"errorCode\":\"null\",\"message\":\"null\"}"));
DaprHttp daprHttp = new DaprHttp(3500, okHttpClient);
Mono<DaprHttp.Response> mono = daprHttp.invokeApi("POST", "v1.0/state", null, null);
Mono<DaprHttp.Response> mono = daprHttp.invokeApi("POST", "v1.0/state", null, null, Context.current());
DaprHttp.Response response = mono.block();
String body = serializer.deserialize(response.getBody(), String.class);
assertEquals(EXPECTED_RESULT, body);
@ -180,10 +191,11 @@ public class DaprHttpTest {
.respond(200, ResponseBody.create(MediaType.parse("application/json"),
serializer.serialize(existingState)));
DaprHttp daprHttp = new DaprHttp(3500, okHttpClient);
Mono<DaprHttp.Response> response = daprHttp.invokeApi("GET", urlExistingState, null, null);
Mono<DaprHttp.Response> response = daprHttp.invokeApi("GET", urlExistingState, null, null, Context.current());
assertEquals(existingState, serializer.deserialize(response.block().getBody(), String.class));
Mono<DaprHttp.Response> responseDeleted = daprHttp.invokeApi("GET", urlDeleteState, null, null);
Mono<DaprHttp.Response> responseDeleteKey = daprHttp.invokeApi("DELETE", urlDeleteState, null, null);
Mono<DaprHttp.Response> responseDeleted = daprHttp.invokeApi("GET", urlDeleteState, null, null, Context.current());
Mono<DaprHttp.Response> responseDeleteKey =
daprHttp.invokeApi("DELETE", urlDeleteState, null, null, Context.current());
assertNull(serializer.deserialize(responseDeleteKey.block().getBody(), String.class));
mockInterceptor.reset();
mockInterceptor.addRule()

6
sdk/src/test/java/io/dapr/runtime/Dapr.java
View File

@ -10,7 +10,11 @@ import io.dapr.client.domain.CloudEvent;
import reactor.core.publisher.Mono;
import java.io.IOException;
import java.util.*;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;

16
sdk/src/test/java/io/dapr/runtime/DaprRuntimeTest.java
View File

@ -7,8 +7,11 @@ package io.dapr.runtime;
import com.fasterxml.jackson.core.JsonFactory;
import com.fasterxml.jackson.core.JsonGenerator;
import io.dapr.client.*;
import io.dapr.client.DaprClient;
import io.dapr.client.DaprClientTestBuilder;
import io.dapr.client.DaprHttpStub;
import io.dapr.client.domain.CloudEvent;
import io.dapr.client.domain.HttpExtension;
import io.dapr.serializer.DaprObjectSerializer;
import io.dapr.serializer.DefaultObjectSerializer;
import io.dapr.utils.Constants;
@ -25,7 +28,12 @@ import java.util.Map;
import java.util.UUID;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.*;
import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
public class DaprRuntimeTest {
@ -116,7 +124,8 @@ public class DaprRuntimeTest {
eq(Constants.PUBLISH_PATH + "/" + PUBSUB_NAME + "/" + TOPIC_NAME),
any(),
eq(serializer.serialize(message.data)),
eq(null)))
eq(null),
any()))
.thenAnswer(invocationOnMock -> this.daprRuntime.handleInvocation(
TOPIC_NAME,
this.serialize(message),
@ -203,6 +212,7 @@ public class DaprRuntimeTest {
eq(Constants.INVOKE_PATH + "/" + APP_ID + "/method/" + METHOD_NAME),
any(),
eq(serializer.serialize(message.data)),
any(),
any()))
.thenAnswer(x ->
this.daprRuntime.handleInvocation(

1
sdk/src/test/java/io/dapr/runtime/MethodListener.java
View File

@ -5,7 +5,6 @@
package io.dapr.runtime;
import com.fasterxml.jackson.core.JsonProcessingException;
import reactor.core.publisher.Mono;
import java.util.Map;

11
sdk/src/test/java/io/dapr/serializer/DefaultObjectSerializerTest.java
View File

@ -5,26 +5,23 @@
package io.dapr.serializer;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.dapr.serializer.DefaultObjectSerializer;
import io.dapr.client.domain.CloudEvent;
import io.dapr.utils.TypeRef;
import org.junit.Assert;
import org.junit.Test;
import java.io.IOException;
import java.io.Serializable;
import java.lang.ref.Reference;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.Base64;
import java.util.List;
import java.util.function.Function;
import static org.junit.Assert.*;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public class DefaultObjectSerializerTest {