🩹 Add stop to platform and wait for it in CLI (#438)

tdrive/backend/node/src/cli/cmds/search_cmds/index_all.ts

@@ -131,7 +131,9 @@ const command: yargs.CommandModule<unknown, unknown> = {
       repository,
       spinner,
     });
-
+    spinner.start("Shutting down platform...");
+    await platform.stop();
+    spinner.succeed("Platform shutdown");
     spinner.stop();
   },
 };

tdrive/backend/node/src/core/platform/services/database/index.ts

@@ -36,6 +36,14 @@ export default class Database extends TdriveService<DatabaseServiceAPI> {
     return this;
   }
 
+  public async doStop(): Promise<this> {
+    if (this.service) {
+      await this.service.disconnect();
+      this.service = null;
+    }
+    return this;
+  }
+
   api(): DatabaseServiceAPI {
     return this.service;
   }

tdrive/backend/node/src/core/platform/services/database/services/index.ts

@@ -32,6 +32,13 @@ export default class DatabaseService implements DatabaseServiceAPI {
     return this.connector;
   }
 
+  async disconnect() {
+    if (this.connector) {
+      await this.connector.disconnect();
+      this.connector = null;
+    }
+  }
+
   getManager(): Manager<unknown> {
     return new Manager<unknown>(this.connector);
   }

tdrive/backend/node/src/core/platform/services/database/services/orm/connectors/abstract-connector.ts

@@ -9,6 +9,7 @@ export abstract class AbstractConnector<T extends ConnectionOptions> implements
   constructor(protected type: DatabaseType, protected options: T, protected secret: string) {}
 
   abstract connect(): Promise<this>;
+  abstract disconnect(): Promise<this>;
 
   abstract drop(): Promise<this>;
 

tdrive/backend/node/src/core/platform/services/database/services/orm/connectors/cassandra/cassandra.ts

@@ -178,6 +178,12 @@ export class CassandraConnector extends AbstractConnector<CassandraConnectionOpt
     return this;
   }
 
+  async disconnect(): Promise<this> {
+    if (this.client) await this.client.shutdown();
+    this.client = null;
+    return this;
+  }
+
   async getTableDefinition(name: string): Promise<string[]> {
     let result: string[];
 

tdrive/backend/node/src/core/platform/services/database/services/orm/connectors/index.ts

@@ -22,6 +22,11 @@ export interface Connector extends Initializable {
    */
   connect(): Promise<this>;
 
+  /**
+   * Disconnect from the database
+   */
+  disconnect(): Promise<this>;
+
   /**
    * Get the type of connector
    */

tdrive/backend/node/src/core/platform/services/database/services/orm/connectors/mongodb/mongodb.ts

@@ -36,6 +36,12 @@ export class MongoConnector extends AbstractConnector<MongoConnectionOptions> {
     return this;
   }
 
+  async disconnect(): Promise<this> {
+    if (this.client) await this.client.close();
+    this.client = null;
+    return this;
+  }
+
   getClient(): mongo.MongoClient {
     return this.client;
   }

tdrive/backend/node/src/core/platform/services/database/services/orm/connectors/postgres/postgres.ts

@@ -50,6 +50,12 @@ export class PostgresConnector extends AbstractConnector<PostgresConnectionOptio
     return this;
   }
 
+  async disconnect(): Promise<this> {
+    if (this.client) await this.client.end();
+    this.client = null;
+    return this;
+  }
+
   async init(): Promise<this> {
     if (!this.client) {
       await this.connect();

tdrive/backend/node/src/core/platform/services/search/adapters/abstract.ts

@@ -30,6 +30,9 @@ export class SearchAdapter implements SearchAdapterInterface {
   connect(): Promise<void> {
     throw new Error("Method not implemented.");
   }
+  disconnect(): Promise<void> {
+    throw new Error("Method not implemented.");
+  }
   upsert(entities: any[]): Promise<void> {
     throw new Error("Method not implemented.");
   }

tdrive/backend/node/src/core/platform/services/search/adapters/elasticsearch/elastic-open-search-adapter.ts

@@ -29,6 +29,7 @@ export default class ESAndOpenSearch extends SearchAdapter implements SearchAdap
   private bulkReaders = 0;
   private buffer: Operation[] = [];
   private client: OpenClient | ESClient;
+  private pendingTimeout: NodeJS.Timeout = null;
 
   constructor(
     readonly database: DatabaseServiceAPI,
@@ -75,6 +76,18 @@ export default class ESAndOpenSearch extends SearchAdapter implements SearchAdap
     this.startBulkReader();
   }
 
+  public async disconnect() {
+    if (this.client) {
+      const client = this.client;
+      this.client = null;
+      if (this.pendingTimeout) {
+        clearTimeout(this.pendingTimeout);
+        this.pendingTimeout = null;
+      }
+      await client.close();
+    }
+  }
+
   private async createIndex(
     entity: EntityDefinition,
     _columns: { [name: string]: ColumnDefinition },
@@ -214,10 +227,17 @@ export default class ESAndOpenSearch extends SearchAdapter implements SearchAdap
 
     let buffer;
     do {
-      await new Promise(r =>
+      await new Promise<void>(
-        setTimeout(r, parseInt(`${this.configuration.flushInterval}`) || 3000),
+        r =>
+          (this.pendingTimeout = setTimeout(() => {
+            this.pendingTimeout = null;
+            r();
+          }, parseInt(`${this.configuration.flushInterval}`) || 3000)),
       );
       buffer = this.buffer;
+      if (!this.client)
+        // disconnect was called; break this tail loop
+        return;
     } while (buffer.length === 0);
     this.buffer = [];
 

tdrive/backend/node/src/core/platform/services/search/adapters/mongosearch/index.ts

@@ -45,6 +45,14 @@ export default class MongoSearch extends SearchAdapter implements SearchAdapterI
     this.mongodb = await service.getDatabase();
   }
 
+  public async disconnect() {
+    if (this.mongodb) {
+      const service = this.database.getConnector() as MongoConnector;
+      await service.disconnect();
+      this.mongodb = null;
+    }
+  }
+
   private async createIndex(
     entityDefinition: EntityDefinition,
     columns: { [name: string]: ColumnDefinition },

tdrive/backend/node/src/core/platform/services/search/api.ts

@@ -46,6 +46,7 @@ export type IndexedEntity = {
 
 export interface SearchAdapterInterface {
   connect(): Promise<void>;
+  disconnect(): Promise<void>;
   upsert(entities: any[]): Promise<void>;
   remove(entities: any[]): Promise<void>;
 

tdrive/backend/node/src/core/platform/services/search/index.ts

@@ -73,6 +73,14 @@ export default class Search extends TdriveService<SearchServiceAPI> {
     return this;
   }
 
+  public async doStop(): Promise<this> {
+    if (this.service) {
+      await this.service.disconnect();
+      this.service = null;
+    }
+    return this;
+  }
+
   public async upsert(entities: never[]) {
     return this.service.upsert(entities);
   }