Read input line by line

packages/cli-repl/package.json

@@ -20,6 +20,8 @@
     "start-async": "node --experimental-repl-await bin/mongosh.js start --async",
     "test": "mocha --timeout 15000 --colors -r ts-node/register \"./{src,test}/**/*.spec.ts\"",
     "test-ci": "mocha --timeout 15000 -r ts-node/register \"./{src,test}/**/*.spec.ts\"",
+    "pretest-e2e": "npm run compile-ts",
+    "test-e2e": "mocha --timeout 15000 --colors -r ts-node/register \"./test/e2e.spec.ts\"",
     "lint": "eslint \"**/*.{js,ts,tsx}\"",
     "check": "npm run lint",
     "prepublish": "npm run compile-ts"

packages/cli-repl/src/arg-parser.spec.ts

@@ -1,5 +1,6 @@
 import parse, { getLocale } from './arg-parser';
 import { expect } from 'chai';
+import stripAnsi from 'strip-ansi';
 
 const NODE = 'node';
 const MONGOSH = 'mongosh';
@@ -237,8 +238,14 @@ describe('arg-parser', () => {
             const argv = [ ...baseArgv, uri, '--what' ];
 
             it('raises an error', () => {
-              expect(parse.bind(null, argv)).to.
-                throw('Error parsing command line: unrecognized option: --what');
+              try {
+                parse(argv);
+                throw new Error('should have thrown');
+              } catch (err) {
+                expect(
+                  stripAnsi(err.message)
+                ).to.contain('Error parsing command line: unrecognized option: --what');
+              }
             });
           });
         });
@@ -731,8 +738,14 @@ describe('arg-parser', () => {
             const argv = [ ...baseArgv, uri, '--what' ];
 
             it('raises an error', () => {
-              expect(parse.bind(null, argv)).to.
-                throw('Error parsing command line: unrecognized option: --what');
+              try {
+                parse(argv);
+                throw new Error('should have thrown');
+              } catch (err) {
+                expect(
+                  stripAnsi(err.message)
+                ).to.contain('Error parsing command line: unrecognized option: --what');
+              }
             });
           });
         });

packages/cli-repl/src/cli-repl.ts

@@ -24,6 +24,7 @@ import read from 'read';
 import os from 'os';
 import fs from 'fs';
 import { redactPwd } from '.';
+import { LineByLineInput } from './line-by-line-input';
 
 /**
  * Connecting text key.
@@ -44,13 +45,15 @@ class CliRepl {
   private userId: ObjectId;
   private options: CliOptions;
   private mongoshDir: string;
+  private lineByLineInput: LineByLineInput;
 
   /**
    * Instantiate the new CLI Repl.
    */
   constructor(driverUri: string, driverOptions: NodeOptions, options: CliOptions) {
     this.options = options;
     this.mongoshDir = path.join(os.homedir(), '.mongodb/mongosh/');
+    this.lineByLineInput = new LineByLineInput(process.stdin);
 
     this.createMongoshDir();
 
@@ -105,11 +108,28 @@ class CliRepl {
     const version = this.buildInfo.version;
 
     this.repl = repl.start({
+      input: this.lineByLineInput,
+      output: process.stdout,
       prompt: '> ',
       writer: this.writer,
       completer: completer.bind(null, version),
+      terminal: true
     });
 
+    const originalDisplayPrompt = this.repl.displayPrompt.bind(this.repl);
+
+    this.repl.displayPrompt = (...args: any[]): any => {
+      originalDisplayPrompt(...args);
+      this.lineByLineInput.nextLine();
+    };
+
+    const originalEditorAction = this.repl.commands.editor.action.bind(this.repl);
+
+    this.repl.commands.editor.action = (): any => {
+      this.lineByLineInput.disableBlockOnNewline();
+      return originalEditorAction();
+    };
+
     this.repl.defineCommand('clear', {
       help: '',
       action: () => {
@@ -120,6 +140,8 @@ class CliRepl {
     const originalEval = util.promisify(this.repl.eval);
 
     const customEval = async(input, context, filename, callback): Promise<any> => {
+      this.lineByLineInput.enableBlockOnNewLine();
+
       let result;
 
       try {
@@ -128,7 +150,7 @@ class CliRepl {
         if (isRecoverableError(input)) {
           return callback(new Recoverable(err));
         }
-        result = err;
+        return callback(err);
       }
       callback(null, result);
     };

packages/cli-repl/src/line-by-line-input.spec.ts

@@ -0,0 +1,65 @@
+import { expect } from 'chai';
+import { StringDecoder } from 'string_decoder';
+import { EventEmitter } from 'events';
+import { LineByLineInput } from './line-by-line-input';
+
+describe('LineByLineInput', () => {
+  let stdinMock: NodeJS.ReadStream;
+  let decoder: StringDecoder;
+  let forwardedChunks: string[];
+  let lineByLineInput: LineByLineInput;
+
+  beforeEach(() => {
+    stdinMock = new EventEmitter() as NodeJS.ReadStream;
+    stdinMock.isPaused = (): boolean => false;
+    decoder = new StringDecoder();
+    forwardedChunks = [];
+    lineByLineInput = new LineByLineInput(stdinMock);
+    lineByLineInput.on('data', (chunk) => {
+      const decoded = decoder.write(chunk);
+      if (decoded) {
+        forwardedChunks.push(decoded);
+      }
+    });
+  });
+
+  context('when block on newline is enabled (default)', () => {
+    it('does not forward characters after newline', () => {
+      stdinMock.emit('data', Buffer.from('ab\nc'));
+      expect(forwardedChunks).to.deep.equal(['a', 'b', '\n']);
+    });
+
+    it('forwards CTRL-C anyway and as soon as is received', () => {
+      stdinMock.emit('data', Buffer.from('\n\u0003'));
+      expect(forwardedChunks).to.contain('\u0003');
+    });
+
+    it('forwards CTRL-D anyway and as soon as is received', () => {
+      stdinMock.emit('data', Buffer.from('\n\u0004'));
+      expect(forwardedChunks).to.contain('\u0004');
+    });
+
+    it('unblocks on nextline', () => {
+      stdinMock.emit('data', Buffer.from('ab\nc'));
+      lineByLineInput.nextLine();
+      expect(forwardedChunks).to.deep.equal(['a', 'b', '\n', 'c']);
+    });
+  });
+
+  context('when block on newline is disabled', () => {
+    it('does forwards all the characters', () => {
+      lineByLineInput.disableBlockOnNewline();
+      stdinMock.emit('data', Buffer.from('ab\nc'));
+      expect(forwardedChunks).to.deep.equal(['ab\nc']);
+    });
+  });
+
+  context('when block on newline is disabled and re-enabled', () => {
+    it('does forwards all the characters', () => {
+      lineByLineInput.disableBlockOnNewline();
+      lineByLineInput.enableBlockOnNewLine();
+      stdinMock.emit('data', Buffer.from('ab\nc'));
+      expect(forwardedChunks).to.deep.equal(['a', 'b', '\n']);
+    });
+  });
+});

packages/cli-repl/src/line-by-line-input.ts

@@ -0,0 +1,172 @@
+import { EventEmitter } from 'events';
+import { StringDecoder } from 'string_decoder';
+
+const LINE_ENDING_RE = /\r?\n|\r(?!\n)/;
+const CTRL_C = '\u0003';
+const CTRL_D = '\u0004';
+
+/**
+ * A proxy for `tty.ReadStream` that allows to read
+ * the stream line by line.
+ *
+ * Each time a newline is encountered the stream wont emit further data
+ * untill `.nextLine()` is called.
+ *
+ * NOTE: the control sequences Ctrl+C and Ctrl+D are not buffered and instead
+ * are forwarded regardless.
+ *
+ * Is possible to disable the "line splitting" by calling `.disableBlockOnNewline()` and
+ * re-enable it by calling `.enableBlockOnNewLine()`.
+ *
+ * If the line splitting is disabled the stream will behave like
+ * the proxied `tty.ReadStream`, forwarding all the characters.
+ */
+export class LineByLineInput {
+  private _emitter: EventEmitter;
+  private _originalInput: NodeJS.ReadStream;
+  private _forwarding: boolean;
+  private _blockOnNewLineEnabled: boolean;
+  private _charQueue: string[];
+  private _decoder: StringDecoder;
+
+  constructor(readable: NodeJS.ReadStream) {
+    this._emitter = new EventEmitter();
+    this._originalInput = readable;
+    this._forwarding = true;
+    this._blockOnNewLineEnabled = true;
+    this._charQueue = [];
+    this._decoder = new StringDecoder('utf-8');
+
+    readable.on('data', this._onData);
+
+    const proxy = new Proxy(readable, {
+      get: (target: NodeJS.ReadStream, property: string): any => {
+        if (typeof property === 'string' &&
+          !property.startsWith('_') &&
+          typeof this[property] === 'function'
+        ) {
+          return this[property].bind(this);
+        }
+
+        return target[property];
+      }
+    });
+
+    return (proxy as unknown) as LineByLineInput;
+  }
+
+  on(event: string, handler: (...args: any[]) => void): void {
+    if (event === 'data') {
+      this._emitter.on('data', handler);
+      // we may have buffered data for the first listener
+      this._flush();
+      return;
+    }
+
+    this._originalInput.on(event, handler);
+    return;
+  }
+
+  nextLine(): void {
+    this._resumeForwarding();
+    this._flush();
+  }
+
+  enableBlockOnNewLine(): void {
+    this._blockOnNewLineEnabled = true;
+  }
+
+  disableBlockOnNewline(): void {
+    this._blockOnNewLineEnabled = false;
+    this._flush();
+  }
+
+  private _onData = (chunk: Buffer): void => {
+    if (this._blockOnNewLineEnabled) {
+      return this._forwardAndBlockOnNewline(chunk);
+    }
+
+    return this._forwardWithoutBlocking(chunk);
+  };
+
+  private _forwardAndBlockOnNewline(chunk: Buffer): void {
+    const chars = this._decoder.write(chunk);
+    for (const char of chars) {
+      if (this._isCtrlC(char) || this._isCtrlD(char)) {
+        this._emitChar(char);
+      } else {
+        this._charQueue.push(char);
+      }
+    }
+    this._flush();
+  }
+
+  private _forwardWithoutBlocking(chunk: Buffer): void {
+    // keeps decoding state consistent
+    this._decoder.write(chunk);
+    this._emitChunk(chunk);
+  }
+
+  private _pauseForwarding(): void {
+    this._forwarding = false;
+  }
+
+  private _resumeForwarding(): void {
+    this._forwarding = true;
+  }
+
+  private _shouldForward(): boolean {
+    // If we are not blocking on new lines
+    // we just forward everything as is,
+    // otherwise we forward only if the forwarding
+    // is not paused.
+
+    return !this._blockOnNewLineEnabled || this._forwarding;
+  }
+
+  private _emitChar(char): void {
+    this._emitChunk(Buffer.from(char, 'utf8'));
+  }
+
+  private _emitChunk(chunk: Buffer): void {
+    this._emitter.emit('data', chunk);
+  }
+
+  private _flush(): void {
+    // there is nobody to flush for
+    if (this._emitter.listenerCount('data') === 0) {
+      return;
+    }
+
+    while (
+      this._charQueue.length &&
+      this._shouldForward() &&
+
+      // We don't forward residual characters we could
+      // have in the buffer if in the meanwhile something
+      // downstream explicitly called pause(), as that may cause
+      // unexpected behaviors.
+      !this._originalInput.isPaused()
+    ) {
+      const char = this._charQueue.shift();
+
+      if (this._isLineEnding(char)) {
+        this._pauseForwarding();
+      }
+
+      this._emitChar(char);
+    }
+  }
+
+  private _isLineEnding(char: string): boolean {
+    return LINE_ENDING_RE.test(char);
+  }
+
+  private _isCtrlD(char: string): boolean {
+    return char === CTRL_D;
+  }
+
+  private _isCtrlC(char: string): boolean {
+    return char === CTRL_C;
+  }
+}