Return explicit error when trying to stream a non-Readable stream

kanongil · kanongil · commit 289337d8393a · 2015-02-20T22:29:21.000+01:00
This replaces "TypeError: Cannot read property 'objectMode' of undefined" error response Fixes hapijs#2368 & hapijs#2301
diff --git a/lib/response.js b/lib/response.js
@@ -447,8 +447,11 @@ internals.Response.prototype._marshal = function (next) {
 internals.Response.prototype._streamify = function (source, next) {
 
     if (source instanceof Stream) {
-        var stream = (source.socket || source);
-        if (stream._readableState.objectMode) {
+        if (typeof source._read !== 'function' || typeof source._readableState !== 'object') {
+            return next(Boom.badImplementation('Stream is not a streams2 Readable'));
+        }
+
+        if (source._readableState.objectMode) {
             return next(Boom.badImplementation('Cannot reply with stream in object mode'));
         }
 
diff --git a/test/reply.js b/test/reply.js
@@ -251,7 +251,56 @@ describe('Reply', function () {
             });
         });
 
-        it('responds with an http client stream reply', function (done) {
+        it('errors on non-readable stream reply', function (done) {
+
+            var streamHandler = function (request, reply) {
+
+                var stream = new Stream();
+                stream.writable = true;
+
+                reply(stream);
+            };
+
+            var writableHandler = function (request, reply) {
+
+                var writable = new Stream.Writable();
+                writable._write = function () {};
+
+                reply(writable);
+            };
+
+            var server = new Hapi.Server({ debug: false });
+            server.connection();
+            server.route({ method: 'GET', path: '/stream', handler: streamHandler });
+            server.route({ method: 'GET', path: '/writable', handler: writableHandler });
+
+            var requestError;
+            server.on('request-error', function (request, err) {
+
+                requestError = err;
+            });
+
+            server.start(function () {
+
+                server.inject('/stream', function (res) {
+
+                    expect(res.statusCode).to.equal(500);
+                    expect(requestError).to.exist();
+                    expect(requestError.message).to.contain('Stream is not a streams2 Readable');
+
+                    requestError = undefined;
+                    server.inject('/writable', function (res) {
+
+                        expect(res.statusCode).to.equal(500);
+                        expect(requestError).to.exist();
+                        expect(requestError.message).to.contain('Stream is not a streams2 Readable');
+                        done();
+                    });
+                });
+            });
+        });
+
+        it('errors on an http client stream reply', function (done) {
 
             var handler = function (request, reply) {
 
@@ -272,7 +321,7 @@ describe('Reply', function () {
 
                 server.inject('/stream', function (res) {
 
-                    expect(res.statusCode).to.equal(200);
+                    expect(res.statusCode).to.equal(500);
                     done();
                 });
             });
diff --git a/test/transmit.js b/test/transmit.js
@@ -1318,6 +1318,78 @@ describe('transmission', function () {
             });
         });
 
+        it('does not leak classic stream data when passed to request and aborted', function (done) {
+
+            var destroyed = false;
+
+            var handler = function (request, reply) {
+
+                var stream = new Stream();
+                stream.readable = true;
+
+                var _read = function () {
+
+                    setImmediate(function () {
+
+                        if (paused) {
+                            return;
+                        }
+
+                        var chunk = new Array(1024).join('x');
+
+                        if (destroyed) {
+                            stream.emit('data', chunk);
+                            stream.readable = false;
+                            stream.emit('end');
+                        } else {
+                            stream.emit('data', chunk);
+                            _read();
+                        }
+                    });
+                };
+
+                var paused = true;
+                stream.resume = function () {
+
+                    if (paused) {
+                        paused = false;
+                        _read();
+                    }
+                };
+                stream.pause = function () {
+
+                    paused = true;
+                };
+
+                stream.resume();
+
+                stream.once('end', function () {
+
+                    done();
+                });
+
+                return reply(stream);
+            };
+
+            var server = new Hapi.Server();
+            server.connection();
+            server.route({ method: 'GET', path: '/', handler: handler });
+
+            server.start(function () {
+
+                Wreck.request('GET', 'http://localhost:' + server.info.port, {}, function (err, res) {
+
+                    res.on('data', function (chunk) {
+
+                        if (!destroyed) {
+                            destroyed = true;
+                            res.destroy();
+                        }
+                    });
+                });
+            });
+        });
+
         it('does not leak stream data when request timeouts before stream drains', function (done) {
 
             var handler = function (request, reply) {
