flow control handling

README.md

@@ -100,6 +100,29 @@ All processes launched from node-pty will launch at the same permission level of
 
 Note that node-pty is not thread safe so running it across multiple worker threads in node.js could cause issues.
 
+## Flow Control
+
+Automatic flow control can be enabled by either providing `handleFlowControl = true` in the constructor options or setting it later on:
+
+```js
+const PAUSE = '\x13';   // XOFF
+const RESUME = '\x11';  // XON
+
+const ptyProcess = pty.spawn(shell, [], {handleFlowControl: true});
+
+// flow control in action
+ptyProcess.write(PAUSE);  // pty will block and pause the slave program
+...
+ptyProcess.write(RESUME); // pty will enter flow mode and resume the slave program
+
+// temporarily disable/re-enable flow control
+ptyProcess.handleFlowControl = false;
+...
+ptyProcess.handleFlowControl = true;
+```
+
+By default `PAUSE` and `RESUME` are XON/XOFF control codes (as shown above). To avoid conflicts in environments that use these control codes for different purposes the messages can be customized as `flowControlPause: string` and `flowControlResume: string` in the constructor options. `PAUSE` and `RESUME` are not passed to the underlying pseudoterminal if flow control is enabled.
+
 ## Troubleshooting
 
 ### Powershell gives error 8009001d

src/interfaces.ts

@@ -117,6 +117,9 @@ export interface IPtyForkOptions {
   gid?: number;
   encoding?: string;
   experimentalUseConpty?: boolean | undefined;
+  handleFlowControl?: boolean;
+  flowControlPause?: string;
+  flowControlResume?: string;
 }
 
 export interface IPtyOpenOptions {

src/terminal.test.ts

@@ -6,6 +6,10 @@
 import * as assert from 'assert';
 import { WindowsTerminal } from './windowsTerminal';
 import { UnixTerminal } from './unixTerminal';
+import pollUntil = require('pollUntil');
+
+const terminalConstructor = (process.platform === 'win32') ? WindowsTerminal : UnixTerminal;
+const SHELL = (process.platform === 'win32') ? 'cmd.exe' : '/bin/bash';
 
 let terminalCtor: WindowsTerminal | UnixTerminal;
 if (process.platform === 'win32') {
@@ -14,6 +18,7 @@ if (process.platform === 'win32') {
   terminalCtor = require('./unixTerminal');
 }
 
+
 describe('Terminal', () => {
   describe('constructor', () => {
     it('should do basic type checks', () => {
@@ -23,4 +28,34 @@ describe('Terminal', () => {
       );
     });
   });
+
+  describe('automatic flow control', () => {
+    it('should respect ctor flow control options', () => {
+      const pty = new terminalConstructor(SHELL, [], {handleFlowControl: true, flowControlPause: 'abc', flowControlResume: '123'});
+      assert.equal(pty.handleFlowControl, true);
+      assert.equal((pty as any)._flowControlPause, 'abc');
+      assert.equal((pty as any)._flowControlResume, '123');
+    });
+    it('should do flow control automatically', async function(): Promise<void> {
+      this.timeout(10000);
+      const pty = new terminalConstructor(SHELL, [], {handleFlowControl: true, flowControlPause: 'PAUSE', flowControlResume: 'RESUME'});
+      let read: string = '';
+      pty.on('data', data => read += data);
+      pty.on('pause', () => read += 'paused');
+      pty.on('resume', () => read += 'resumed');
+      pty.write('1');
+      pty.write('PAUSE');
+      pty.write('2');
+      pty.write('RESUME');
+      pty.write('3');
+      await (<any>pollUntil)(() => {
+        // important here: no data should be delivered between 'paused' and 'resumed'
+        if (process.platform === 'win32') {
+          read.endsWith('1\u001b[0Kpausedresumed2\u001b[0K3\u001b[0K');
+        } else {
+          read.endsWith('1pausedresumed23');
+        }
+      }, [], 20, 10);
+    });
+  });
 });

src/terminal.ts

@@ -13,6 +13,14 @@ import { IExitEvent } from './types';
 export const DEFAULT_COLS: number = 80;
 export const DEFAULT_ROWS: number = 24;
 
+/**
+ * Default messages to indicate PAUSE/RESUME for automatic flow control.
+ * To avoid conflicts with rebound XON/XOFF control codes (such as on-my-zsh),
+ * the sequences can be customized in `IPtyForkOptions`.
+ */
+const FLOW_CONTROL_PAUSE =  '\x13';   // defaults to XOFF
+const FLOW_CONTROL_RESUME = '\x11';   // defaults to XON
+
 export abstract class Terminal implements ITerminal {
   protected _socket: Socket;
   protected _pid: number;
@@ -28,6 +36,10 @@ export abstract class Terminal implements ITerminal {
   protected _writable: boolean;
 
   protected _internalee: EventEmitter;
+  protected _writeMethod: (data: string) => void;
+  private _flowControlPause: string;
+  private _flowControlResume: string;
+  public handleFlowControl: boolean;
 
   private _onData = new EventEmitter2<string>();
   public get onData(): IEvent<string> { return this._onData.event; }
@@ -56,6 +68,27 @@ export abstract class Terminal implements ITerminal {
     this._checkType('uid', opt.uid ? opt.uid : null, 'number');
     this._checkType('gid', opt.gid ? opt.gid : null, 'number');
     this._checkType('encoding', opt.encoding ? opt.encoding : null, 'string');
+
+    // setup flow control handling
+    this.handleFlowControl = !!(opt.handleFlowControl);
+    this._flowControlPause = opt.flowControlPause || FLOW_CONTROL_PAUSE;
+    this._flowControlResume = opt.flowControlResume || FLOW_CONTROL_RESUME;
+  }
+
+  public write(data: string): void {
+    if (this.handleFlowControl) {
+      // PAUSE/RESUME messages are not forwarded to the pty
+      if (data === this._flowControlPause) {
+        this.pause();
+        return;
+      }
+      if (data === this._flowControlResume) {
+        this.resume();
+        return;
+      }
+    }
+    // everything else goes to the real pty
+    this._writeMethod(data);
   }
 
   protected _forwardEvents(): void {
@@ -131,7 +164,6 @@ export abstract class Terminal implements ITerminal {
     this._socket.once(eventName, listener);
   }
 
-  public abstract write(data: string): void;
   public abstract resize(cols: number, rows: number): void;
   public abstract destroy(): void;
   public abstract kill(signal?: string): void;

src/unixTerminal.ts

@@ -156,6 +156,9 @@ export class UnixTerminal extends Terminal {
     });
 
     this._forwardEvents();
+
+    // attach write method
+    this._writeMethod = (data: string) => this._socket.write(data);
   }
 
   /**
@@ -217,10 +220,6 @@ export class UnixTerminal extends Terminal {
     return self;
   }
 
-  public write(data: string): void {
-    this._socket.write(data);
-  }
-
   public destroy(): void {
     this._close();
 

src/windowsTerminal.ts

@@ -120,6 +120,9 @@ export class WindowsTerminal extends Terminal {
     this._writable = true;
 
     this._forwardEvents();
+
+    // attach write method
+    this._writeMethod = (data: string) => this._defer(() => this._agent.inSocket.write(data));
   }
 
   /**
@@ -130,16 +133,6 @@ export class WindowsTerminal extends Terminal {
     throw new Error('open() not supported on windows, use Fork() instead.');
   }
 
-  /**
-   * Events
-   */
-
-  public write(data: string): void {
-    this._defer(() => {
-      this._agent.inSocket.write(data);
-    });
-  }
-
   /**
    * TTY
    */

typings/node-pty.d.ts

@@ -48,6 +48,23 @@ declare module 'node-pty' {
      * @see https://docs.microsoft.com/en-us/windows/console/createpseudoconsole
      */
     conptyInheritCursor?: boolean;
+    /**
+     * Whether to enable flow control handling (false by default). If enabled a message of `flowControlPause`
+     * will pause the socket and thus blocking the slave program execution due to buffer back pressure.
+     * A message of `flowControlResume` will resume the socket into flow mode.
+     * For performance reasons only a single message as a whole will match (no message part matching).
+     * If flow control is enabled the `flowControlPause` and `flowControlResume` messages are not forwarded to
+     * the underlying pseudoterminal.
+     */
+    handleFlowControl?: boolean;
+    /**
+     * The string that should pause the pty when `handleFlowControl` is true. Default is XOFF ('\x13').
+     */
+    flowControlPause?: string;
+    /**
+     * The string that should resume the pty when `handleFlowControl` is true. Default is XON ('\x11').
+     */
+    flowControlResume?: string;
   }
 
   /**
@@ -74,6 +91,12 @@ declare module 'node-pty' {
      */
     readonly process: string;
 
+    /**
+     * Whether to handle flow control. Useful to disable/re-enable flow control during runtime.
+     * Use this for binary data that is likely to contain the `flowControlPause` string by accident.
+     */
+    handleFlowControl: boolean;
+
     /**
      * Adds an event listener for when a data event fires. This happens when data is returned from
      * the pty.