From bfb8aa4534daf94ad59674062993d5bf980114e5 Mon Sep 17 00:00:00 2001
From: Anton Medvedev <anton@medv.io>
Date: Tue, 26 Feb 2019 02:06:12 +0700
Subject: [PATCH] Fix bug in stream processor

---
 README.md    |  2 +-
 index.js     | 74 ++--------------------------------------------------
 package.json |  3 ++-
 stream.js    | 74 ++++++++++++++++++++++++++++++++++++++++++++++++++++
 test.js      | 28 ++++++++++++++++++++
 5 files changed, 107 insertions(+), 74 deletions(-)
 create mode 100644 stream.js

diff --git a/README.md b/README.md
index 24c1dc8..9529f8f 100644
--- a/README.md
+++ b/README.md
@@ -6,7 +6,7 @@ _* Function eXecution_
 [![Build Status](https://travis-ci.org/antonmedv/fx.svg?branch=master)](https://travis-ci.org/antonmedv/fx)
 [![Npm Version](https://img.shields.io/npm/v/fx.svg)](https://www.npmjs.com/package/fx)
 [![Brew Version](https://img.shields.io/homebrew/v/fx.svg)](https://formulae.brew.sh/formula/fx)
-[![Snap Version](https://img.shields.io/badge/snap-12.0.0-blue.svg)](https://snapcraft.io/fx)
+[![Snap Version](https://img.shields.io/badge/snap-12.0.2-blue.svg)](https://snapcraft.io/fx)
 
 Command-line JSON processing tool
 
diff --git a/index.js b/index.js
index 2cb43ff..c223f8d 100755
--- a/index.js
+++ b/index.js
@@ -14,6 +14,7 @@ try {
 
 const print = require('./print')
 const reduce = require('./reduce')
+const stream = require('./stream')
 
 const usage = `
   Usage
@@ -54,7 +55,7 @@ void function main() {
     return
   }
 
-  const reader = stream()
+  const reader = stream(stdin, apply)
 
   stdin.on('readable', reader.read)
   stdin.on('end', () => {
@@ -130,74 +131,3 @@ function select(cb) {
     return json
   }
 }
-
-
-function stream() {
-  let buff = ''
-  let len = 0
-  let depth = 0
-  let isString = false
-
-  let count = 0
-  let head = ''
-  const check = (i) => {
-    if (depth <= 0) {
-      const input = buff.substring(0, len + i + 1)
-
-      if (count > 0) {
-        if (head !== '') {
-          const json = JSON.parse(head)
-          apply(json)
-          head = ''
-        }
-
-        const json = JSON.parse(input)
-        apply(json)
-      } else {
-        head = input
-      }
-
-      buff = buff.substring(len + i + 1)
-      len = buff.length
-      count++
-    }
-  }
-
-  return {
-    isStream() {
-      return count > 1
-    },
-    value() {
-      return head + buff
-    },
-    read() {
-      let chunk
-
-      while ((chunk = stdin.read())) {
-        len = buff.length
-        buff += chunk
-
-        for (let i = 0; i < chunk.length; i++) {
-          if (isString) {
-            if (chunk[i] === '"') {
-              if ((i === 0 && buff[len - 1] !== '\\') || (i > 0 && chunk[i - 1] !== '\\')) {
-                isString = false
-                check(i)
-              }
-            }
-            continue
-          }
-
-          if (chunk[i] === '{' || chunk[i] === '[') {
-            depth++
-          } else if (chunk[i] === '}' || chunk[i] === ']') {
-            depth--
-            check(i)
-          } else if (chunk[i] === '"') {
-            isString = true
-          }
-        }
-      }
-    }
-  }
-}
diff --git a/package.json b/package.json
index 7619ac1..c36a0c7 100644
--- a/package.json
+++ b/package.json
@@ -15,7 +15,8 @@
     "fx.js",
     "index.js",
     "print.js",
-    "reduce.js"
+    "reduce.js",
+    "stream.js"
   ],
   "scripts": {
     "test": "ava",
diff --git a/stream.js b/stream.js
new file mode 100644
index 0000000..00db811
--- /dev/null
+++ b/stream.js
@@ -0,0 +1,74 @@
+'use strict'
+
+function stream(from, cb) {
+  let buff = ''
+  let lastChar = ''
+  let len = 0
+  let depth = 0
+  let isString = false
+
+  let count = 0
+  let head = ''
+  const check = (i) => {
+    if (depth <= 0) {
+      const input = buff.substring(0, len + i + 1)
+
+      if (count > 0) {
+        if (head !== '') {
+          const json = JSON.parse(head)
+          cb(json)
+          head = ''
+        }
+
+        const json = JSON.parse(input)
+        cb(json)
+      } else {
+        head = input
+      }
+
+      buff = buff.substring(len + i + 1)
+      len = -i - 1
+      count++
+    }
+  }
+
+  return {
+    isStream() {
+      return count > 1
+    },
+    value() {
+      return head + buff
+    },
+    read() {
+      let chunk
+
+      while ((chunk = from.read())) {
+        len = buff.length
+        buff += chunk
+
+        for (let i = 0; i < chunk.length; i++) {
+          if (isString) {
+            if (chunk[i] === '"' && ((i === 0 && lastChar !== '\\') || (i > 0 && chunk[i - 1] !== '\\'))) {
+              isString = false
+              check(i)
+            }
+            continue
+          }
+
+          if (chunk[i] === '{' || chunk[i] === '[') {
+            depth++
+          } else if (chunk[i] === '}' || chunk[i] === ']') {
+            depth--
+            check(i)
+          } else if (chunk[i] === '"') {
+            isString = true
+          }
+        }
+
+        lastChar = chunk[chunk.length - 1]
+      }
+    }
+  }
+}
+
+module.exports = stream
diff --git a/test.js b/test.js
index 5ac70f8..052b180 100644
--- a/test.js
+++ b/test.js
@@ -1,6 +1,7 @@
 'use strict'
 const test = require('ava')
 const {execSync} = require('child_process')
+const stream = require('./stream')
 
 function fx(json, code = '') {
   return execSync(`echo '${JSON.stringify(json)}' | node index.js ${code}`).toString('utf8')
@@ -45,3 +46,30 @@ test('file argument', t => {
   const r = execSync(`node index.js package.json .name`).toString('utf8')
   t.deepEqual(r, 'fx\n')
 })
+
+test('stream', t => {
+  const input = `
+  {"index": 0} {"index": 1}
+  {"index": 2, "quote": "\\""}
+  {"index": 3} "Hello" "world"
+  {"index": 6, "key": "one \\"two\\" three"}
+  `
+  t.plan(7 * (input.length - 1))
+
+  for (let i = 0; i < input.length; i++) {
+    const parts = [input.substring(0, i), input.substring(i)]
+
+    const reader = stream(
+      {
+        read() {
+          return parts.shift()
+        }
+      },
+      json => {
+        t.pass()
+      }
+    )
+
+    reader.read()
+  }
+})