Pass X-Api-Key for GRPC streaming client (#14)

https://proxima-one.atlassian.net/browse/PRX-1331

---------

Co-authored-by: Denis Kanonik <[email protected]>

src/client/proximaStreamClient.ts

@@ -12,13 +12,15 @@ export class ProximaStreamClient {
   private readonly registry: StreamRegistry;
   private readonly clients: Record<string, StreamDBConsumerClient>;
   private readonly offsetsCache: NodeCache;
+  private readonly apiKey?: string;
 
   public constructor(
     private readonly options: StreamClientOptions = {
       registry: new StreamRegistryClient(),
     }
   ) {
     this.registry = options.registry ?? new StreamRegistryClient();
+    this.apiKey = options.apiKey ?? this.registry.getApiKey();
     this.clients = {};
     this.offsetsCache = new NodeCache({ maxKeys: 10 * 1000 * 1000 });
   }
@@ -133,11 +135,15 @@ export class ProximaStreamClient {
   private getStreamConsumerClient(endpoint: string) {
     return (
       this.clients[endpoint] ??
-      (this.clients[endpoint] = new StreamDBConsumerClient(endpoint))
+      (this.clients[endpoint] = new StreamDBConsumerClient(
+        endpoint,
+        this.apiKey
+      ))
     );
   }
 }
 
 export interface StreamClientOptions {
   registry?: StreamRegistry;
+  apiKey?: string;
 }

src/client/singleStreamDbRegistry.ts

@@ -4,6 +4,10 @@ import { Offset, StreamEndpoint, StreamStats } from "../model";
 export class SingleStreamDbRegistry implements StreamRegistry {
   public constructor(private readonly streamDbUrl: string) {}
 
+  public getApiKey(): string | undefined {
+    return undefined;
+  }
+
   public async getStreamEndpoints(
     // eslint-disable-next-line @typescript-eslint/no-unused-vars
     streamName: string,

src/client/streamRegistry.ts

@@ -2,4 +2,5 @@ import { Offset, StreamEndpoint } from "../model";
 
 export interface StreamRegistry {
   getStreamEndpoints(stream: string, offset: Offset): Promise<StreamEndpoint[]>;
+  getApiKey(): string | undefined;
 }

src/client/streamRegistryClient.ts

@@ -22,6 +22,7 @@ const DefaultRegistryClientOptions: StreamRegistryOptions = {
 
 export class StreamRegistryClient implements StreamRegistry {
   private readonly client: Axios;
+  private readonly apiKey?: string;
 
   public constructor(
     private readonly options: StreamRegistryOptions = DefaultRegistryClientOptions
@@ -32,6 +33,11 @@ export class StreamRegistryClient implements StreamRegistry {
         (status >= 200 && status < 300) || status == 404,
       headers: options.apiKey ? { "x-api-key": options.apiKey } : {},
     });
+    this.apiKey = options.apiKey;
+  }
+
+  public getApiKey(): string | undefined {
+    return this.apiKey;
   }
 
   public async getStreamEndpoints(

src/streamdb/consumerClient.ts

@@ -13,12 +13,26 @@ import { PausableStream, StreamController } from "../lib/pausableStream";
 export class StreamDBConsumerClient {
   private consumer: StreamConsumerServiceClient;
 
-  constructor(uri: string) {
-    const secure = uri.includes(":443");
-    const credentials = secure
+  constructor(uri: string, apiKey?: string) {
+    const isSecure = uri.includes(":443");
+    const channelCredentials = isSecure
       ? grpc.credentials.createSsl()
       : grpc.credentials.createInsecure();
-    this.consumer = new StreamConsumerServiceClient(uri, credentials, {
+    const callCredentials = grpc.credentials.createFromMetadataGenerator(
+      (opts, callback) => {
+        const metadata = new grpc.Metadata();
+        if (apiKey !== undefined) {
+          metadata.add("x-api-key", apiKey);
+        }
+        callback(null, metadata);
+      }
+    );
+    const mergedCredentials = grpc.credentials.combineChannelCredentials(
+      channelCredentials,
+      callCredentials
+    );
+
+    this.consumer = new StreamConsumerServiceClient(uri, mergedCredentials, {
       "grpc.keepalive_timeout_ms": 100 * 1000,
       "grpc.keepalive_time_ms": 100 * 1000,
       "grpc.keepalive_permit_without_calls": 1,