remove LineTransformStream

I went in favor of a native solution. LineTransformStream had some bugs.

Author: Almenon

CHANGELOG.md

@@ -2,6 +2,7 @@
 ### Changed
 - **BREAKING** Default python path changed back to `python` on Windows. [#237](https://github.com/extrabacon/python-shell/issues/237)
 - **BREAKING** `error` event renamed to `pythonError` event. [#118](https://github.com/extrabacon/python-shell/issues/118)
+- **BREAKING** `receive` methods removed in favor of `splitter` arguments in the constructor. This lets the default splitting logic reside in a reuseable stream transformer. Now if you have extra pipes you can reuse `newlineTransformer` to split incoming data into newline-seperated lines.
 
 ### Added
 - `error` event that is fired upon failure to launch process, among other things. [#118](https://github.com/extrabacon/python-shell/issues/118)

README.md

@@ -161,17 +161,19 @@ Creates an instance of `PythonShell` and starts the Python process
 * `script`: the path of the script to execute
 * `options`: the execution options, consisting of:
   * `mode`: Configures how data is exchanged when data flows through stdin and stdout. The possible values are:
-    * `text`: each line of data (ending with "\n") is emitted as a message (default)
-    * `json`: each line of data (ending with "\n") is parsed as JSON and emitted as a message
+    * `text`: each line of data is emitted as a message (default)
+    * `json`: each line of data is parsed as JSON and emitted as a message
     * `binary`: data is streamed as-is through `stdout` and `stdin`
-  * `formatter`: each message to send is transformed using this method, then appended with "\n"
-  * `parser`: each line of data (ending with "\n") is parsed with this function and its result is emitted as a message
-  * `stderrParser`: each line of logs (ending with "\n") is parsed with this function and its result is emitted as a message
+  * `formatter`: each message to send is transformed using this method, then appended with a newline
+  * `parser`: each line of data is parsed with this function and its result is emitted as a message
+  * `stderrParser`: each line of logs is parsed with this function and its result is emitted as a message
   * `encoding`: the text encoding to apply on the child process streams (default: "utf8")
   * `pythonPath`: The path where to locate the "python" executable. Default: "python3" ("python" for Windows)
   * `pythonOptions`: Array of option switches to pass to "python"
   * `scriptPath`: The default path where to look for scripts. Default is the current working directory.
   * `args`: Array of arguments to pass to the script
+* `stdoutSplitter`: splits stdout into chunks, defaulting to splitting into newline-seperated lines
+* `stderrSplitter`: splits stderr into chunks, defaulting to splitting into newline-seperated lines
 
 Other options are forwarded to `child_process.spawn`.
 
@@ -269,14 +271,6 @@ let shell = new PythonShell('script.py', { mode: 'json'});
 shell.send({ command: "do_stuff", args: [1, 2, 3] });
 ```
 
-#### `.receive(data)`
-
-Parses incoming data from the Python script written via stdout and emits `message` events. This method is called automatically as data is being received from stdout.
-
-#### `.receiveStderr(data)`
-
-Parses incoming logs from the Python script written via stderr and emits `stderr` events. This method is called automatically as data is being received from stderr.
-
 #### `.end(callback)`
 
 Closes the stdin stream, allowing the Python script to finish and exit. The optional callback is invoked when the process is terminated.
@@ -287,7 +281,7 @@ Terminates the python script. A kill signal may be provided by `signal`, if `sig
 
 #### event: `message`
 
-Fires when a chunk of data is parsed from the stdout stream via the `receive` method. If a `parser` method is specified, the result of this function will be the message value. This event is not emitted in binary mode.
+After the stdout stream is split into chunks by stdoutSplitter the chunks are parsed by the parser and a message event is emitted for each parsed chunk. This event is not emitted in binary mode.
 
 Example:
 
@@ -307,7 +301,7 @@ shell.on('message', function (message) {
 
 #### event: `stderr`
 
-Fires when a chunk of logs is parsed from the stderr stream via the `receiveStderr` method. If a `stderrParser` method is specified, the result of this function will be the message value. This event is not emitted in binary mode.
+After the stderr stream is split into chunks by stderrSplitter the chunks are parsed by the parser and a message event is emitted for each parsed chunk. This event is not emitted in binary mode.
 
 Example:
 
@@ -336,6 +330,10 @@ Fires when:
 
 If the process could not be spawned please double-check that python can be launched from the terminal.
 
+### NewlineTransformer
+
+A utility class for splitting stream data into newlines. Used as the default for stdoutSplitter and stderrSplitter if they are unspecified. You can use this class for any extra python streams if you'd like.
+
 ## Used By:
 
 Python-Shell is used by [arepl-vscode](https://github.com/almenon/arepl-vscode), [gitinspector](https://github.com/ejwa/gitinspector), [pyspreadsheet](https://github.com/extrabacon/pyspreadsheet), [AtlantOS Ocean Data QC](https://github.com/ocean-data-qc/ocean-data-qc) and more!

index.ts

@@ -2,10 +2,9 @@ import { EventEmitter } from 'events';
 import { ChildProcess, spawn, SpawnOptions, exec, execSync } from 'child_process';
 import { EOL as newline, tmpdir } from 'os';
 import { join, sep } from 'path'
-import { Readable, Writable } from 'stream'
+import { Readable, Transform, TransformCallback, Writable } from 'stream'
 import { writeFile, writeFileSync } from 'fs';
 import { promisify } from 'util';
-const LineTransformStream = require('line-transform-stream')
 
 function toArray<T>(source?: T | T[]): T[] {
     if (typeof source === 'undefined' || source === null) {
@@ -44,9 +43,9 @@ export interface Options extends SpawnOptions {
      * if binary is enabled message and stderr events will not be emitted
      */
     mode?: 'text' | 'json' | 'binary'
-    formatter?: (param: string) => any
-    parser?: (param: string) => any
-    stderrParser?: (param: string) => any
+    formatter?: string | ((param: string) => any)
+    parser?: string | ((param: string) => any)
+    stderrParser?: string | ((param: string) => any)
     encoding?: string
     pythonPath?: string
     /**
@@ -68,6 +67,28 @@ export class PythonShellError extends Error {
     exitCode?: number;
 }
 
+/**
+ * Takes in a string stream and emits batches seperated by newlines
+ */
+export class NewlineTransformer extends Transform {
+    // NewlineTransformer: Megatron's little known once-removed cousin
+    private _lastLineData: string;
+    _transform(chunk: any, encoding: string, callback: TransformCallback){
+        let data: string = chunk.toString()
+        if (this._lastLineData) data = this._lastLineData + data
+        const lines = data.split(newline)
+        this._lastLineData = lines.pop()
+        //@ts-ignore this works, node ignores the encoding if it's a number
+        lines.forEach(this.push.bind(this))
+        callback()
+    }
+    _flush(done: TransformCallback){
+        if (this._lastLineData) this.push(this._lastLineData)
+        this._lastLineData = null;
+        done()
+    }
+}
+
 /**
  * An interactive Python shell exchanging data through stdio
  * @param {string} script    The python script to execute
@@ -103,7 +124,7 @@ export class PythonShell extends EventEmitter {
      * @param scriptPath path to script. Relative to current directory or options.scriptFolder if specified
      * @param options 
      */
-    constructor(scriptPath: string, options?: Options) {
+    constructor(scriptPath: string, options?: Options, stdoutSplitter: Transform = null, stderrSplitter: Transform = null) {
         super();
 
         /**
@@ -151,16 +172,24 @@ export class PythonShell extends EventEmitter {
         // Node buffers stdout&stderr in batches regardless of newline placement
         // This is troublesome if you want to recieve distinct individual messages
         // for example JSON parsing breaks if it recieves partial JSON
-        // so we use LineTransformStream to emit each batch seperated by newline
+        // so we use newlineTransformer to emit each batch seperated by newline
         if (this.parser && this.stdout) {
-            this.stdout.pipe(new LineTransformStream((data) => {
-                this.emit('message', self.parser(data));
-            }))
+            if(!stdoutSplitter) stdoutSplitter = new NewlineTransformer()
+            // note that setting the encoding turns the chunk into a string
+            stdoutSplitter.setEncoding(options.encoding || 'utf8')
+            this.stdout.pipe(stdoutSplitter).on('data', (chunk: string) => {
+                this.emit('message', self.parser(chunk));
+            });
         }
+
+        // listen to stderr and emit errors for incoming data
         if (this.stderrParser && this.stderr) {
-            this.stderr.pipe(new LineTransformStream((data) => {
-                this.emit('stderr', self.stderrParser(data));
-            }))
+            if(!stderrSplitter) stderrSplitter = new NewlineTransformer()
+            // note that setting the encoding turns the chunk into a string
+            stderrSplitter.setEncoding(options.encoding || 'utf8')
+            this.stderr.pipe(stderrSplitter).on('data', (chunk: string) => {
+                this.emit('stderr', self.stderrParser(chunk));
+            });
         }
 
         if (this.stderr) {

package.json

@@ -10,12 +10,10 @@
     "appveyorTest": "tsc -p ./ && nyc mocha --reporter mocha-appveyor-reporter test/*.js",
     "compile": "tsc -watch -p ./"
   },
-  "dependencies": {
-    "line-transform-stream": "^0.1.0"
-  },
   "devDependencies": {
     "@types/mocha": "^8.2.1",
     "@types/node": "^10.5.2",
+    "@types/should": "^13.0.0",
     "mocha": "^8.2.1",
     "mocha-appveyor-reporter": "^0.4.0",
     "should": "^13.2.1",

test/python/echo_text_with_newline_control.py

@@ -0,0 +1,5 @@
+import sys, json, os
+
+for line in sys.stdin:
+  line = line.replace('$', os.linesep)
+  print(line[:-1], end='')

test/test-python-shell.ts

@@ -328,7 +328,7 @@ describe('PythonShell', function () {
         });
     });
 
-    describe('.receive(data)', function () {
+    describe('stdout', function () {
         it('should emit messages as strings when mode is "text"', function (done) {
             let pyshell = new PythonShell('echo_text.py', {
                 mode: 'text'
@@ -358,15 +358,16 @@ describe('PythonShell', function () {
             }).end(done);
         });
         it('should properly buffer partial messages', function (done) {
-            let pyshell = new PythonShell('echo_text.py', {
-                mode: 'json'
+            // echo_text_with_newline_control echoes text with $'s replaced with newlines
+            let pyshell = new PythonShell('echo_text_with_newline_control.py', {
+                mode: 'text'
             });
             pyshell.on('message', (message) => {
                 console.log(message)
-                message.should.be.an.Object;
-                message.should.eql({ a: true });
-            }).send('{"a"').send(':').send('true}' + newline + '{').send('"a":true}' + newline).end(() => {
-                console.log('done called')
+                let messageObject = JSON.parse(message)
+                messageObject.should.be.an.Object;
+                messageObject.should.eql({ a: true });
+            }).send('{"a"').send(':').send('true}${').send('"a":true}$').end(() => {
                 done()
             });
         });
@@ -375,10 +376,10 @@ describe('PythonShell', function () {
                 args: ['hello', 'world'],
                 mode: 'binary'
             });
-            pyshell.receive = function () {
+            pyshell.on('message', ()=>{
                 done('should not emit messages in binary mode');
                 return undefined
-            };
+            });
             pyshell.end(done);
         });
         it('should use a custom parser function', function (done) {
@@ -399,7 +400,7 @@ describe('PythonShell', function () {
         });
     });
 
-    describe('.receiveStderr(data)', function () {
+    describe('stderr', function () {
         it('should emit stderr logs as strings when mode is "text"', function (done) {
             let pyshell = new PythonShell('stderrLogging.py', {
                 mode: 'text'
@@ -415,13 +416,14 @@ describe('PythonShell', function () {
         });
         it('should not be invoked when mode is "binary"', function (done) {
             let pyshell = new PythonShell('stderrLogging.py', {
-                mode: 'binary'
+                stderrParser: 'binary'
             });
-            pyshell.receiveStderr = function () {
+            pyshell.on('stderr', ()=>{
                 done('should not emit stderr in binary mode');
-                return undefined
-            };
-            pyshell.end(done);
+            });
+            pyshell.end(()=>{
+                done()
+            });
         });
         it('should use a custom parser function', function (done) {
             let pyshell = new PythonShell('stderrLogging.py', {
@@ -501,6 +503,15 @@ describe('PythonShell', function () {
                 done();
             });
         });
+        it('should work in json mode', function (done) {
+            let pyshell = new PythonShell('error.py', {mode: 'json'});
+            pyshell.on('pythonError', function (err) {
+                err.stack.should.containEql('----- Python Traceback -----');
+                err.stack.should.containEql('File "test' + sep + 'python' + sep + 'error.py", line 4');
+                err.stack.should.containEql('File "test' + sep + 'python' + sep + 'error.py", line 6');
+                done();
+            });
+        });
     });
 
     describe('.terminate()', function () {