server connections handling

packages/api/src/controllers/serverConnections.js

@@ -1,24 +1,31 @@
 const connections = require('./connections');
 const socket = require('../utility/socket');
 const { fork } = require('child_process');
+const _ = require('lodash');
 
 module.exports = {
   opened: [],
+  closed: [],
 
   handle_databases(conid, { databases }) {
-    const existing = this.opened.find(x => x.conid == conid);
+    const existing = this.opened.find((x) => x.conid == conid);
     if (!existing) return;
     existing.databases = databases;
     socket.emitChanged(`database-list-changed-${conid}`);
   },
+  handle_status(conid, { status }) {
+    const existing = this.opened.find((x) => x.conid == conid);
+    if (!existing) return;
+    existing.status = status;
+    socket.emitChanged(`server-status-changed`);
+  },
   handle_error(conid, { error }) {
     console.log(`Error in server connection ${conid}: ${error}`);
   },
-  handle_ping() {
-  },
+  handle_ping() {},
 
   async ensureOpened(conid) {
-    const existing = this.opened.find(x => x.conid == conid);
+    const existing = this.opened.find((x) => x.conid == conid);
     if (existing) return existing;
     const connection = await connections.get({ conid });
     const subprocess = fork(process.argv[1], ['serverConnectionProcess']);
@@ -27,19 +34,75 @@ module.exports = {
       subprocess,
       databases: [],
       connection,
+      status: {
+        name: 'pending',
+      },
+      disconnected: false,
     };
     this.opened.push(newOpened);
+    this.closed = this.closed.filter((x) => x != conid);
+    socket.emitChanged(`server-status-changed`);
     // @ts-ignore
     subprocess.on('message', ({ msgtype, ...message }) => {
+      if (newOpened.disconnected) return;
       this[`handle_${msgtype}`](conid, message);
     });
+    subprocess.on('exit', () => {
+      if (newOpened.disconnected) return;
+      this.opened = this.opened.filter((x) => x.conid != conid);
+      this.closed.push(conid);
+      socket.emitChanged(`server-status-changed`);
+    });
     subprocess.send({ msgtype: 'connect', ...connection });
     return newOpened;
   },
 
+  close(conid) {
+    const existing = this.opened.find((x) => x.conid == conid);
+    if (existing) {
+      existing.disconnected = true;
+      existing.subprocess.kill();
+      this.opened = this.opened.filter((x) => x.conid != conid);
+      this.closed.push(conid);
+    }
+  },
+
   listDatabases_meta: 'get',
   async listDatabases({ conid }) {
     const opened = await this.ensureOpened(conid);
     return opened.databases;
   },
+
+  serverStatus_meta: 'get',
+  async serverStatus() {
+    return {
+      ...this.closed.reduce(
+        (res, conid) => ({
+          ...res,
+          [conid]: {
+            name: 'error',
+          },
+        }),
+        {}
+      ),
+      ..._.mapValues(_.keyBy(this.opened, 'conid'), 'status'),
+    };
+  },
+
+  ping_meta: 'post',
+  async ping({ connections }) {
+    for (const conid of connections) {
+      const opened = await this.ensureOpened(conid);
+      opened.subprocess.send({ msgtype: 'ping' });
+    }
+    return { status: 'ok' };
+  },
+
+  refresh_meta: 'post',
+  async refresh({ conid }) {
+    this.close(conid);
+
+    await this.ensureOpened(conid);
+    return { status: 'ok' };
+  },
 };

packages/api/src/proc/serverConnectionProcess.js

@@ -1,27 +1,67 @@
 const engines = require('@dbgate/engines');
 const driverConnect = require('../utility/driverConnect');
 const childProcessChecker = require('../utility/childProcessChecker');
+const stableStringify = require('json-stable-stringify');
 
 let systemConnection;
 let storedConnection;
+let lastDatabases = null;
+let lastStatus = null;
+let lastPing = null;
 
-async function handleRefreshDatabases() {
+async function handleRefresh() {
   const driver = engines(storedConnection);
-  const databases = await driver.listDatabases(systemConnection);
-  process.send({ msgtype: 'databases', databases });
+  try {
+    const databases = await driver.listDatabases(systemConnection);
+    setStatusName('ok');
+    const databasesString = stableStringify(databases);
+    if (lastDatabases != databasesString) {
+      process.send({ msgtype: 'databases', databases });
+      lastDatabases = databasesString;
+    }
+  } catch (err) {
+    setStatusName('error');
+    console.error(err);
+    process.exit(1);
+  }
+}
+
+function setStatus(status) {
+  const statusString = stableStringify(status);
+  if (lastStatus != statusString) {
+    process.send({ msgtype: 'status', status });
+    lastStatus = statusString;
+  }
+}
+
+function setStatusName(name) {
+  setStatus({ name });
 }
 
 async function handleConnect(connection) {
   storedConnection = connection;
+  setStatusName('pending');
+  lastPing = new Date().getTime();
 
   const driver = engines(storedConnection);
-  systemConnection = await driverConnect(driver, storedConnection);
-  handleRefreshDatabases();
-  setInterval(handleRefreshDatabases, 30 * 1000);
+  try {
+    systemConnection = await driverConnect(driver, storedConnection);
+    handleRefresh();
+    setInterval(handleRefresh, 30 * 1000);
+  } catch (err) {
+    setStatusName('error');
+    console.error(err);
+    process.exit(1);
+  }
+}
+
+function handlePing() {
+  lastPing = new Date().getTime();
 }
 
 const messageHandlers = {
   connect: handleConnect,
+  ping: handlePing,
 };
 
 async function handleMessage({ msgtype, ...other }) {
@@ -31,6 +71,14 @@ async function handleMessage({ msgtype, ...other }) {
 
 function start() {
   childProcessChecker();
+
+  setInterval(() => {
+    const time = new Date().getTime();
+    if (time - lastPing > 60 * 1000) {
+      process.exit(0);
+    }
+  }, 60 * 1000);
+
   process.on('message', async (message) => {
     try {
       await handleMessage(message);