Make sure server starts before doing anything, instead of hardcoded 250ms we now check the Healthz endpoint

Signed-off-by: Xavier Geerinck <[email protected]>

Author: XavierGeerinck

src/implementation/Server/DaprServer.ts

@@ -16,9 +16,8 @@ import HTTPServerPubSub from './HTTPServer/pubsub';
 import HTTPServerBinding from './HTTPServer/binding';
 import HTTPServerInvoker from './HTTPServer/invoker';
 import HTTPServerActor from './HTTPServer/actor';
-import GRPCClient from '../Client/GRPCClient/GRPCClient';
-import HTTPClient from '../Client/HTTPClient/HTTPClient';
 import { DaprClientOptions } from '../../types/DaprClientOptions';
+import { DaprClient } from '../..';
 
 export default class DaprServer {
   // App details
@@ -27,14 +26,13 @@ export default class DaprServer {
   // Dapr Sidecar
   private readonly daprHost: string;
   private readonly daprPort: string;
-  private readonly communicationProtocol: CommunicationProtocolEnum;
 
   readonly daprServer: IServer;
   readonly pubsub: IServerPubSub;
   readonly binding: IServerBinding;
   readonly invoker: IServerInvoker;
   readonly actor: IServerActor;
-  readonly clientOptions: DaprClientOptions;
+  readonly client: DaprClient;
 
   constructor(
     serverHost = "127.0.0.1"
@@ -50,9 +48,9 @@ export default class DaprServer {
     this.serverPort = serverPort;
     this.daprHost = daprHost;
     this.daprPort = daprPort;
-    this.clientOptions = clientOptions;
 
-    this.communicationProtocol = communicationProtocol;
+    // Create a client to interface with the sidecar from the server side
+    this.client = new DaprClient(daprHost, daprPort, communicationProtocol, clientOptions);
 
     // If DAPR_SERVER_PORT was not set, we set it
     process.env.DAPR_SERVER_PORT = this.serverPort;
@@ -70,26 +68,24 @@ export default class DaprServer {
     // Builder
     switch (communicationProtocol) {
       case CommunicationProtocolEnum.GRPC: {
-        const client = new GRPCClient(daprHost, daprPort, this.clientOptions);
-        const server = new GRPCServer();
-
+        const server = new GRPCServer(this.client);
         this.daprServer = server;
-        this.pubsub = new GRPCServerPubSub(server, client);
-        this.binding = new GRPCServerBinding(server, client);
-        this.invoker = new GRPCServerInvoker(server, client);
-        this.actor = new GRPCServerActor(server, client);
+
+        this.pubsub = new GRPCServerPubSub(server);
+        this.binding = new GRPCServerBinding(server);
+        this.invoker = new GRPCServerInvoker(server);
+        this.actor = new GRPCServerActor(server);
         break;
       }
       case CommunicationProtocolEnum.HTTP:
       default: {
-        const client = new HTTPClient(daprHost, daprPort, this.clientOptions);
-        const server = new HTTPServer();
-
+        const server = new HTTPServer(this.client);
         this.daprServer = server;
-        this.pubsub = new HTTPServerPubSub(server, client);
-        this.binding = new HTTPServerBinding(server, client);
-        this.invoker = new HTTPServerInvoker(server, client);
-        this.actor = new HTTPServerActor(server, client);
+
+        this.pubsub = new HTTPServerPubSub(server);
+        this.binding = new HTTPServerBinding(server);
+        this.invoker = new HTTPServerInvoker(server);
+        this.actor = new HTTPServerActor(server, this.client);
         break;
       }
     }

src/implementation/Server/GRPCServer/GRPCServer.ts

@@ -2,6 +2,8 @@ import * as grpc from "@grpc/grpc-js";
 import GRPCServerImpl from "./GRPCServerImpl";
 import { AppCallbackService } from "../../../proto/dapr/proto/runtime/v1/appcallback_grpc_pb";
 import IServer from "../../../interfaces/Server/IServer";
+import GRPCClient from "../../Client/GRPCClient/GRPCClient";
+import { DaprClient } from "../../..";
 
 // eslint-disable-next-line
 export interface IServerType extends grpc.Server { }
@@ -15,12 +17,14 @@ export default class GRPCServer implements IServer {
   server: IServerType;
   serverImpl: IServerImplType;
   serverCredentials: grpc.ServerCredentials;
+  client: DaprClient;
 
-  constructor() {
+  constructor(client: DaprClient) {
     this.isInitialized = false;
 
     this.serverHost = "";
     this.serverPort = "";
+    this.client = client;
 
     // Create Server
     this.server = new grpc.Server();

src/implementation/Server/GRPCServer/actor.ts

@@ -7,13 +7,11 @@ import GRPCClient from '../../Client/GRPCClient/GRPCClient';
 // https://docs.dapr.io/reference/api/bindings_api/
 export default class GRPCServerActor implements IServerActor {
   server: GRPCServer;
-  client: GRPCClient;
 
-  constructor(server: GRPCServer, client: GRPCClient) {
-      this.server = server;
-      this.client = client;
+  constructor(server: GRPCServer) {
+    this.server = server;
   }
-  
+
   deactivateActor(actorType: string, actorId: string): Promise<void> {
     throw new Error('Method not implemented.');
   }
@@ -25,7 +23,7 @@ export default class GRPCServerActor implements IServerActor {
   getRegisteredActors(): Promise<string[]> {
     throw new Error('Method not implemented.');
   }
-  
+
   registerActor<T extends AbstractActor>(cls: Class<T>): Promise<void> {
     throw new Error('Method not implemented.');
   }

src/implementation/Server/GRPCServer/binding.ts

@@ -6,13 +6,11 @@ import GRPCClient from '../../Client/GRPCClient/GRPCClient';
 // https://docs.dapr.io/reference/api/bindings_api/
 export default class DaprBinding implements IServerBinding {
   server: GRPCServer;
-  client: GRPCClient;
 
-  constructor(server: GRPCServer, client: GRPCClient) {
-      this.server = server;
-      this.client = client;
+  constructor(server: GRPCServer) {
+    this.server = server;
   }
-  
+
   // Receive an input from an external system
   async receive(bindingName: string, cb: TypeDaprBindingCallback): Promise<any> {
     console.log(`Registering GRPC onBindingInput Handler: Binding = ${bindingName}`);

src/implementation/Server/GRPCServer/invoker.ts

@@ -8,13 +8,11 @@ import GRPCClient from '../../Client/GRPCClient/GRPCClient';
 // https://docs.dapr.io/reference/api/service_invocation_api/
 export default class DaprInvoker implements IServerInvoker {
   server: GRPCServer;
-  client: GRPCClient;
 
-  constructor(server: GRPCServer, client: GRPCClient) {
-      this.server = server;
-      this.client = client;
+  constructor(server: GRPCServer) {
+    this.server = server;
   }
-  
+
   async listen(methodName: string, cb: TypeDaprInvokerCallback, options: InvokerListenOptionsType = {}): Promise<any> {
     const httpMethod: HttpMethod = options?.method?.toLowerCase() as HttpMethod || HttpMethod.GET;
     console.log(`Registering onInvoke Handler ${httpMethod} /${methodName}`);

src/implementation/Server/GRPCServer/pubsub.ts

@@ -1,18 +1,15 @@
 import GRPCServer from "./GRPCServer";
 import { TypeDaprPubSubCallback } from "../../../types/DaprPubSubCallback.type";
 import IServerPubSub from "../../../interfaces/Server/IServerPubSub";
-import GRPCClient from "../../Client/GRPCClient/GRPCClient";
 
 // https://docs.dapr.io/reference/api/pubsub_api/
 export default class DaprPubSub implements IServerPubSub {
   server: GRPCServer;
-  client: GRPCClient;
 
-  constructor(server: GRPCServer, client: GRPCClient) {
-      this.server = server;
-      this.client = client;
+  constructor(server: GRPCServer) {
+    this.server = server;
   }
-  
+
   async subscribe(pubSubName: string, topic: string, cb: TypeDaprPubSubCallback): Promise<void> {
     console.log(`Registering onTopicEvent Handler: PubSub = ${pubSubName}; Topic = ${topic}`);
     this.server.getServerImpl().registerPubSubSubscriptionHandler(pubSubName, topic, cb);

src/implementation/Server/HTTPServer/HTTPServer.ts

@@ -3,6 +3,7 @@ import bodyParser from "body-parser";
 import HTTPServerImpl from "./HTTPServerImpl";
 import IServer from "../../../interfaces/Server/IServer";
 import * as NodeJSUtils from "../../../utils/NodeJS.util";
+import { DaprClient } from "../../..";
 
 // eslint-disable-next-line
 export interface IServerImplType extends HTTPServerImpl { }
@@ -16,11 +17,13 @@ export default class HTTPServer implements IServer {
   server: IServerType;
   serverAddress: string;
   serverImpl: IServerImplType;
-  serverStartupDelay = 250;
+  serverStartupDelay = 1000; // @todo: use health api https://docs.dapr.io/reference/api/health_api/
+  client: DaprClient;
 
-  constructor() {
+  constructor(client: DaprClient) {
     this.serverHost = "";
     this.serverPort = "";
+    this.client = client;
 
     this.isInitialized = false;
 
@@ -94,15 +97,29 @@ export default class HTTPServer implements IServer {
     this.server.get('/dapr/subscribe', (req, res) => {
       res.send(this.serverImpl.pubSubSubscriptionRoutes);
       console.log(`[Dapr API][PubSub] Registered ${this.serverImpl.pubSubSubscriptionRoutes.length} PubSub Subscriptions`);
-    })
+    });
 
     // We need to call the Singleton to start listening on the port, else Dapr will not pick it up correctly
     // Dapr will probe every 50ms to see if we are listening on our port: https://github.com/dapr/dapr/blob/a43712c97ead550ca2f733e9f7e7769ecb195d8b/pkg/runtime/runtime.go#L1694
     // if we are using actors we will change this to 4s to let the placement tables update
-    console.log(`Letting Dapr pick-up the server (${this.serverStartupDelay}ms)`);
-    await NodeJSUtils.sleep(this.serverStartupDelay);
+    let isHealthy = false;
+    let isHealthyRetryCount = 0;
+    let isHealthyMaxRetryCount = 60; // 1s startup delay and we try max for 60s
+
+    console.log(`[Dapr-JS] Letting Dapr pick-up the server (Maximum 60s wait time)`);
+    while (!isHealthy) {
+      console.log(`[Dapr-JS] - Waiting till Dapr Started (#${isHealthyRetryCount})`);
+      await NodeJSUtils.sleep(this.serverStartupDelay);
+      isHealthy = await this.client.health.isHealthy();
+      isHealthyRetryCount++;
+
+      if (isHealthyRetryCount > isHealthyMaxRetryCount) {
+        throw new Error("DAPR_SIDECAR_COULD_NOT_BE_STARTED");
+      }
+    }
 
     // We are initialized
+    console.log(`[Dapr-JS] Server Started`);
     this.isInitialized = true;
   }
 

src/implementation/Server/HTTPServer/actor.ts

@@ -7,30 +7,31 @@ import IRequest from '../../../types/http/IRequest';
 import IResponse from '../../../types/http/IResponse';
 import { GetRegisteredActorsType } from '../../../types/response/GetRegisteredActors.type';
 import BufferSerializer from '../../../actors/runtime/BufferSerializer';
-import HTTPClient from '../../Client/HTTPClient/HTTPClient';
+import { DaprClient } from '../../..';
 
 // https://docs.dapr.io/reference/api/bindings_api/
 export default class HTTPServerActor implements IServerActor {
   private readonly server: HTTPServer;
-  private readonly client: HTTPClient;
+  private readonly client: DaprClient;
   private readonly serializer: BufferSerializer;
 
-  constructor(server: HTTPServer, client: HTTPClient) {
+  constructor(server: HTTPServer, client: DaprClient) {
     this.client = client;
     this.server = server;
     this.serializer = new BufferSerializer();
   }
 
-  async deactivateActor(actorType: string, actorId: string): Promise<void> {
-    await this.client.execute(`http://localhost:${this.server.serverPort}/actors/${actorType}/${actorId}`, { method: "DELETE" });
-  }
+  // async deactivateActor(actorType: string, actorId: string): Promise<void> {
+  //   await this.client.execute(`http://localhost:${this.server.serverPort}/actors/${actorType}/${actorId}`, { method: "DELETE" });
+  //   await this.client
+  // }
 
   async registerActor<T extends AbstractActor>(cls: Class<T>): Promise<void> {
-    ActorRuntime.getInstance(this.client).registerActor(cls);
+    ActorRuntime.getInstance(this.client.getDaprClient()).registerActor(cls);
   }
 
   async getRegisteredActors(): Promise<string[]> {
-    return await ActorRuntime.getInstance(this.client).getRegisteredActorTypes();
+    return await ActorRuntime.getInstance(this.client.getDaprClient()).getRegisteredActorTypes();
   }
 
   /**
@@ -61,7 +62,7 @@ export default class HTTPServerActor implements IServerActor {
   }
 
   private async handlerConfig(req: IRequest, res: IResponse): Promise<void> {
-    const actorRuntime = ActorRuntime.getInstance(this.client);
+    const actorRuntime = ActorRuntime.getInstance(this.client.getDaprClient());
 
     const result: GetRegisteredActorsType = {
       entities: actorRuntime.getRegisteredActorTypes(),
@@ -73,7 +74,7 @@ export default class HTTPServerActor implements IServerActor {
 
   private async handlerDeactivate(req: IRequest, res: IResponse): Promise<void> {
     const { actorTypeName, actorId } = req.params;
-    const result = await ActorRuntime.getInstance(this.client).deactivate(actorTypeName, actorId);
+    const result = await ActorRuntime.getInstance(this.client.getDaprClient()).deactivate(actorTypeName, actorId);
     return this.handleResult(res, result);
   }
 
@@ -84,7 +85,7 @@ export default class HTTPServerActor implements IServerActor {
     // @todo: reentrancy id? (https://github.com/dapr/python-sdk/blob/master/ext/flask_dapr/flask_dapr/actor.py#L91)
 
     const dataSerialized = this.serializer.serialize(body);
-    const result = await ActorRuntime.getInstance(this.client).invoke(actorTypeName, actorId, methodName, dataSerialized);
+    const result = await ActorRuntime.getInstance(this.client.getDaprClient()).invoke(actorTypeName, actorId, methodName, dataSerialized);
     return this.handleResult(res, result);
   }
 
@@ -93,7 +94,7 @@ export default class HTTPServerActor implements IServerActor {
     const body = req.body;
 
     const dataSerialized = this.serializer.serialize(body);
-    const result = await ActorRuntime.getInstance(this.client).fireTimer(actorTypeName, actorId, timerName, dataSerialized);
+    const result = await ActorRuntime.getInstance(this.client.getDaprClient()).fireTimer(actorTypeName, actorId, timerName, dataSerialized);
     return res.send(result, 200);
   }
 
@@ -102,7 +103,7 @@ export default class HTTPServerActor implements IServerActor {
     const body = req.body;
 
     const dataSerialized = this.serializer.serialize(body);
-    const result = await ActorRuntime.getInstance(this.client).fireReminder(actorTypeName, actorId, reminderName, dataSerialized);
+    const result = await ActorRuntime.getInstance(this.client.getDaprClient()).fireReminder(actorTypeName, actorId, reminderName, dataSerialized);
     return res.send(result, 200);
   }
 

src/implementation/Server/HTTPServer/binding.ts

@@ -1,17 +1,14 @@
 import HTTPServer from './HTTPServer';
 import HttpStatusCode from '../../../enum/HttpStatusCode.enum';
 import IServerBinding from '../../../interfaces/Server/IServerBinding';
-import HTTPClient from '../../Client/HTTPClient/HTTPClient';
 
 // https://docs.dapr.io/reference/api/bindings_api/
 type FunctionDaprInputCallback = (data: any) => Promise<any>;
 
 export default class HTTPServerBinding implements IServerBinding {
   private readonly server: HTTPServer;
-  private readonly client: HTTPClient;
 
-  constructor(server: HTTPServer, client: HTTPClient) {
-    this.client = client;
+  constructor(server: HTTPServer) {
     this.server = server;
   }
 

src/implementation/Server/HTTPServer/invoker.ts

@@ -3,15 +3,12 @@ import { InvokerListenOptionsType } from '../../../types/InvokerListenOptions.ty
 import { HttpMethod } from '../../../enum/HttpMethod.enum';
 import HTTPServer from './HTTPServer';
 import IServerInvoker from '../../../interfaces/Server/IServerInvoker';
-import HTTPClient from '../../Client/HTTPClient/HTTPClient';
 
 // https://docs.dapr.io/reference/api/service_invocation_api/
 export default class HTTPServerInvoker implements IServerInvoker {
   private readonly server: HTTPServer;
-  private readonly client: HTTPClient;
 
-  constructor(server: HTTPServer, client: HTTPClient) {
-    this.client = client;
+  constructor(server: HTTPServer) {
     this.server = server;
   }
 
@@ -28,10 +25,10 @@ export default class HTTPServerInvoker implements IServerInvoker {
           contentType: req.headers['content-type']
         }
       });
-      
+
       // Make sure we close the request after the callback
       // @TODO this should send header and http status code to client (same as grpc)
-      
+
       if (!res.writableEnded) {
         if (invokeResponse) {
           return res.end(JSON.stringify(invokeResponse));