Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -17,9 +17,18 @@
import com.microsoft.durabletask.DurableTaskGrpcClientBuilder;
import com.microsoft.durabletask.OrchestrationMetadata;
import com.microsoft.durabletask.PurgeResult;
import io.dapr.client.Headers;
import io.dapr.config.Properties;
import io.dapr.utils.NetworkUtils;
import io.dapr.workflows.Workflow;
import io.grpc.CallOptions;
import io.grpc.Channel;
import io.grpc.ClientCall;
import io.grpc.ClientInterceptor;
import io.grpc.ForwardingClientCall;
import io.grpc.ManagedChannel;
import io.grpc.Metadata;
import io.grpc.MethodDescriptor;

import javax.annotation.Nullable;

Expand All @@ -39,7 +48,7 @@ public class DaprWorkflowClient implements AutoCloseable {
* Public constructor for DaprWorkflowClient. This layer constructs the GRPC Channel.
*/
public DaprWorkflowClient() {
this(NetworkUtils.buildGrpcManagedChannel());
this(NetworkUtils.buildGrpcManagedChannel(WORKFLOW_INTERCEPTOR));
}

/**
Expand Down Expand Up @@ -239,4 +248,26 @@ public void close() throws InterruptedException {
}
}
}

private static ClientInterceptor WORKFLOW_INTERCEPTOR = new ClientInterceptor() {
@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
MethodDescriptor<ReqT, RespT> methodDescriptor,
CallOptions options,
Channel channel) {
// TBD: do we need timeout in workflow client?
ClientCall<ReqT, RespT> clientCall = channel.newCall(methodDescriptor, options);
return new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(clientCall) {
@Override
public void start(final Listener<RespT> responseListener, final Metadata metadata) {
String daprApiToken = Properties.API_TOKEN.get();
if (daprApiToken != null) {
metadata.put(Metadata.Key.of(Headers.DAPR_API_TOKEN, Metadata.ASCII_STRING_MARSHALLER), daprApiToken);
}
super.start(responseListener, metadata);
}
};
}
};
}

8 changes: 4 additions & 4 deletions sdk/src/main/java/io/dapr/client/Headers.java
Original file line number Diff line number Diff line change
Expand Up @@ -16,20 +16,20 @@
/**
* Common headers for GRPC and HTTP communication.
*/
class Headers {
public final class Headers {

/**
* OpenCensus's metadata for GRPC.
*/
static final String GRPC_TRACE_BIN = "grpc-trace-bin";
public static final String GRPC_TRACE_BIN = "grpc-trace-bin";

/**
* Token for authentication from Application to Dapr runtime.
*/
static final String DAPR_API_TOKEN = "dapr-api-token";
public static final String DAPR_API_TOKEN = "dapr-api-token";

/**
* Header for Api Logging User-Agent.
*/
static final String DAPR_USER_AGENT = "User-Agent";
public static final String DAPR_USER_AGENT = "User-Agent";
}
7 changes: 6 additions & 1 deletion sdk/src/main/java/io/dapr/utils/NetworkUtils.java
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
package io.dapr.utils;

import io.dapr.config.Properties;
import io.grpc.ClientInterceptor;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;

Expand Down Expand Up @@ -57,9 +58,10 @@ public static void waitForSocket(String host, int port, int timeoutInMillisecond

/**
* Creates a GRPC managed channel.
* @param interceptors Optional interceptors to add to the channel.
* @return GRPC managed channel to communicate with the sidecar.
*/
public static ManagedChannel buildGrpcManagedChannel() {
public static ManagedChannel buildGrpcManagedChannel(ClientInterceptor... interceptors) {
String address = Properties.SIDECAR_IP.get();
int port = Properties.GRPC_PORT.get();
boolean insecure = true;
Expand All @@ -78,6 +80,9 @@ public static ManagedChannel buildGrpcManagedChannel() {
if (insecure) {
builder = builder.usePlaintext();
}
if (interceptors != null && interceptors.length > 0) {
builder = builder.intercept(interceptors);
}
return builder.build();
}

Expand Down