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 `flowPause: string` and `flowResume: 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;
+  flowPause?: string;
+  flowResume?: string;
 }
 
 export interface IPtyOpenOptions {

src/terminal.test.ts

@@ -7,13 +7,17 @@ import * as assert from 'assert';
 import { WindowsTerminal } from './windowsTerminal';
 import { UnixTerminal } from './unixTerminal';
 
+const terminalConstructor = (process.platform === 'win32') ? WindowsTerminal : UnixTerminal;
+const SHELL = (process.platform === 'win32') ? 'cmd.exe' : '/bin/bash';
+
 let terminalCtor: WindowsTerminal | UnixTerminal;
 if (process.platform === 'win32') {
   terminalCtor = require('./windowsTerminal');
 } else {
   terminalCtor = require('./unixTerminal');
 }
 
+
 describe('Terminal', () => {
   describe('constructor', () => {
     it('should do basic type checks', () => {
@@ -23,4 +27,36 @@ describe('Terminal', () => {
       );
     });
   });
+
+  describe('automatic flow control', () => {
+    it('should respect ctor flow control options', () => {
+      const pty = new terminalConstructor(SHELL, [], {handleFlowControl: true, flowPause: 'abc', flowResume: '123'});
+      assert.equal(pty.handleFlowControl, true);
+      assert.equal((pty as any)._flowPause, 'abc');
+      assert.equal((pty as any)._flowResume, '123');
+    });
+    it('should do flow control automatically', function(done: Function): void {
+      this.timeout(10000);
+      const pty = new terminalConstructor(SHELL, [], {handleFlowControl: true, flowPause: 'PAUSE', flowResume: 'RESUME'});
+      const read: string[] = [];
+      pty.on('data', data => read.push(data));
+      pty.on('pause', () => read.push('paused'));
+      pty.on('resume', () => read.push('resumed'));
+      setTimeout(() => pty.write('1'), 7000);
+      setTimeout(() => pty.write('PAUSE'), 7200);
+      setTimeout(() => pty.write('2'), 7400);
+      setTimeout(() => pty.write('RESUME'), 7600);
+      setTimeout(() => pty.write('3'), 7800);
+      setTimeout(() => {
+        // important here: no data should be delivered between 'paused' and 'resumed'
+        if (process.platform === 'win32') {
+          // cmd.exe always clears to the end of line?
+          assert.deepEqual(read.slice(-5), ['1\u001b[0K', 'paused', 'resumed', '2\u001b[0K', '3\u001b[0K']);
+        } else {
+          assert.deepEqual(read.slice(-5), ['1', 'paused', 'resumed', '2', '3']);
+        }
+        done();
+      }, 9500);
+    });
+  });
 });

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,
+ * the sequences can be customized in `IPtyForkOptions`.
+ */
+const FLOW_PAUSE =  '\x13';   // defaults to XOFF
+const FLOW_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 _flowPause: string;
+  private _flowResume: 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._flowPause = opt.flowPause || FLOW_PAUSE;
+    this._flowResume = opt.flowResume || FLOW_RESUME;
+  }
+
+  public write(data: string): void {
+    if (this.handleFlowControl) {
+      // PAUSE/RESUME messages are not forwarded to the pty
+      if (data === this._flowPause) {
+        this.pause();
+        return;
+      }
+      if (data === this._flowResume) {
+        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

@@ -119,6 +119,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));
   }
 
   /**
@@ -129,16 +132,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

@@ -34,6 +34,23 @@ declare module 'node-pty' {
      * This setting does nothing on non-Windows.
      */
     experimentalUseConpty?: boolean;
+    /**
+     * Whether to enable flow control handling (false by default). If enabled a message of `flowPause`
+     * will pause the socket and thus blocking the slave program execution due to buffer back pressure.
+     * A message of `flowResume` 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 `flowPause` and `flowResume` 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').
+     */
+    flowPause: string;
+    /**
+     * The string that should resume the pty when `handleFlowControl` is true. Default is XON ('\x11').
+     */
+    flowResume: string;
   }
 
   /**
@@ -110,6 +127,12 @@ declare module 'node-pty' {
      * @throws Will throw when signal is used on Windows.
      */
     kill(signal?: string): void;
+
+    /**
+     * 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 `flowPause` string by accident.
+     */
+    handleFlowControl: boolean;
   }
 
   /**