Support remote endpoint.

Signed-off-by: Artur Souza <asouza.pro@gmail.com>

Author: artursouza

sdk/src/main/java/io/dapr/client/DaprClientBuilder.java

@@ -24,6 +24,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.Closeable;
+import java.net.URI;
 
 /**
  * A builder for the DaprClient,
@@ -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");
+      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();
   }
 
   /**

sdk/src/main/java/io/dapr/client/DaprClientGrpc.java

@@ -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;
@@ -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;
@@ -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.
@@ -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);
   }
 
@@ -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);
   }
 
   /**
@@ -193,7 +185,6 @@ public Mono<Void> publishEvent(PublishEventRequest request) {
   }
 
   /**
-   *
    * {@inheritDoc}
    */
   @Override
@@ -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();
@@ -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(
@@ -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);
@@ -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);
@@ -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 -> {
@@ -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);
@@ -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);
@@ -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());
@@ -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);
         }
@@ -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);
@@ -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()
     );

sdk/src/main/java/io/dapr/client/DaprHttp.java

@@ -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;
@@ -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.
@@ -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;
   }
 
@@ -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);
     }

sdk/src/main/java/io/dapr/client/DaprHttpBuilder.java

@@ -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);
   }
 }