feat: Support for the DAPR_HTTP_ENDPOINT and DAPR_GRPC_ENDPOINT environment variables. Adds support for DAPR_API_TOKEN to gRPC client

.gitignore

@@ -1,5 +1,5 @@
 ### vscode ###
-.vscode/*
+.vscode
 !.vscode/settings.json
 !.vscode/tasks.json
 !.vscode/launch.json
@@ -144,4 +144,11 @@ temp/
 build/
 
 # version file is auto-generated
-src/version.ts
+src/version.ts
+
+# OSX
+/.DS_Store
+
+# JetBrains
+/.idea
+

src/implementation/Client/GRPCClient/GRPCClient.ts

@@ -52,8 +52,32 @@ export default class GRPCClient implements IClient {
     return this.clientCredentials;
   }
 
-  private generateChannelOptions(): Record<string, string | number> {
-    const options: Record<string, string | number> = {};
+  private generateCredentials(): grpc.ChannelCredentials {
+    if (this.options.daprHost.startsWith("https")) {
+      return grpc.ChannelCredentials.createSsl();
+    }
+    return grpc.ChannelCredentials.createInsecure();
+  }
+
+  private generateClient(host: string, port: string, credentials: grpc.ChannelCredentials): GrpcDaprClient {
+    return new GrpcDaprClient(GRPCClient.getEndpoint(host, port), credentials, this.generateChannelOptions());
+  }
+
+  // The grpc client doesn't allow http:// or https:// for grpc connections
+  // so we need to remove it, if it exists
+  static getEndpoint(host: string, port: string): string {
+    let endpoint = `${host}:${port}`;
+    const parts = endpoint.split("://");
+    if (parts.length > 1 && parts[0].startsWith("http")) {
+      endpoint = parts[1];
+    }
+
+    return endpoint;
+  }
+
+  private generateChannelOptions(): Partial<grpc.ClientOptions> {
+    // const options: Record<string, string | number> = {};
+    let options: Partial<grpc.ClientOptions> = {};
 
     // See: GRPC_ARG_MAX_SEND_MESSAGE_LENGTH, it is in bytes
     // https://grpc.github.io/grpc/core/group__grpc__arg__keys.html#ga813f94f9ac3174571dd712c96cdbbdc1
@@ -67,20 +91,28 @@ export default class GRPCClient implements IClient {
     // Add user agent
     options["grpc.primary_user_agent"] = "dapr-sdk-js/v" + SDK_VERSION;
 
-    return options;
-  }
-
-  private generateClient(host: string, port: string, credentials: grpc.ChannelCredentials): GrpcDaprClient {
-    const options = this.generateChannelOptions();
-    const client = new GrpcDaprClient(`${host}:${port}`, credentials, options);
+    // Add interceptors if we have an API token
+    if (this.options.daprApiToken !== "") {
+      options = {
+        interceptors: [this.generateInterceptors()],
+        ...options,
+      };
+    }
 
-    return client;
+    return options;
   }
 
-  // @todo: look into making secure credentials
-  private generateCredentials(): grpc.ChannelCredentials {
-    const credsChannel = grpc.ChannelCredentials.createInsecure();
-    return credsChannel;
+  private generateInterceptors(): (options: any, nextCall: any) => grpc.InterceptingCall {
+    return (options: any, nextCall: any) => {
+      return new grpc.InterceptingCall(nextCall(options), {
+        start: (metadata, listener, next) => {
+          if (metadata.get("dapr-api-token").length == 0) {
+            metadata.add("dapr-api-token", this.options.daprApiToken as grpc.MetadataValue);
+          }
+          next(metadata, listener);
+        },
+      });
+    };
   }
 
   setIsInitialized(isInitialized: boolean): void {

src/utils/Client.util.ts

@@ -263,9 +263,32 @@ export function getClientOptions(
   defaultLoggerOptions: LoggerOptions | undefined,
 ): DaprClientOptions {
   const clientCommunicationProtocol = clientoptions?.communicationProtocol ?? defaultCommunicationProtocol;
+
+  // We decide the host/port/endpoint here
+  let host: string;
+  let port: string;
+  let daprEndpoint = "";
+  if (clientCommunicationProtocol == CommunicationProtocolEnum.HTTP) {
+    daprEndpoint = Settings.getDefaultHttpEndpoint();
+  } else if (clientCommunicationProtocol == CommunicationProtocolEnum.GRPC) {
+    daprEndpoint = Settings.getDefaultGrpcEndpoint();
+  }
+
+  if (clientoptions?.daprHost || clientoptions?.daprPort) {
+    host = clientoptions?.daprHost ?? Settings.getDefaultHost();
+    port = clientoptions?.daprPort ?? Settings.getDefaultPort(clientCommunicationProtocol);
+  } else if (daprEndpoint != "") {
+    const [scheme, fqdn, p] = parseEndpoint(daprEndpoint);
+    host = `${scheme}://${fqdn}`;
+    port = p.toString();
+  } else {
+    host = Settings.getDefaultHost();
+    port = Settings.getDefaultPort(clientCommunicationProtocol);
+  }
+
   return {
-    daprHost: clientoptions?.daprHost ?? Settings.getDefaultHost(),
-    daprPort: clientoptions?.daprPort ?? Settings.getDefaultPort(clientCommunicationProtocol),
+    daprHost: host,
+    daprPort: port,
     communicationProtocol: clientCommunicationProtocol,
     isKeepAlive: clientoptions?.isKeepAlive,
     logger: clientoptions?.logger ?? defaultLoggerOptions,
@@ -274,3 +297,68 @@ export function getClientOptions(
     maxBodySizeMb: clientoptions?.maxBodySizeMb,
   };
 }
+
+/**
+ * Scheme, fqdn and port
+ */
+type EndpointTuple = [string, string, number];
+
+/**
+ * Parses an endpoint to scheme, fqdn and port
+ * @param address Endpoint address
+ * @returns EndpointTuple (scheme, fqdn, port)
+ */
+export function parseEndpoint(address: string): EndpointTuple {
+  let scheme = "http";
+  let fqdn = "localhost";
+  let port = 80;
+  let addr = address;
+
+  const addrList = address.split("://");
+
+  if (addrList.length === 2) {
+    // A scheme was explicitly specified
+    scheme = addrList[0];
+    if (scheme === "https") {
+      port = 443;
+    }
+    addr = addrList[1];
+  }
+
+  const addrParts = addr.split(":");
+  if (addrParts.length === 2) {
+    // A port was explicitly specified
+    if (addrParts[0].length > 0) {
+      fqdn = addrParts[0];
+    }
+    // Account for Endpoints of the type http://localhost:3500/v1.0/invoke
+    const portParts = addrParts[1].split("/");
+    port = parseInt(portParts[0], 10);
+  } else if (addrParts.length === 1) {
+    // No port was specified
+    // Account for Endpoints of the type :3500/v1.0/invoke
+    const fqdnParts = addrParts[0].split("/");
+    fqdn = fqdnParts[0];
+  } else {
+    // IPv6 address
+    const ipv6Parts = addr.split("]:");
+    if (ipv6Parts.length === 2) {
+      // A port was explicitly specified
+      fqdn = ipv6Parts[0].replace("[", "");
+      const portParts = ipv6Parts[1].split("/");
+      port = parseInt(portParts[0], 10);
+    } else if (ipv6Parts.length === 1) {
+      // No port was specified
+      const fqdnParts = ipv6Parts[0].split("/");
+      fqdn = fqdnParts[0].replace("[", "").replace("]", "");
+    } else {
+      throw new Error(`Invalid address: ${address}`);
+    }
+  }
+
+  if (isNaN(port)) {
+    throw new Error(`Invalid port: ${port}`);
+  }
+
+  return [scheme, fqdn, port];
+}

src/utils/Settings.util.ts

@@ -20,6 +20,8 @@ export class Settings {
   private static readonly defaultHttpPort: string = "3500";
   private static readonly defaultGrpcAppPort: string = "50000";
   private static readonly defaultGrpcPort: string = "50001";
+  private static readonly defaultHttpEndpoint: string = "";
+  private static readonly defaultGrpcEndpoint: string = "";
   private static readonly defaultCommunicationProtocol: CommunicationProtocolEnum = CommunicationProtocolEnum.HTTP;
   private static readonly defaultKeepAlive: boolean = true;
   private static readonly defaultStateGetBulkParallelism: number = 10;
@@ -85,6 +87,22 @@ export class Settings {
     return process.env.APP_PORT ?? Settings.defaultGrpcAppPort;
   }
 
+  static getDefaultHttpEndpoint(): string {
+    if (process.env.DAPR_HTTP_ENDPOINT && process.env.DAPR_HTTP_ENDPOINT !== "") {
+      return process.env.DAPR_HTTP_ENDPOINT as string;
+    } else {
+      return Settings.defaultHttpEndpoint;
+    }
+  }
+
+  static getDefaultGrpcEndpoint(): string {
+    if (process.env.DAPR_GRPC_ENDPOINT && process.env.DAPR_GRPC_ENDPOINT !== "") {
+      return process.env.DAPR_GRPC_ENDPOINT as string;
+    } else {
+      return Settings.defaultGrpcEndpoint;
+    }
+  }
+
   /**
    * Gets the default port that the application is listening on.
    * @param communicationProtocolEnum communication protocol

test/e2e/common/client.test.ts

@@ -21,6 +21,7 @@ import {
 } from "../../../src";
 import { sleep } from "../../../src/utils/NodeJS.util";
 import { LockStatus } from "../../../src/types/lock/UnlockResponse";
+import { Settings } from "../../../src/utils/Settings.util";
 
 const daprHost = "127.0.0.1";
 const daprGrpcPort = "50000";
@@ -557,3 +558,128 @@ describe("common/client", () => {
     });
   });
 });
+
+describe("http/client with environment variables", () => {
+  let client: DaprClient;
+
+  // We need to start listening on some endpoints already
+  // this because Dapr is not dynamic and registers endpoints on boot
+  // we put a timeout of 10s since it takes around 4s for Dapr to boot up
+
+  afterAll(async () => {
+    await client.stop();
+  });
+
+  it("should give preference to host and port in constructor arguments over endpoint environment variables ", async () => {
+    process.env.DAPR_HTTP_ENDPOINT = "https://httpdomain.com";
+    process.env.DAPR_GRPC_ENDPOINT = "https://grpcdomain.com";
+
+    client = new DaprClient({
+      daprHost,
+      daprPort: daprHttpPort,
+      communicationProtocol: CommunicationProtocolEnum.HTTP,
+      isKeepAlive: false,
+    });
+
+    expect(client.options.daprHost).toEqual(daprHost);
+    expect(client.options.daprPort).toEqual(daprHttpPort);
+
+    client = new DaprClient({
+      daprHost,
+      daprPort: daprGrpcPort,
+      communicationProtocol: CommunicationProtocolEnum.GRPC,
+      isKeepAlive: false,
+    });
+
+    expect(client.options.daprHost).toEqual(daprHost);
+    expect(client.options.daprPort).toEqual(daprGrpcPort);
+  });
+
+  it("should give preference to port with no host in constructor arguments over environment variables ", async () => {
+    process.env.DAPR_HTTP_ENDPOINT = "https://httpdomain.com";
+    process.env.DAPR_GRPC_ENDPOINT = "https://grpcdomain.com";
+
+    client = new DaprClient({
+      daprPort: daprHttpPort,
+      communicationProtocol: CommunicationProtocolEnum.HTTP,
+      isKeepAlive: false,
+    });
+
+    expect(client.options.daprHost).toEqual(Settings.getDefaultHost());
+    expect(client.options.daprPort).toEqual(daprHttpPort);
+
+    client = new DaprClient({
+      daprPort: daprGrpcPort,
+      communicationProtocol: CommunicationProtocolEnum.GRPC,
+      isKeepAlive: false,
+    });
+
+    expect(client.options.daprHost).toEqual(Settings.getDefaultHost());
+    expect(client.options.daprPort).toEqual(daprGrpcPort);
+  });
+
+  it("should give preference to host with no port in constructor arguments over environment variables ", async () => {
+    process.env.DAPR_HTTP_ENDPOINT = "https://httpdomain.com";
+    process.env.DAPR_GRPC_ENDPOINT = "https://grpcdomain.com";
+
+    client = new DaprClient({
+      daprHost: daprHost,
+      communicationProtocol: CommunicationProtocolEnum.HTTP,
+      isKeepAlive: false,
+    });
+
+    expect(client.options.daprHost).toEqual(daprHost);
+    expect(client.options.daprPort).toEqual(Settings.getDefaultPort(CommunicationProtocolEnum.HTTP));
+
+    client = new DaprClient({
+      daprHost: daprHost,
+      communicationProtocol: CommunicationProtocolEnum.GRPC,
+      isKeepAlive: false,
+    });
+
+    expect(client.options.daprHost).toEqual(daprHost);
+    expect(client.options.daprPort).toEqual(Settings.getDefaultPort(CommunicationProtocolEnum.GRPC));
+  });
+
+  it("should use environment variable endpoint for HTTP", async () => {
+    process.env.DAPR_HTTP_ENDPOINT = "https://httpdomain.com";
+    client = new DaprClient({
+      communicationProtocol: CommunicationProtocolEnum.HTTP,
+      isKeepAlive: false,
+    });
+
+    expect(client.options.daprHost).toEqual("https://httpdomain.com");
+    expect(client.options.daprPort).toEqual("443");
+  });
+
+  it("should use environment variable endpoint for GRPC", async () => {
+    process.env.DAPR_GRPC_ENDPOINT = "https://grpcdomain.com";
+    client = new DaprClient({
+      communicationProtocol: CommunicationProtocolEnum.GRPC,
+      isKeepAlive: false,
+    });
+
+    expect(client.options.daprHost).toEqual("https://grpcdomain.com");
+    expect(client.options.daprPort).toEqual("443");
+  });
+
+  it("should use default host and port when no other parameters provided", async () => {
+    process.env.DAPR_HTTP_ENDPOINT = "";
+    process.env.DAPR_GRPC_ENDPOINT = "";
+    client = new DaprClient({
+      communicationProtocol: CommunicationProtocolEnum.HTTP,
+      isKeepAlive: false,
+    });
+
+    expect(client.options.daprHost).toEqual(Settings.getDefaultHost());
+    expect(client.options.daprPort).toEqual(Settings.getDefaultPort(CommunicationProtocolEnum.HTTP));
+
+    client = new DaprClient({
+      communicationProtocol: CommunicationProtocolEnum.GRPC,
+      isKeepAlive: false,
+    });
+
+    expect(client.options.daprHost).toEqual(Settings.getDefaultHost());
+    expect(client.options.daprPort).toEqual(Settings.getDefaultPort(CommunicationProtocolEnum.GRPC));
+  });
+});

test/unit/grpc/GRPCClient.test.ts

@@ -0,0 +1,28 @@
+/*
+Copyright 2023 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+import { GRPCClient } from "../../../src";
+
+describe("grpc", () => {
+  it("should remove http and https from endpoint", () => {
+    const testCases = [
+      { host: "http://localhost", port: "5000", expected: "localhost:5000" },
+      { host: "https://localhost", port: "5000", expected: "localhost:5000" },
+      { host: "localhost", port: "5000", expected: "localhost:5000" },
+    ];
+
+    testCases.forEach((testCase) => {
+      expect(GRPCClient.getEndpoint(testCase.host, testCase.port)).toBe(testCase.expected);
+    });
+  });
+});