fix: add write-stream test, fix implementation

packages/SwingSet/src/netstring.js

@@ -17,7 +17,13 @@ export function encode(data) {
 export function createWriter() {
   function transform(chunk, encoding, callback) {
     assert.equal(encoding, 'buffer');
-    callback(encode(chunk));
+    let err;
+    try {
+      this.push(encode(chunk));
+    } catch (e) {
+      err = e;
+    }
+    callback(err);
   }
   return new Transform({ transform });
 }
@@ -64,17 +70,20 @@ export function createReader() {
   function transform(chunk, encoding, callback) {
     assert.equal(encoding, 'buffer');
     buffered = Buffer.concat([buffered, chunk]);
-    let res = null;
+    let err;
     try {
       const { leftover, payloads } = decode(buffered);
       buffered = leftover;
       for (let i = 0; i < payloads.length; i += 1) {
         this.push(payloads[i]);
       }
-    } catch (err) {
-      res = err;
+    } catch (e) {
+      err = e;
     }
-    callback(res);
+    // we buffer all data internally, to accomodate netstrings larger than
+    // Transform's default buffer size, and callback() indicates that we've
+    // consumed the input
+    callback(err);
   }
 
   return new Transform({ transform });

packages/SwingSet/test/test-netstring.js

@@ -1,6 +1,7 @@
 import '@agoric/install-ses'; // adds 'harden' to global
+
 import test from 'ava';
-import { encode, decode } from '../src/netstring';
+import { encode, createWriter, decode } from '../src/netstring';
 
 test('encode', t => {
   function eq(input, expected) {
@@ -24,6 +25,18 @@ test('encode', t => {
   eq(umlautBuffer, expectedBuffer);
 });
 
+test('encode stream', async t => {
+  const e = createWriter();
+  const chunks = [];
+  e.on('data', data => chunks.push(data));
+  e.write('');
+  t.deepEqual(Buffer.concat(chunks), Buffer.from('0:,'));
+  e.write('hello');
+  t.deepEqual(Buffer.concat(chunks), Buffer.from('0:,5:hello,'));
+  e.end();
+  t.deepEqual(Buffer.concat(chunks), Buffer.from('0:,5:hello,'));
+});
+
 test('decode', t => {
   function eq(input, expPayloads, expLeftover) {
     const encPayloads = expPayloads.map(Buffer.from);