ecal/ecal-support/src/ecalDebugClient.ts

/**
 * Debug client implementation for the ECAL debugger.
 */

import * as net from 'net';
import { EventEmitter } from 'events';
import PromiseSocket from 'promise-socket';
import { LogOutputStream, DebugStatus, ThreadInspection } from './types';

interface BacklogCommand {
  cmd:string,
  args? :string[],
}

/**
 * Debug client for ECAL debug server.
 */
export class ECALDebugClient extends EventEmitter {
  private socket : PromiseSocket<net.Socket>;
  private socketLock : any;
  private connected : boolean = false
  private backlog : BacklogCommand[] = []
  private threadInspection : Record<string, ThreadInspection> = {}

  /**
     * Create a new debug client.
     */
  public constructor (
        protected out: LogOutputStream
  ) {
    super()
    this.socket = new PromiseSocket(new net.Socket());

      const AsyncLock = require('async-lock');
      this.socketLock = new AsyncLock();
  }

  public async conect (host: string, port: number) {
      try {
          this.out.log(`Connecting to: ${host}:${port}`);
          await this.socket.connect({ port, host });
          // this.socket.setTimeout(2000);
          this.connected = true;
          this.pollEvents(); // Start emitting events
      } catch (e) {
          this.out.error(`Could not connect to debug server: ${e}`);
      }
  }

  public async status () : Promise<DebugStatus | null> {
      try {
          return await this.sendCommand('status') as DebugStatus;
      } catch (e) {
          this.out.error(`Could not query for status: ${e}`);
          return null;
      }
  }

  public async inspect (tid:string) : Promise<ThreadInspection | null> {
      try {
          return await this.sendCommand('inspect', [tid]) as ThreadInspection;
      } catch (e) {
          this.out.error(`Could not inspect thread ${tid}: ${e}`);
          return null;
      }
  }

  public async setBreakpoint (breakpoint: string) {
      try {
          await this.sendCommand(`break ${breakpoint}`) as DebugStatus;
      } catch (e) {
          this.out.error(`Could not set breakpoint ${breakpoint}: ${e}`);
      }
  }

  public async clearBreakpoints (source: string) {
      try {
          await this.sendCommand('rmbreak', [source]) as DebugStatus;
      } catch (e) {
          this.out.error(`Could not remove breakpoints for ${source}: ${e}`);
      }
  }

  public async shutdown () {
      this.connected = false;
      await this.socket.destroy();
  }

  /**
   * PollEvents is the polling loop for debug events.
   */
  private async pollEvents () {
      let nextLoop = 1000;
      try {
          const status = await this.status();

          for (const [tid, thread] of Object.entries(status?.threads || [])) {
              if (thread.threadRunning === false && !this.threadInspection[tid]) {

                console.log("#### Thread was stopped!!")

                // A thread was stopped inspect it

                let inspection : ThreadInspection = {
                    callstack: [],
                    threadRunning: false
                }

                try {
                    inspection = await this.sendCommand('describe', [tid]) as ThreadInspection;
                } catch (e) {
                    this.out.error(`Could not get description for ${tid}: ${e}`);
                }

                this.threadInspection[tid] = inspection;

                console.log("#### Description result:", inspection)

                this.emit('pauseOnBreakpoint', tid);
            }
          }

      } catch (e) {
          this.out.error(`Error during event loop: ${e}`);
          nextLoop = 5000;
      }

      if (this.connected) {
          setTimeout(this.pollEvents.bind(this), nextLoop);
      } else {
          this.out.log('Stop emitting events' + nextLoop);
      }
  }

  public async sendCommand (cmd:string, args? :string[]): Promise<any> {
      // Create or process the backlog depending on the connection status

      if (!this.connected) {
          this.backlog.push({
              cmd,
              args
          });
          return null;
      } else if (this.backlog.length > 0) {
          const backlog = this.backlog;
          this.backlog = [];
          for (const item of backlog) {
              await this.sendCommand(item.cmd, item.args);
          }
      }

      return await this.sendCommandString(`##${cmd} ${args ? args.join(' ') : ''}\r\n`);
  }

  public async sendCommandString (cmdString:string): Promise<any> {
      // Socket needs to be locked. Reading and writing to the socket is seen
      // by the interpreter as async (i/o bound) code. Separate calls to
      // sendCommand will be executed in different event loops. Without the lock
      // the different sendCommand calls would mix their responses.

      return await this.socketLock.acquire('socket', async () => {
          await this.socket.write(cmdString, 'utf8');

          let text = '';
          while (!text.endsWith('\n\n')) {
              text += await this.socket.read(1);
          }

          let res : any = {};
          try {
              res = JSON.parse(text);
          } catch (e) {
              throw (new Error(`Could not parse response: ${text} - error:${e}`));
          }
          if (res?.DebuggerError) {
              throw (new Error(`Unexpected internal error for command "${cmdString}": ${res.DebuggerError}`));
          }
          if (res?.EncodedOutput !== undefined) {
              res = Buffer.from(res.EncodedOutput, 'base64').toString('utf8');
          }
          return res;
      });
  }
}