Add if PubSub was successful or not, implement shutdown to close handle and work on load tests

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

Author: XavierGeerinck

package.json

@@ -5,6 +5,8 @@
     "types": "http/index.d.ts",
     "scripts": {
         "test": "npm run test:unit:all && npm run test:e2e:all",
+        "test:load": "jest --runInBand --detectOpenHandles --forceExit",
+        "test:load:http": "TEST_SECRET_1=secret_val_1 TEST_SECRET_2=secret_val_2 dapr run --app-id test-suite --app-protocol http --app-port 50001 --dapr-http-port 50000 --components-path ./test/components npm run test:load 'test/load'",
         "test:e2e": "jest --runInBand --detectOpenHandles --forceExit",
         "test:e2e:all": "npm run test:e2e:grpc:main && npm run test:e2e:http:main && npm run test:e2e:http:actors",
         "test:e2e:grpc:main": "TEST_SECRET_1=secret_val_1 TEST_SECRET_2=secret_val_2 dapr run --app-id test-suite --app-protocol grpc --app-port 50001 --dapr-grpc-port 50000 --components-path ./test/components npm run test:e2e 'test/e2e/main.grpc.test.ts'",

src/implementation/Client/GRPCClient/pubsub.ts

@@ -11,7 +11,7 @@ export default class GRPCClientPubSub implements IClientPubSub {
   }
 
   // @todo: should return a specific typed Promise<TypePubSubPublishResponse> instead of Promise<any>
-  async publish(pubSubName: string, topic: string, data: object = {}): Promise<any> {
+  async publish(pubSubName: string, topic: string, data: object = {}): Promise<Boolean> {
     const msgService = new PublishEventRequest();
     msgService.setPubsubName(pubSubName);
     msgService.setTopic(topic);
@@ -21,11 +21,12 @@ export default class GRPCClientPubSub implements IClientPubSub {
       const client = this.client.getClient();
       client.publishEvent(msgService, (err, res) => {
         if (err) {
-          return reject(err);
+          console.error(err);
+          return reject(false);
         }
 
         // https://docs.dapr.io/reference/api/pubsub_api/#expected-http-response
-        return resolve({});
+        return resolve(true);
       });
     });
   }

src/implementation/Client/HTTPClient/pubsub.ts

@@ -9,13 +9,20 @@ export default class HTTPClientPubSub implements IClientPubSub {
     this.client = client;
   }
 
-  async publish(pubSubName: string, topic: string, data: object = {}): Promise<void> {
-    await this.client.execute(`/publish/${pubSubName}/${topic}`, {
-      method: 'POST',
-      headers: {
-        'Content-Type': 'application/json',
-      },
-      body: JSON.stringify(data),
-    });
+  async publish(pubSubName: string, topic: string, data: object = {}): Promise<Boolean> {
+    try {
+      await this.client.execute(`/publish/${pubSubName}/${topic}`, {
+        method: 'POST',
+        headers: {
+          'Content-Type': 'application/json',
+        },
+        body: JSON.stringify(data),
+      });
+    } catch (e) {
+      console.error(e);
+      return false;
+    }
+
+    return true;
   }
 }

src/implementation/Server/DaprServer.ts

@@ -93,6 +93,10 @@ export default class DaprServer {
     await this.daprServer.startServer(this.serverHost, this.serverPort.toString());
   }
 
+  async stopServer(): Promise<void> {
+    await this.daprServer.stopServer();
+  }
+
   getDaprClient(): IServer {
     return this.daprServer;
   }

src/implementation/Server/GRPCServer/GRPCServer.ts

@@ -99,6 +99,20 @@ export default class GRPCServer implements IServer {
     this.isInitialized = true;
   }
 
+  async stopServer(): Promise<void> {
+    return new Promise((resolve, reject) => {
+
+      this.server.tryShutdown((err) => {
+        if (err) {
+          return reject(err);
+        }
+
+        this.isInitialized = false;
+        return resolve();
+      });
+    });
+  }
+
   private async initializeBind(): Promise<void> {
     console.log(`[Dapr-JS][gRPC] Starting to listen on ${this.serverHost}:${this.serverPort}`);
     return new Promise((resolve, reject) => {

src/implementation/Server/HTTPServer/HTTPServer.ts

@@ -118,4 +118,9 @@ export default class HTTPServer implements IServer {
     // We are initialized
     this.isInitialized = true;
   }
+
+  async stopServer(): Promise<void> {
+    await this.server.close();
+    this.isInitialized = false;
+  }
 }

src/interfaces/Client/IClientPubSub.ts

@@ -1,3 +1,3 @@
 export default interface IClientPubSub {
-    publish(pubSubName: string, topic: string, data?: object): Promise<void>
+  publish(pubSubName: string, topic: string, data?: object): Promise<Boolean>
 }

src/interfaces/Server/IServer.ts

@@ -1,7 +1,8 @@
 export default interface IServer {
-    getServerAddress(): string;
-    getServer(): any; // this is dependent on the implementation
-    getServerImpl(): any; // this is dependent on the implementation
-    close(): Promise<void>;
-    startServer(host: string, port: string): Promise<void>;
+  getServerAddress(): string;
+  getServer(): any; // this is dependent on the implementation
+  getServerImpl(): any; // this is dependent on the implementation
+  close(): Promise<void>;
+  startServer(host: string, port: string): Promise<void>;
+  stopServer(): Promise<void>;
 }

test/components/pubsub-mqtt.yaml

@@ -0,0 +1,18 @@
+# https://docs.dapr.io/reference/components-reference/supported-bindings/rabbitmq/
+apiVersion: dapr.io/v1alpha1
+kind: Component
+metadata:
+  name: pubsub-mqtt
+  namespace: default
+spec:
+  type: pubsub.mqtt
+  version: v1
+  metadata:
+  - name: url
+    value: "tcp://admin:public@localhost:1883"
+  - name: qos
+    value: 1
+  - name: cleanSession
+    value: "true"
+  - name: backOffMaxRetries
+    value: "0"

test/e2e/main.grpc.test.ts

@@ -55,6 +55,11 @@ describe('grpc/main', () => {
       // @ts-ignore
       expect(mockPubSubSubscribe.mock.calls[0][0]['hello']).toEqual('world');
     });
+
+    it('should receive if it was successful or not', async () => {
+      const res = await client.pubsub.publish('pubsub-redis', 'test-topic', { hello: 'world' });
+      expect(res).toEqual(true);
+    });
   });
 
   describe('invoker', () => {