feat: Configuration API

daprdocs/content/en/js-sdk-docs/js-server/_index.md

@@ -168,7 +168,7 @@ const serverHost = "127.0.0.1";
 const serverPort = "5051";
 
 async function start() {
-  const server = new DaprServer(serverHost, serverPort, daprHost, daprPort);;
+  const server = new DaprServer(serverHost, serverPort, daprHost, daprPort);
 
   const bindingName = "my-binding-name";
 
@@ -185,6 +185,58 @@ start().catch((e) => {
 
 > For a full guide on output bindings visit [How-To: Use bindings]({{< ref howto-bindings.md >}}).
 
+### Configuration API
+
+> 💡 The configuration API is currently only available through gRPC
+
+#### Getting a configuration value
+
+```javascript
+import { DaprServer } from "dapr-client";
+
+const daprHost = "127.0.0.1"; 
+const daprPort = "3500"; 
+const serverHost = "127.0.0.1";
+const serverPort = "5051";
+
+async function start() {
+    const client = new DaprClient(daprHost, daprPort, CommunicationProtocolEnum.GRPC);
+    const config = await client.configuration.get("config-redis", ["myconfigkey1", "myconfigkey2"]);
+}
+
+start().catch((e) => {
+  console.error(e);
+  process.exit(1);
+});
+```
+
+#### Subscribing to Key Changes
+
+```javascript
+import { DaprServer } from "dapr-client";
+
+const daprHost = "127.0.0.1"; 
+const daprPort = "3500"; 
+const serverHost = "127.0.0.1";
+const serverPort = "5051";
+
+async function start() {
+    const client = new DaprClient(daprHost, daprPort, CommunicationProtocolEnum.GRPC);
+    const stream = await client.configuration.subscribeWithKeys("config-redis", ["myconfigkey1", "myconfigkey2"], () => {
+        // Received a key update
+    });
+
+    // When you are ready to stop listening, call the following
+    await stream.close();
+}
+
+start().catch((e) => {
+  console.error(e);
+  process.exit(1);
+});
+```
+
+
 ## Related links
 
 - [JavaScript SDK examples](https://github.com/dapr/js-sdk/tree/master/examples)

src/implementation/Client/DaprClient.ts

@@ -19,6 +19,7 @@ import IClientSecret from '../../interfaces/Client/IClientSecret';
 import IClientHealth from '../../interfaces/Client/IClientHealth';
 import IClientMetadata from '../../interfaces/Client/IClientMetadata';
 import IClientSidecar from '../../interfaces/Client/IClientSidecar';
+import IClientConfiguration from '../../interfaces/Client/IClientConfiguration';
 import IClientActorBuilder from '../../interfaces/Client/IClientActorBuilder';
 import IClient from '../../interfaces/Client/IClient';
 
@@ -30,6 +31,7 @@ import GRPCClientSecret from './GRPCClient/secret';
 import GRPCClientHealth from './GRPCClient/health';
 import GRPCClientMetadata from './GRPCClient/metadata';
 import GRPCClientSidecar from './GRPCClient/sidecar';
+import GRPCClientConfiguration from './GRPCClient/configuration';
 import GRPCClientActor from './GRPCClient/actor';
 import GRPCClient from './GRPCClient/GRPCClient';
 
@@ -41,6 +43,7 @@ import HTTPClientSecret from './HTTPClient/secret';
 import HTTPClientHealth from './HTTPClient/health';
 import HTTPClientMetadata from './HTTPClient/metadata';
 import HTTPClientSidecar from './HTTPClient/sidecar';
+import HTTPClientConfiguration from './HTTPClient/configuration';
 import HTTPClientActor from './HTTPClient/actor';
 import HTTPClient from './HTTPClient/HTTPClient';
 
@@ -63,6 +66,7 @@ export default class DaprClient {
   readonly health: IClientHealth;
   readonly metadata: IClientMetadata;
   readonly sidecar: IClientSidecar;
+  readonly configuration: IClientConfiguration;
   readonly actor: IClientActorBuilder;
 
   constructor(
@@ -97,6 +101,7 @@ export default class DaprClient {
         this.health = new GRPCClientHealth(client);
         this.metadata = new GRPCClientMetadata(client);
         this.sidecar = new GRPCClientSidecar(client);
+        this.configuration = new GRPCClientConfiguration(client);
         this.actor = new GRPCClientActor(client); // we use a abstractor here since we interface through a builder with the Actor Runtime
         break;
       }
@@ -113,6 +118,7 @@ export default class DaprClient {
         this.health = new HTTPClientHealth(client);
         this.metadata = new HTTPClientMetadata(client);
         this.sidecar = new HTTPClientSidecar(client);
+        this.configuration = new HTTPClientConfiguration(client);
         this.actor = new HTTPClientActor(client); // we use a abstractor here since we interface through a builder with the Actor Runtime
         break;
       }

src/implementation/Client/GRPCClient/configuration.ts

@@ -0,0 +1,152 @@
+/*
+Copyright 2022 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 './GRPCClient';
+import * as grpc from "@grpc/grpc-js";
+import { GetConfigurationRequest, GetConfigurationResponse, SubscribeConfigurationRequest, SubscribeConfigurationResponse, UnsubscribeConfigurationRequest, UnsubscribeConfigurationResponse } from '../../../proto/dapr/proto/runtime/v1/dapr_pb';
+import IClientConfiguration from '../../../interfaces/Client/IClientConfiguration';
+import { KeyValueType } from '../../../types/KeyValue.type';
+import { GetConfigurationResponse as GetConfigurationResponseResult } from '../../../types/configuration/GetConfigurationResponse';
+import { SubscribeConfigurationResponse as SubscribeConfigurationResponseResult } from '../../../types/configuration/SubscribeConfigurationResponse';
+import { SubscribeConfigurationCallback } from '../../../types/configuration/SubscribeConfigurationCallback';
+import { SubscribeConfigurationStream } from '../../../types/configuration/SubscribeConfigurationStream';
+
+export default class GRPCClientConfiguration implements IClientConfiguration {
+  client: GRPCClient;
+
+  constructor(client: GRPCClient) {
+    this.client = client;
+  }
+
+  async get(storeName: string, keys: string[], metadataObj?: KeyValueType): Promise<GetConfigurationResponseResult> {
+    const metadata = new grpc.Metadata();
+
+    const msg = new GetConfigurationRequest();
+    msg.setStoreName(storeName);
+
+    if (keys && keys.length > 0) {
+      msg.setKeysList(keys.filter(i => i !== ""));
+    }
+
+    if (metadataObj) {
+      for (const [key, value] of Object.entries(metadataObj)) {
+        metadata.add(key, value);
+      }
+    }
+
+    return new Promise((resolve, reject) => {
+      const client = this.client.getClient();
+      client.getConfigurationAlpha1(msg, metadata, (err, res: GetConfigurationResponse) => {
+        if (err) {
+          return reject(err);
+        }
+
+        const wrapped: GetConfigurationResponseResult = {
+          items: res.getItemsList().map((item) => ({
+            key: item.getKey(),
+            value: item.getValue(),
+            version: item.getVersion(),
+            metadata: item.getMetadataMap().toObject().reduce((result: object, [key, value]) => {
+              // @ts-ignore
+              result[key] = value;
+              return result
+            }, {}),
+          }))
+        }
+
+        return resolve(wrapped);
+      });
+    });
+  }
+
+  async subscribe(storeName: string, cb: SubscribeConfigurationCallback): Promise<SubscribeConfigurationStream> {
+    return this._subscribe(storeName, cb)
+  }
+
+  async subscribeWithKeys(storeName: string, keys: string[], cb: SubscribeConfigurationCallback): Promise<SubscribeConfigurationStream> {
+    return this._subscribe(storeName, cb, keys)
+  }
+
+  async subscribeWithMetadata(storeName: string, keys: string[], metadata: KeyValueType, cb: SubscribeConfigurationCallback): Promise<SubscribeConfigurationStream> {
+    return this._subscribe(storeName, cb, keys, metadata)
+  }
+
+  async _subscribe(storeName: string, cb: SubscribeConfigurationCallback, keys?: string[], metadataObj?: KeyValueType): Promise<SubscribeConfigurationStream> {
+    const metadata = new grpc.Metadata();
+
+    const msg = new SubscribeConfigurationRequest();
+    msg.setStoreName(storeName);
+
+    if (keys && keys.length > 0) {
+      msg.setKeysList(keys.filter(i => i !== ""));
+    } else {
+      msg.setKeysList([]);
+    }
+
+    if (metadataObj) {
+      for (const [key, value] of Object.entries(metadataObj)) {
+        metadata.add(key, value);
+      }
+    }
+
+    const client = this.client.getClient();
+
+    // Open a stream. Note that this is a never-ending stream 
+    // and will stay open as long as the client is open
+    // we will thus create a set with our listeners so we don't 
+    // break on multi listeners
+    const stream = client.subscribeConfigurationAlpha1(msg, metadata);
+    let streamId: string;
+
+    stream.on("data", async (data: SubscribeConfigurationResponse) => {
+      streamId = data.getId();
+
+      const wrapped: SubscribeConfigurationResponseResult = {
+        items: data.getItemsList().map((item) => ({
+          key: item.getKey(),
+          value: item.getValue(),
+          version: item.getVersion(),
+          metadata: item.getMetadataMap().toObject().reduce((result: object, [key, value]) => {
+            // @ts-ignore
+            result[key] = value;
+            return result
+          }, {}),
+        }))
+      }
+
+      await cb(wrapped);
+    });
+
+    return {
+      stop: async () => {
+        return new Promise((resolve, reject) => {
+          const req = new UnsubscribeConfigurationRequest();
+          req.setStoreName(storeName);
+          req.setId(streamId);
+
+          client.unsubscribeConfigurationAlpha1(req, (err, res: UnsubscribeConfigurationResponse) => {
+            if (err || !res.getOk()) {
+              return reject(res.getMessage());
+            }
+
+            // Clean up the node.js event emitter
+            stream.removeAllListeners();
+            stream.destroy();
+
+            return resolve();
+          });
+        })
+      }
+    };
+  }
+}

src/implementation/Client/HTTPClient/configuration.ts

@@ -0,0 +1,43 @@
+/*
+Copyright 2022 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 IClientConfiguration from '../../../interfaces/Client/IClientConfiguration';
+import { KeyValueType } from '../../../types/KeyValue.type';
+import { GetConfigurationResponse as GetConfigurationResponseResult } from '../../../types/configuration/GetConfigurationResponse';
+import HTTPClient from './HTTPClient';
+import { SubscribeConfigurationCallback } from '../../../types/configuration/SubscribeConfigurationCallback';
+import { SubscribeConfigurationStream } from '../../../types/configuration/SubscribeConfigurationStream';
+
+export default class HTTPClientConfiguration implements IClientConfiguration {
+  client: HTTPClient;
+
+  constructor(client: HTTPClient) {
+    this.client = client;
+  }
+
+  async subscribe(_storeName: string, _cb: SubscribeConfigurationCallback): Promise<SubscribeConfigurationStream> {
+    throw new Error('HTTP is currently not supported.');
+  }
+
+  async subscribeWithKeys(_storeName: string, _keys: string[], _cb: SubscribeConfigurationCallback): Promise<SubscribeConfigurationStream> {
+    throw new Error('HTTP is currently not supported.');
+  }
+
+  async subscribeWithMetadata(_storeName: string, _keys: string[], _metadata: KeyValueType, _cb: SubscribeConfigurationCallback): Promise<SubscribeConfigurationStream> {
+    throw new Error('HTTP is currently not supported.');
+  }
+
+  async get(_storeName: string, _keys: string[], _metadata?: KeyValueType): Promise<GetConfigurationResponseResult> {
+    throw new Error('HTTP is currently not supported.');
+  }
+}

src/implementation/Server/GRPCServer/actor.ts

@@ -25,18 +25,18 @@ export default class GRPCServerActor implements IServerActor {
   }
 
   deactivateActor(_actorType: string, _actorId: string): Promise<void> {
-    throw new Error('Method not implemented.');
+    throw new Error('GRPC is currently not supported.');
   }
 
   init(): Promise<void> {
-    throw new Error('Method not implemented.');
+    throw new Error('GRPC is currently not supported.');
   }
 
   getRegisteredActors(): Promise<string[]> {
-    throw new Error('Method not implemented.');
+    throw new Error('GRPC is currently not supported.');
   }
 
   registerActor<T extends AbstractActor>(_cls: Class<T>): Promise<void> {
-    throw new Error('Method not implemented.');
+    throw new Error('GRPC is currently not supported.');
   }
 }

src/interfaces/Client/IClientConfiguration.ts

@@ -0,0 +1,25 @@
+/*
+Copyright 2022 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 { GetConfigurationResponse } from "../../types/configuration/GetConfigurationResponse";
+import { SubscribeConfigurationCallback } from "../../types/configuration/SubscribeConfigurationCallback";
+import { SubscribeConfigurationStream } from "../../types/configuration/SubscribeConfigurationStream";
+import { KeyValueType } from "../../types/KeyValue.type";
+
+export default interface IClientConfiguration {
+  // https://github.com/dapr/dapr/blob/master/dapr/proto/runtime/v1/dapr.proto#L90
+  get(storeName: string, keys?: string[], metadata?: KeyValueType): Promise<GetConfigurationResponse>;
+  subscribe(storeName: string, cb: SubscribeConfigurationCallback): Promise<SubscribeConfigurationStream>;
+  subscribeWithKeys(storeName: string, keys: string[], cb: SubscribeConfigurationCallback): Promise<SubscribeConfigurationStream>;
+  subscribeWithMetadata(storeName: string, keys: string[], metadata: KeyValueType, cb: SubscribeConfigurationCallback): Promise<SubscribeConfigurationStream>;
+}