upgrade to stream 2

bin/cmd.js

@@ -68,7 +68,6 @@ if (argv._[0] === 'verify' || argv._[0] === 'run') {
 
         if (setup.stdin) {
             setup.stdin.pipe(v);
-            setup.stdin.resume();
         }
 
         if (setup.a && setup.a.resume) setup.a.resume();

bin/verify.js

@@ -1,6 +1,6 @@
 var spawn = require('child_process').spawn;
 var tuple = require('tuple-stream');
-var through = require('through');
+var through = require('through2');
 var split = require('split');
 var path = require('path');
 
@@ -79,28 +79,28 @@ module.exports = function (acmd, bcmd, opts) {
 
 function compare (actual, expected, opts) {
     var equal = true;
-    var output = through(write, end).pause();
+    var output = through.obj(write, end);
 
-    output.queue(COLORS.RESET);
+    output.push(COLORS.RESET);
 
     if (!opts.long) {
-        output.queue(wrap('ACTUAL', 30) + '     EXPECTED\n');
-        output.queue(wrap('------', 30) + '     --------\n');
+        output.push(wrap('ACTUAL', 30) + '     EXPECTED\n');
+        output.push(wrap('------', 30) + '     --------\n');
     }
 
     tuple(actual.pipe(split()), expected.pipe(split()))
         .pipe(output)
         .pipe(process.stdout)
     ;
-    output.resume();
+    output;
     return output;
 
-    function write (pair) {
+    function write (pair, _, next) {
         var eq = pair[0] === pair[1];
         equal = equal && eq;
 
         if (opts.long) {
-            this.queue('ACTUAL:   '
+            this.push('ACTUAL:   '
                 + COLORS[eq ? 'PASS' : 'FAIL']
                 + JSON.stringify(pair[0])
                 + COLORS.RESET + '\n'
@@ -110,19 +110,20 @@ function compare (actual, expected, opts) {
             );
         }
         else {
-            this.queue(
+            this.push(
                 COLORS[eq ? 'PASS' : 'FAIL']
                 + wrap(JSON.stringify(pair[0]), 30)
                 + ' ' + (eq ? '   ' : '!==') + ' '
                 + wrap(JSON.stringify(pair[1]), 30)
                 + '\n'
             );
         }
+        next();
     }
 
-    function end () {
-        output.queue(COLORS.RESET);
-        this.queue(null);
+    function end (done) {
+        this.push(COLORS.RESET);
+        done();
         this.emit(equal ? 'pass' : 'fail');
     }
 }

package.json

@@ -6,27 +6,27 @@
     "stream-adventure": "bin/cmd.js"
   },
   "dependencies": {
-    "x256": "~0.0.1",
-    "optimist": "~0.5.2",
-    "through": "~2.3.4",
-    "tuple-stream": "~0.0.2",
-    "terminal-menu": "~0.0.0",
-    "split": "~0.2.5",
-    "duplexer": "~0.1.1",
+    "browserify": "~2.35.4",
+    "chunky": "~0.0.0",
+    "clone": "~0.1.9",
     "concat-stream": "~1.0.0",
+    "duplexer2": "0.0.2",
     "hyperquest": "~0.1.6",
+    "mkdirp": "~0.3.5",
+    "optimist": "~0.5.2",
+    "provinces": "~0.2.0",
     "request": "~2.21.0",
-    "chunky": "~0.0.0",
-    "clone": "~0.1.9",
-    "wordwrap": "~0.0.2",
+    "split": "~0.2.5",
+    "stream-combiner": "~0.0.4",
     "tar": "~0.1.17",
-    "websocket-stream": "~0.2.0",
+    "terminal-menu": "~0.0.0",
+    "through2": "^0.6.3",
     "trumpet": "~1.6.2",
+    "tuple-stream": "0.0.2",
+    "websocket-stream": "~0.2.0",
+    "wordwrap": "~0.0.2",
     "ws": "~0.4.25",
-    "browserify": "~2.35.4",
-    "mkdirp": "~0.3.5",
-    "stream-combiner": "~0.0.4",
-    "provinces": "~0.2.0"
+    "x256": "~0.0.1"
   },
   "devDependencies": {
     "tape": "~2.3.0"

problems/combiner/solution.js

@@ -1,31 +1,32 @@
 var combine = require('stream-combiner');
-var through = require('through');
+var through = require('through2');
 var split = require('split');
 var zlib = require('zlib');
 
 module.exports = function () {
     var grouper = through(write, end);
     var current;
 
-    function write (line) {
-        if (line.length === 0) return;
+    function write (line, _, next) {
+        if (line.length === 0) return next();
         var row = JSON.parse(line);
 
         if (row.type === 'genre') {
             if (current) {
-                this.queue(JSON.stringify(current) + '\n');
+                this.push(JSON.stringify(current) + '\n');
             }
             current = { name: row.name, books: [] };
         }
         else if (row.type === 'book') {
             current.books.push(row.name);
         }
+        next();
     }
-    function end () {
+    function end (next) {
         if (current) {
-            this.queue(JSON.stringify(current) + '\n');
+            this.push(JSON.stringify(current) + '\n');
         }
-        this.queue(null);
+        next();
     }
 
     return combine(split(), grouper, zlib.createGzip());

problems/concat/setup.js

@@ -1,4 +1,4 @@
-var through = require('through');
+var through = require('through2');
 var clone = require('clone');
 var chunky = require('chunky');
 var wrap = require('wordwrap')(30);
@@ -12,9 +12,9 @@ module.exports = function () {
         var buf = bufs.pop();
         if (buf === undefined) {
             clearInterval(iv);
-            tr.queue(null);
+            tr.end();
         }
-        else tr.queue(buf.toString().split('').reverse().join(''));
+        else tr.write(buf.toString().split('').reverse().join(''));
     }, 50);
 
     return { args: [], stdin: tr };

problems/crypt/setup.js

@@ -1,4 +1,3 @@
-var through = require('through');
 var fs = require('fs');
 var crypto = require('crypto');
 var words = require('./words.json');
@@ -12,7 +11,7 @@ module.exports = function () {
         console.error('Please use node >= 0.10.');
         process.exit(1);
     }
-    input.pause();
+    input;
 
     fs.createReadStream(__dirname + '/finnegans_wake.txt').pipe(input);
     return { args: [ pw ], stdin: input, long: true };

problems/duplexer/command.js

@@ -1,18 +1,19 @@
-var through = require('through');
+var through2 = require('through2');
 var split = require('split');
 var combine = require('stream-combiner');
 var offset = Number(process.argv[2]);
 
 var tr = combine(split(), through(write));
 process.stdin.pipe(tr).pipe(process.stdout);
 
-function write (line) {
-    this.queue(line.replace(/[A-Za-z]/g, function (s) {
+function write (line, _, next) {
+    this.push(line.replace(/[A-Za-z]/g, function (s) {
         var c = s.charCodeAt(0);
         return String.fromCharCode(
             c < 97
             ? (c - 97 + offset) % 26 + 97
             : (c - 65 + offset) % 26 + 97
         );
     }) + '\n');
+    next();
 }

problems/duplexer/setup.js

@@ -1,6 +1,6 @@
-var through = require('through');
 var path = require('path');
 var cmd = __dirname + '/command.js';
+var through = require('through2');
 var words = [
     'beetle',
     'biscuit',
@@ -40,14 +40,15 @@ module.exports = function () {
         var run = require(path.resolve(args[0]));
         var ps = run(process.execPath, [ cmd, n ]);
         var queue = writes.slice();
-
-        var iv = setInterval(function () {
-            if (queue.length === 0) {
-                clearInterval(iv);
-                ps.end();
-            }
-            else ps.write(queue.shift() + '\n');
-        }, 50);
-        return ps;
+        var stream = through(function (chunk, _, next){
+            setTimeout(function (){
+                next(null, chunk);
+            }, 50);
+        })
+        var i = -1;
+        while (++i < queue.length) {
+            stream.write(queue[i]);
+        }
+        return stream.pipe(ps);
     }
 };

problems/duplexer/solution.js

@@ -1,5 +1,5 @@
 var spawn = require('child_process').spawn;
-var duplexer = require('duplexer');
+var duplexer = require('duplexer2');
 
 module.exports = function (cmd, args) {
     var ps = spawn(cmd, args);

problems/duplexer_redux/problem.txt

@@ -16,7 +16,7 @@ written objects with a 2-character `country` field as input, such as these:
 Create an object to keep a count of all the countries in the input. Once the
 input ends, call `counter.setCounts()` with your country counts.
 
-The `duplexer` module will again be very handy in this example.
+The `duplexer2` module will again be very handy in this example.
 
-If you use duplexer, make sure to `npm install duplexer` in the directory where
+If you use duplexer, make sure to `npm install duplexer2` in the directory where
 your solution file is located.

problems/duplexer_redux/setup.js

@@ -1,4 +1,3 @@
-var through = require('through');
 var cmd = __dirname + '/command.js';
 var path = require('path');
 var provinces = require('provinces');
@@ -16,7 +15,7 @@ module.exports = function () {
 
     function runCmd (args) {
         var fn = require(path.resolve(args[0]));
-        var counter = new Readable;
+        var counter = new Readable({objectMode: true});
         counter._read = function () {};
         counter.setCounts = function (counts) {
             var self = this;

problems/duplexer_redux/solution.js

@@ -1,13 +1,17 @@
 var duplexer = require('duplexer');
-var through = require('through');
+var through = require('through2').obj;
 
 module.exports = function (counter) {
     var counts = {};
     var input = through(write, end);
     return duplexer(input, counter);
 
-    function write (row) {
+    function write (row, _, next) {
         counts[row.country] = (counts[row.country] || 0) + 1;
+        next();
+    }
+    function end (done) {
+        counter.setCounts(counts);
+        done();
     }
-    function end () { counter.setCounts(counts) }
 };

problems/html_stream/setup.js

@@ -1,8 +1,7 @@
-var through = require('through');
 var fs = require('fs');
 
 module.exports = function () {
     var stdin = fs.createReadStream(__dirname + '/input.html');
-    stdin.pause();
+    stdin;
     return { args: [], stdin: stdin, long: true };
 };

problems/html_stream/solution.js

@@ -1,10 +1,11 @@
 var trumpet = require('trumpet');
-var through = require('through');
+var through = require('through2');
 var tr = trumpet();
 
 var loud = tr.select('.loud').createStream();
-loud.pipe(through(function (buf) {
-    this.queue(buf.toString().toUpperCase());
+loud.pipe(through(function (buf, _, next) {
+    this.push(buf.toString().toUpperCase());
+    next();
 })).pipe(loud);
 
 process.stdin.pipe(tr).pipe(process.stdout);

problems/http_client/setup.js

@@ -1,27 +1,27 @@
-var through = require('through');
+var through = require('through2');
 var http = require('http');
 var words = require('./words.json');
 
 module.exports = function () {
-    var tr = through().pause();
+    var tr = through();
 
     var count = 0;
     var iv = setInterval(function () {
         if (++count === 10) {
             clearInterval(iv);
-            return tr.queue(null);
+            return tr.end();
         }
         var word = words[Math.floor(Math.random() * words.length)];
-        tr.queue(word.toLowerCase() + '\n');
+        tr.write(word.toLowerCase() + '\n');
     }, 50);
 
     var server = http.createServer(function (req, res) {
         if (req.method !== 'POST') {
             res.end('not a POST request');
         }
         else {
-            req.pipe(through(function (buf) {
-                this.queue(buf.toString().replace(/\S/g, function (c) {
+            req.pipe(through(function (buf, _, next) {
+                this.push(buf.toString().replace(/\S/g, function (c) {
                     var x = c.charCodeAt(0);
                     if (/[a-z]/.test(c)) {
                         return String.fromCharCode(137 * (x - 97) % 26 + 97);
@@ -31,6 +31,7 @@ module.exports = function () {
                     }
                     else return c;
                 }));
+                next();
             })).pipe(res);
         }
     });

problems/http_server/setup.js

@@ -1,5 +1,5 @@
-var through = require('through');
-var duplexer = require('duplexer');
+var through = require('through2');
+var duplexer = require('duplexer2');
 var hyperquest = require('hyperquest');
 var words = require('./words.json');
 var spawn = require('child_process').spawn;

problems/http_server/solution.js

@@ -1,10 +1,11 @@
 var http = require('http');
-var through = require('through');
+var through = require('through2');
 
 var server = http.createServer(function (req, res) {
     if (req.method === 'POST') {
-        req.pipe(through(function (buf) {
-            this.queue(buf.toString().toUpperCase());
+        req.pipe(through(function (buf, _, next) {
+            this.push(buf.toString().toUpperCase());
+            next();
         })).pipe(res);
     }
     else res.end('send me a POST\n');