shim: refactor nodejs shim to work on node8 or node10

internal/shim/index.js

@@ -1,6 +1,4 @@
-
-var child = require('child_process');
-var byline = require('./byline');
+const child = require('child_process');
 
 /**
  * Debug env var.
@@ -13,13 +11,13 @@ const debug = process.env.DEBUG_SHIM;
  * many concurrent requests are outstanding.
  */
 
-var callbacks = {};
+const callbacks = new Map();
 
 /**
  * The last id attached to a request / callback pair
  */
 
-var lastId = (Date.now() / 1000) | 0;
+let lastId = (Date.now() / 1000) | 0;
 
 /**
  * nextId generates ids which will only be repeated every 2^52 times being generated
@@ -28,19 +26,56 @@ var lastId = (Date.now() / 1000) | 0;
 function nextId(){
   // Prevent bugs where integer precision wraps around on floating point numbers
   // (usually around 52-53 bits)
-  var id = (lastId + 1) | 0;
+  let id = (lastId + 1) | 0;
   if (id === lastId) {
     id = 1;
   }
+
   lastId = id;
   return String(id);
 }
 
+/**
+ * handleLine is responsible for taking a line of output from the child process
+ * and calling the appropiate callbacks.
+ */
+function handleLine(line) {
+  if (debug) {
+    console.log('[shim] parsing: `%s`', line);
+  }
+
+  let msg;
+  try {
+    msg = JSON.parse(line);
+  } catch (err) {
+    console.log('[shim] unexpected non-json line: `%s`', line);
+    return;
+  }
+
+  if (typeof msg.id !== 'string') {
+    console.log('[shim] unexpected line - do not use stdout: `%s`', line);
+    return;
+  }
+
+  const c = callbacks.get(msg.id);
+  callbacks.delete(msg.id);
+
+  if (!c) {
+    if (debug) {
+      console.log('[shim] unexpected duplicate response: `%s`', line);
+    }
+
+    return;
+  }
+
+  c(msg.error, msg.value);
+}
+
 /**
  * Child process for binary I/O.
  */
 
-var proc = child.spawn('./main', { stdio: ['pipe', 'pipe', process.stderr] });
+const proc = child.spawn('./main', { stdio: ['pipe', 'pipe', process.stderr] });
 
 proc.on('error', function(err){
   console.error('[shim] error: %s', err);
@@ -56,44 +91,64 @@ proc.on('exit', function(code, signal){
  * Newline-delimited JSON stdout.
  */
 
-var out = byline(proc.stdout)
-
-out.on('data', function(line){
-  if (debug) console.log('[shim] parsing: `%s`', line)
-
-  var msg;
-  try {
-    msg = JSON.parse(line);
-  } catch (err) {
-    console.log('[shim] unexpected non-json line: `%s`', line);
-    return
+// Chunks holds onto partial chunks received in the absense of a newline.
+// invariant: an array of Buffer objects, all of which do not have any newline characters
+let chunks = [];
+const NEWLINE = '\n'.charCodeAt(0);
+
+// Find successive newlines in this chunk, and pass them along to `handleChunk`
+function handleChunk(chunk) {
+  // since this current chunk can have multple lines inside of it
+  // keep track of how much of the current chunk we've consumed
+  let chunkPos = 0;
+  for (;;) {
+    // Find the first newline in the current, in the part of the current chunk we have not
+    // looked yet.
+    const newlinePos = chunk.indexOf(NEWLINE, chunkPos);
+
+    // We were not able to find any more newline characters in this chunk,
+    // save the remaineder in `chunks` for later processing
+    if (newlinePos === -1) {
+      chunks.push(chunk.slice(chunkPos));
+      break;
+    }
+
+    // We have found an end of a whole line, the beginning of the line will be the combination
+    // of all Buffers currently buffered in the `chunks` array (if any)
+    const start = chunk.slice(chunkPos, newlinePos);
+
+    chunks.push(start);
+    const line = Buffer.concat(chunks);
+    chunks = [];
+
+    // increase the chunk position, to skip over the last line we just found
+    chunkPos = newlinePos + 1;
+    handleLine(line)
   }
+}
 
-  if (typeof msg.id !== 'string') {
-    console.log('[shim] unexpected line - do not use stdout: `%s`', line);
-    return
-  }
+const out = proc.stdout;
 
-  const c = callbacks[msg.id];
-  delete callbacks[msg.id];
+out.on('readable', () => {
+  for (;;) {
+    const chunk = out.read();
+    if (chunk === null) {
+      break;
+    }
 
-  if (!c) {
-    if (debug) console.log('[shim] unexpected duplicate response: `%s`', line)
-    return
+    // Pump all data chunks into chunk handler
+    handleChunk(chunk);
   }
-
-  c(msg.error, msg.value);
 });
 
-
 /**
  * Handle events.
  */
 exports.handle = function(event, ctx, cb) {
   ctx.callbackWaitsForEmptyEventLoop = false;
 
   const id = nextId();
-  callbacks[id] = cb;
+  callbacks.set(id, cb);
 
   proc.stdin.write(JSON.stringify({
     "id": id,

internal/zip/zip.go

@@ -8,7 +8,7 @@ import (
 	"strings"
 
 	"github.com/pkg/errors"
-	"github.com/tj/go-archive"
+	archive "github.com/tj/go-archive"
 )
 
 var transform = archive.TransformFunc(func(r io.Reader, i os.FileInfo) (io.Reader, os.FileInfo) {
@@ -37,7 +37,7 @@ func Build(dir string) (io.ReadCloser, *archive.Stats, error) {
 		strings.NewReader(".*\n"),
 		strings.NewReader("\n!vendor\n!node_modules/**\n!.pypath/**\n"),
 		upignore,
-		strings.NewReader("\n!main\n!server\n!_proxy.js\n!byline.js\n!up.json\n!pom.xml\n!build.gradle\n!project.clj\ngin-bin\nup\n"))
+		strings.NewReader("\n!main\n!server\n!_proxy.js\n!up.json\n!pom.xml\n!build.gradle\n!project.clj\ngin-bin\nup\n"))
 
 	filter, err := archive.FilterPatterns(r)
 	if err != nil {

platform/lambda/lambda.go

@@ -907,10 +907,6 @@ func (p *Platform) injectProxy() error {
 		return errors.Wrap(err, "writing up-proxy")
 	}
 
-	if err := ioutil.WriteFile("byline.js", shim.MustAsset("byline.js"), 0755); err != nil {
-		return errors.Wrap(err, "writing byline.js")
-	}
-
 	if err := ioutil.WriteFile("_proxy.js", shim.MustAsset("index.js"), 0755); err != nil {
 		return errors.Wrap(err, "writing _proxy.js")
 	}
@@ -923,7 +919,6 @@ func (p *Platform) removeProxy() error {
 	log.Debugf("removing proxy")
 	os.Remove("main")
 	os.Remove("_proxy.js")
-	os.Remove("byline.js")
 	return nil
 }