fix(extension-driver-ksqldb): review suggestions, add comments, timeout option

Author: andreashimin

packages/doc/docs/connect/ksqldb.mdx

@@ -28,6 +28,7 @@
        host: www.example.com:8088
        username: <username>
        password: <password>
+       timeout: 25000
      allow: '*'
    ```
 
@@ -37,4 +38,5 @@
 | ------- | -------- | --------- | ----------- |
 | host     | N        |  http://localhost:8088  | ksqlDB instance.  |
 | username | N        |                         | The name of the user on whose behalf requests are made.  |
-| password | N        |                         | The user's password.  |
+| password | N        |                         | The user's password.  |
+| timeout  | N        |  25000                  | Request timeout in milliseconds.  |

packages/extension-driver-ksqldb/README.md

@@ -28,7 +28,9 @@ This is the KSqlDb driver for VulcanSQL, provided by [Canner](https://cannerdata
        # Optional: The name of the user on whose behalf requests are made.
        username: '<username>',
        # The user's password
-       password: '<password>'
+       password: '<password>',
+       # Optional: Request timeout in milliseconds. Default value: 25000
+       timeout: 25000
    ```
 
 ## Testing

packages/extension-driver-ksqldb/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@vulcan-sql/extension-driver-ksqldb",
   "description": "KSqlDb driver for VulcanSQL",
-  "version": "0.7.0",
+  "version": "0.7.1",
   "type": "commonjs",
   "publishConfig": {
     "access": "public"
@@ -23,6 +23,6 @@
   },
   "license": "MIT",
   "peerDependencies": {
-    "@vulcan-sql/core": "~0.7.0-0"
+    "@vulcan-sql/core": "~0.7.1-0"
   }
 }

packages/extension-driver-ksqldb/project.json

@@ -32,6 +32,19 @@
           "target": "install-dependencies"
         }
       ]
+    },
+    "publish": {
+      "executor": "@nrwl/workspace:run-commands",
+      "options": {
+        "command": "node ../../../tools/scripts/publish.mjs {args.tag} {args.version}",
+        "cwd": "dist/packages/extension-driver-ksqldb"
+      },
+      "dependsOn": [
+        {
+          "projects": "self",
+          "target": "build"
+        }
+      ]
     }
   },
   "tags": []

packages/extension-driver-ksqldb/src/lib/restfulClient.ts

@@ -54,6 +54,12 @@ export interface RestfulClientOptions {
   host?: string;
   username?: string;
   password?: string;
+  timeout?: number;
+}
+
+const DEFAULT_OPTIONS: Required<Pick<RestfulClientOptions, 'host'|'timeout'>> = {
+  host: 'http://localhost:8088',
+  timeout: 25000,
 }
 
 export class RestfulClient {
@@ -67,10 +73,24 @@ export class RestfulClient {
     this.connect();
   }
 
+  /**
+   * The connect method will create a promise "startSession" method, not really to connect http2 immediately.
+   * To let users establish a "startSession" promise request only when they need to query or exec a statement.
+   */
   public connect() {
     this.startSession = () =>
       new Promise((resolve, reject) => {
-        this.client = http2.connect(this.options.host || 'http://localhost:8088');
+        this.client = http2.connect(this.options.host || DEFAULT_OPTIONS.host, {
+          timeout: this.options.timeout || DEFAULT_OPTIONS.timeout,
+        });
+
+        this.client.setTimeout(this.options.timeout || DEFAULT_OPTIONS.timeout, () => {
+          if (this.connected === false) {
+              const timeoutError = new Error("Connection timeout.");
+              reject(timeoutError);
+              this.client?.destroy(timeoutError);
+          }
+      });
 
         this.client.on('connect', () => {
           this.connected = true;
@@ -83,7 +103,7 @@ export class RestfulClient {
       });
   }
 
-  public close(): Promise<void> {
+  public closeSession(): Promise<void> {
     return new Promise((resolve, reject) => {
       if (this.client) {
         !this.client.destroyed && this.client.destroy();
@@ -116,6 +136,11 @@ export class RestfulClient {
     return isRunning;
   }
 
+  /**
+   * According to ksqldb restful API: https://docs.ksqldb.io/en/latest/developer-guide/ksqldb-rest-api/query-endpoint
+   * To run a SELECT statement and stream back the results.
+   * SELECT statement: https://docs.ksqldb.io/en/latest/developer-guide/ksqldb-reference/select-pull-query
+   */
   public async query({
     query,
     query_params = {},
@@ -131,6 +156,11 @@ export class RestfulClient {
     return res;
   }
 
+  /**
+   * According to ksqldb restful API: https://docs.ksqldb.io/en/latest/developer-guide/ksqldb-rest-api/ksql-endpoint
+   * All statements, except those starting with SELECT and PRINT, can be run on this exec method. 
+   * To run SELECT and PRINT statements use the "query" method instead.
+   */
   public async exec({
     query,
     query_params = {},
@@ -153,6 +183,8 @@ export class RestfulClient {
     const ksql = query.replace(/\$(\d+)/g, (_, index) => {
       const valueIndex = parseInt(index) - 1;
       const paramValue = values[valueIndex];
+      // Because the ksqldb queries are expressed using a strict subset of ANSI SQL.
+      // It didn't support the string auto conversion, so we need to add the single quote for string value.
       return typeof paramValue === 'string' ? `'${paramValue}'` : paramValue;
     });
 
@@ -203,12 +235,12 @@ export class RestfulClient {
         } else {
           reject(responseData);
         }
-        this.close();
+        this.closeSession();
       });
 
       req.on('error', (error) => {
         reject(error);
-        this.close();
+        this.closeSession();
       });
 
       buffer && req.write(buffer);