Skip to content
New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Support remote endpoint. #877

Merged
merged 5 commits into from
Jun 19, 2023
Merged

Support remote endpoint. #877

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
9f190d2
Support remote endpoint.
artursouza Jun 10, 2023
3ea95a4
Merge branch 'master' into remote_endpoint
artursouza Jun 13, 2023
591137a
Merge branch 'master' into remote_endpoint
mukundansundar Jun 13, 2023
1f4bcf5
Use GRPC_ENDPOINT and HTTP_ENDPOINT in integration tests.
artursouza Jun 13, 2023
f5c1364
Fix happy path for waiting for sidecar test.
artursouza Jun 15, 2023
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
4 changes: 4 additions & 0 deletions sdk-tests/src/test/java/io/dapr/it/DaprRun.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -140,6 +140,10 @@ public void use() {
System.getProperties().setProperty(
Properties.API_METHOD_INVOCATION_PROTOCOL.getName(),
DaprApiProtocol.GRPC.name());
System.getProperties().setProperty(
Properties.GRPC_ENDPOINT.getName(), "http://127.0.0.1:" + this.ports.getGrpcPort());
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

shouldn't this just be 127.0.0.1? Why is there an http prefix?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is an endpoint, so the schema is part of it. Users will also use http vs https and the port will be parsed automatically.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why are we using a http scheme for a gRPC endpoint?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

gRPC can operate over HTTP. In this case, we use to parse the port. The .NET SDK for gRPC communication takes in the entire URL as an endpoint. See https://learn.microsoft.com/en-us/aspnet/core/tutorials/grpc/grpc-start?view=aspnetcore-7.0&tabs=visual-studio#create-the-greeter-client

System.getProperties().setProperty(
Properties.HTTP_ENDPOINT.getName(), "http://127.0.0.1:" + this.ports.getHttpPort());
}

public void switchToGRPC() {
Expand Down
32 changes: 21 additions & 11 deletions sdk/src/main/java/io/dapr/client/DaprClientBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.net.URI;

/**
* A builder for the DaprClient,
Expand Down Expand Up @@ -162,19 +163,28 @@ private DaprClient buildDaprClient(DaprApiProtocol protocol) {
* @throws java.lang.IllegalStateException if either host is missing or if port is missing or a negative number.
*/
private DaprClient buildDaprClientGrpc() {
final ManagedChannel channel = buildGrpcManagedChanel();
final GrpcChannelFacade channelFacade = new GrpcChannelFacade(channel);
DaprGrpc.DaprStub asyncStub = DaprGrpc.newStub(channel);
return new DaprClientGrpc(channelFacade, asyncStub, this.objectSerializer, this.stateSerializer);
}

private ManagedChannel buildGrpcManagedChanel() {
String host = Properties.SIDECAR_IP.get();
int port = Properties.GRPC_PORT.get();
if (port <= 0) {
throw new IllegalArgumentException("Invalid port.");
boolean insecure = true;
String grpcEndpoint = Properties.GRPC_ENDPOINT.get();
if ((grpcEndpoint != null) && !grpcEndpoint.isEmpty()) {
URI uri = URI.create(grpcEndpoint);
insecure = uri.getScheme().equalsIgnoreCase("http");
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

See, we use the schema here.

port = uri.getPort() > 0 ? uri.getPort() : (insecure ? 80 : 443);
}
ManagedChannel channel = ManagedChannelBuilder.forAddress(
Properties.SIDECAR_IP.get(), port).usePlaintext().userAgent(Version.getSdkVersion()).build();
Closeable closeableChannel = () -> {
if (channel != null && !channel.isShutdown()) {
channel.shutdown();
}
};
DaprGrpc.DaprStub asyncStub = DaprGrpc.newStub(channel);
return new DaprClientGrpc(closeableChannel, asyncStub, this.objectSerializer, this.stateSerializer);
ManagedChannelBuilder builder = ManagedChannelBuilder.forAddress(host, port)
.userAgent(Version.getSdkVersion());
if (insecure) {
builder = builder.usePlaintext();
}
return builder.build();
}

/**
Expand Down
119 changes: 55 additions & 64 deletions sdk/src/main/java/io/dapr/client/DaprClientGrpc.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,6 @@
import io.dapr.serializer.DaprObjectSerializer;
import io.dapr.serializer.DefaultObjectSerializer;
import io.dapr.utils.DefaultContentTypeConverter;
import io.dapr.utils.NetworkUtils;
import io.dapr.utils.TypeRef;
import io.dapr.v1.CommonProtos;
import io.dapr.v1.DaprGrpc;
Expand All @@ -69,7 +68,6 @@
import reactor.core.publisher.MonoSink;
import reactor.util.context.ContextView;

import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
Expand All @@ -92,7 +90,7 @@ public class DaprClientGrpc extends AbstractDaprClient {
/**
* The GRPC managed channel to be used.
*/
private Closeable channel;
private final GrpcChannelFacade channel;

/**
* The async gRPC stub.
Expand All @@ -102,19 +100,19 @@ public class DaprClientGrpc extends AbstractDaprClient {
/**
* Default access level constructor, in order to create an instance of this class use io.dapr.client.DaprClientBuilder
*
* @param closeableChannel A closeable for a Managed GRPC channel
* @param channel Facade for the managed GRPC channel
* @param asyncStub async gRPC stub
* @param objectSerializer Serializer for transient request/response objects.
* @param stateSerializer Serializer for state objects.
* @see DaprClientBuilder
*/
DaprClientGrpc(
Closeable closeableChannel,
GrpcChannelFacade channel,
DaprGrpc.DaprStub asyncStub,
DaprObjectSerializer objectSerializer,
DaprObjectSerializer stateSerializer) {
super(objectSerializer, stateSerializer);
this.channel = closeableChannel;
this.channel = channel;
this.asyncStub = intercept(asyncStub);
}

Expand Down Expand Up @@ -145,13 +143,7 @@ private CommonProtos.StateOptions.StateConcurrency getGrpcStateConcurrency(State
*/
@Override
public Mono<Void> waitForSidecar(int timeoutInMilliseconds) {
return Mono.fromRunnable(() -> {
try {
NetworkUtils.waitForSocket(Properties.SIDECAR_IP.get(), Properties.GRPC_PORT.get(), timeoutInMilliseconds);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
return this.channel.waitForChannelReady(timeoutInMilliseconds);
}

/**
Expand Down Expand Up @@ -193,7 +185,6 @@ public Mono<Void> publishEvent(PublishEventRequest request) {
}

/**
*
* {@inheritDoc}
*/
@Override
Expand All @@ -209,7 +200,7 @@ public <T> Mono<BulkPublishResponse<T>> publishEvents(BulkPublishRequest<T> requ
throw new IllegalArgumentException("pubsubName and topic name cannot be null or empty");
}

for (BulkPublishEntry<?> entry: request.getEntries()) {
for (BulkPublishEntry<?> entry : request.getEntries()) {
Object event = entry.getEvent();
byte[] data;
String contentType = entry.getContentType();
Expand Down Expand Up @@ -251,7 +242,7 @@ public <T> Mono<BulkPublishResponse<T>> publishEvents(BulkPublishRequest<T> requ
}

Map<String, BulkPublishEntry<T>> entryMap = new HashMap<>();
for (BulkPublishEntry<T> entry: request.getEntries()) {
for (BulkPublishEntry<T> entry : request.getEntries()) {
entryMap.put(entry.getEntryId(), entry);
}
return Mono.deferContextual(
Expand Down Expand Up @@ -299,17 +290,17 @@ public <T> Mono<T> invokeMethod(InvokeMethodRequest invokeMethodRequest, TypeRef
// gRPC to HTTP does not map correctly in Dapr runtime as per https://github.com/dapr/dapr/issues/2342

return Mono.deferContextual(
context -> this.<CommonProtos.InvokeResponse>createMono(
it -> intercept(context, asyncStub).invokeService(envelope, it)
)
).flatMap(
it -> {
try {
return Mono.justOrEmpty(objectSerializer.deserialize(it.getData().getValue().toByteArray(), type));
} catch (IOException e) {
throw DaprException.propagate(e);
}
}
context -> this.<CommonProtos.InvokeResponse>createMono(
it -> intercept(context, asyncStub).invokeService(envelope, it)
)
).flatMap(
it -> {
try {
return Mono.justOrEmpty(objectSerializer.deserialize(it.getData().getValue().toByteArray(), type));
} catch (IOException e) {
throw DaprException.propagate(e);
}
}
);
} catch (Exception ex) {
return DaprException.wrapMono(ex);
Expand Down Expand Up @@ -346,17 +337,17 @@ public <T> Mono<T> invokeBinding(InvokeBindingRequest request, TypeRef<T> type)
DaprProtos.InvokeBindingRequest envelope = builder.build();

return Mono.deferContextual(
context -> this.<DaprProtos.InvokeBindingResponse>createMono(
it -> intercept(context, asyncStub).invokeBinding(envelope, it)
)
).flatMap(
it -> {
try {
return Mono.justOrEmpty(objectSerializer.deserialize(it.getData().toByteArray(), type));
} catch (IOException e) {
throw DaprException.propagate(e);
}
}
context -> this.<DaprProtos.InvokeBindingResponse>createMono(
it -> intercept(context, asyncStub).invokeBinding(envelope, it)
)
).flatMap(
it -> {
try {
return Mono.justOrEmpty(objectSerializer.deserialize(it.getData().toByteArray(), type));
} catch (IOException e) {
throw DaprException.propagate(e);
}
}
);
} catch (Exception ex) {
return DaprException.wrapMono(ex);
Expand Down Expand Up @@ -442,12 +433,12 @@ public <T> Mono<List<State<T>>> getBulkState(GetBulkStateRequest request, TypeRe
DaprProtos.GetBulkStateRequest envelope = builder.build();

return Mono.deferContextual(
context -> this.<DaprProtos.GetBulkStateResponse>createMono(it -> intercept(context, asyncStub)
.getBulkState(envelope, it)
)
).map(
it ->
it
context -> this.<DaprProtos.GetBulkStateResponse>createMono(it -> intercept(context, asyncStub)
.getBulkState(envelope, it)
)
).map(
it ->
it
.getItemsList()
.stream()
.map(b -> {
Expand Down Expand Up @@ -705,8 +696,8 @@ public Mono<Map<String, String>> getSecret(GetSecretRequest request) {
}

DaprProtos.GetSecretRequest.Builder requestBuilder = DaprProtos.GetSecretRequest.newBuilder()
.setStoreName(secretStoreName)
.setKey(key);
.setStoreName(secretStoreName)
.setKey(key);

if (metadata != null) {
requestBuilder.putAllMetadata(metadata);
Expand Down Expand Up @@ -740,18 +731,18 @@ public Mono<Map<String, Map<String, String>>> getBulkSecret(GetBulkSecretRequest

return Mono.deferContextual(
context ->
this.<DaprProtos.GetBulkSecretResponse>createMono(
it -> intercept(context, asyncStub).getBulkSecret(envelope, it)
)
this.<DaprProtos.GetBulkSecretResponse>createMono(
it -> intercept(context, asyncStub).getBulkSecret(envelope, it)
)
).map(it -> {
Map<String, DaprProtos.SecretResponse> secretsMap = it.getDataMap();
if (secretsMap == null) {
return Collections.emptyMap();
}
return secretsMap
.entrySet()
.stream()
.collect(Collectors.toMap(Map.Entry::getKey, s -> s.getValue().getSecretsMap()));
.entrySet()
.stream()
.collect(Collectors.toMap(Map.Entry::getKey, s -> s.getValue().getSecretsMap()));
});
} catch (Exception ex) {
return DaprException.wrapMono(ex);
Expand Down Expand Up @@ -805,7 +796,7 @@ public <T> Mono<QueryStateResponse<T>> queryState(QueryStateRequest request, Typ
try {
return buildQueryStateKeyValue(v, type);
} catch (Exception e) {
throw DaprException.propagate(e);
throw DaprException.propagate(e);
}
})
.collect(Collectors.toList());
Expand Down Expand Up @@ -900,7 +891,7 @@ private Mono<Map<String, ConfigurationItem>> getConfiguration(DaprProtos.GetConf
Iterator<Map.Entry<String, CommonProtos.ConfigurationItem>> itr = it.getItems().entrySet().iterator();
while (itr.hasNext()) {
Map.Entry<String, CommonProtos.ConfigurationItem> entry = itr.next();
configMap.put(entry.getKey(), buildConfigurationItem(entry.getValue(), entry.getKey()));
configMap.put(entry.getKey(), buildConfigurationItem(entry.getValue(), entry.getKey()));
}
return Collections.unmodifiableMap(configMap);
}
Expand Down Expand Up @@ -934,15 +925,15 @@ public Flux<SubscribeConfigurationResponse> subscribeConfiguration(SubscribeConf
return this.<DaprProtos.SubscribeConfigurationResponse>createFlux(
it -> intercept(asyncStub).subscribeConfiguration(envelope, it)
).map(
it -> {
Map<String, ConfigurationItem> configMap = new HashMap<>();
Iterator<Map.Entry<String, CommonProtos.ConfigurationItem>> itr = it.getItemsMap().entrySet().iterator();
while (itr.hasNext()) {
Map.Entry<String, CommonProtos.ConfigurationItem> entry = itr.next();
configMap.put(entry.getKey(), buildConfigurationItem(entry.getValue(), entry.getKey()));
it -> {
Map<String, ConfigurationItem> configMap = new HashMap<>();
Iterator<Map.Entry<String, CommonProtos.ConfigurationItem>> itr = it.getItemsMap().entrySet().iterator();
while (itr.hasNext()) {
Map.Entry<String, CommonProtos.ConfigurationItem> entry = itr.next();
configMap.put(entry.getKey(), buildConfigurationItem(entry.getValue(), entry.getKey()));
}
return new SubscribeConfigurationResponse(it.getId(), Collections.unmodifiableMap(configMap));
}
return new SubscribeConfigurationResponse(it.getId(), Collections.unmodifiableMap(configMap));
}
);
} catch (Exception ex) {
return DaprException.wrapFlux(ex);
Expand Down Expand Up @@ -990,8 +981,8 @@ public Mono<UnsubscribeConfigurationResponse> unsubscribeConfiguration(Unsubscri
private ConfigurationItem buildConfigurationItem(
CommonProtos.ConfigurationItem configurationItem, String key) {
return new ConfigurationItem(
key,
configurationItem.getValue(),
key,
configurationItem.getValue(),
configurationItem.getVersion(),
configurationItem.getMetadataMap()
);
Expand Down
35 changes: 23 additions & 12 deletions sdk/src/main/java/io/dapr/client/DaprHttp.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
import reactor.util.context.ContextView;

import java.io.IOException;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Collections;
Expand Down Expand Up @@ -141,14 +142,9 @@ public int getStatusCode() {
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();

/**
* Hostname used to communicate to Dapr's HTTP endpoint.
* Endpoint used to communicate to Dapr's HTTP endpoint.
*/
private final String hostname;

/**
* Port used to communicate to Dapr's HTTP endpoint.
*/
private final int port;
private final URI uri;

/**
* Http client used for all API calls.
Expand All @@ -163,8 +159,18 @@ public int getStatusCode() {
* @param httpClient RestClient used for all API calls in this new instance.
*/
DaprHttp(String hostname, int port, OkHttpClient httpClient) {
this.hostname = hostname;
this.port = port;
this.uri = URI.create(DEFAULT_HTTP_SCHEME + "://" + hostname + ":" + port);
this.httpClient = httpClient;
}

/**
* Creates a new instance of {@link DaprHttp}.
*
* @param uri Endpoint for calling Dapr. (e.g. "https://my-dapr-api.company.com")
* @param httpClient RestClient used for all API calls in this new instance.
*/
DaprHttp(String uri, OkHttpClient httpClient) {
this.uri = URI.create(uri);
this.httpClient = httpClient;
}

Expand Down Expand Up @@ -273,9 +279,14 @@ private CompletableFuture<Response> doInvokeApi(String method,
body = RequestBody.Companion.create(content, mediaType);
}
HttpUrl.Builder urlBuilder = new HttpUrl.Builder();
urlBuilder.scheme(DEFAULT_HTTP_SCHEME)
.host(this.hostname)
.port(this.port);
urlBuilder.scheme(uri.getScheme())
.host(uri.getHost());
if (uri.getPort() > 0) {
urlBuilder.port(uri.getPort());
}
if (uri.getPath() != null) {
urlBuilder.addPathSegments(uri.getPath());
}
for (String pathSegment : pathSegments) {
urlBuilder.addPathSegment(pathSegment);
}
Expand Down
5 changes: 5 additions & 0 deletions sdk/src/main/java/io/dapr/client/DaprHttpBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,11 @@ private DaprHttp buildDaprHttp() {
}
}

String endpoint = Properties.HTTP_ENDPOINT.get();
if ((endpoint != null) && !endpoint.isEmpty()) {
return new DaprHttp(endpoint, OK_HTTP_CLIENT);
}

return new DaprHttp(Properties.SIDECAR_IP.get(), Properties.HTTP_PORT.get(), OK_HTTP_CLIENT);
}
}
Loading