Fixes recorder to work with clients relying on 'readable' event

lib/recorder.js

@@ -197,17 +197,7 @@ function record(rec_options) {
         options = parse(options);
       }
 
-      var dataChunks = [];
-
-      res.on('data', function(data) {
-        debug(thisRecordingId, 'new', proto, 'data chunk');
-        dataChunks.push(data);
-      });
-
-      if (proto === 'https') {
-        options._https_ = true;
-      }
-
+      //  We put our 'end' listener to the front of the listener array.
       res.once('end', function() {
         debug(thisRecordingId, proto, 'intercepted request ended');
 
@@ -248,10 +238,83 @@ function record(rec_options) {
         }
       });
 
+      var dataChunks = [];
+
+      //  Give the actual client a chance to setup its listeners.
+      //  We will use the listener information to figure out
+      //  how we need to feed the intercepted data back to the client.
       if (callback) {
         callback(res, options, callback);
       }
 
+      //  Handle clients that listen to 'readable' by intercepting them
+      //  and feeding them the data manually.
+      var readableListeners = res.listeners('readable');
+      if (readableListeners && readableListeners.length > 0) {
+          //  We will replace the client's listeners with our own and manually
+          //  invoke them.
+          _.each(readableListeners, function(listener) {
+            res.removeListener('readable', listener);
+          });
+
+          //  Repleace read so that we can control what the client
+          //  listener will be reading.
+          var originalRead = res.read;
+          var readIndex = 0;
+          res.read = function() {
+            debug(thisRecordingId, 'client reading data on', proto, readIndex, dataChunks.length);
+
+            //  Feed the data to the client through from our collected data chunks.
+            if (dataChunks.length > readIndex) {
+              var chunk = dataChunks[readIndex];
+              ++readIndex;
+              return chunk;
+            } else {
+              return null;
+            }
+          }
+
+          //  Put our own listener instead of the removed client listener.
+          var onReadable = function(data) {
+            debug(thisRecordingId, 'new readable data on', proto);
+            var chunk;
+            //  Use the originalRead function to actually read the data.
+            while (null !== (chunk = originalRead.call(res))) {
+              debug('read', chunk);
+              dataChunks.push(chunk);
+            }
+            //  Manually invoke the user listeners emulating 'readable' event.
+            _.each(readableListeners, function(listener) {
+              listener();
+            });
+          };
+
+          res.on('readable', onReadable);
+      } else {
+        //  In all other cases we (for now at least) fall back on intercepting
+        //  'data' events.
+        debug('fall back on our original implementation');
+
+        //  Since we gave client the chance to setup its listeners
+        //  before us, we need to remove them and setup our own.
+        _.each(res.listeners('data'), function(listener) {
+          res.removeListener('data', listener);
+        })
+        _.each(res.listeners('readable'), function(listener) {
+          res.removeListener('readable', listener);
+        })
+
+        var onData = function(data) {
+          debug(thisRecordingId, 'new data chunk on', proto);
+          dataChunks.push(data);
+        };
+        res.on('data', onData);
+      }
+
+      if (proto === 'https') {
+        options._https_ = true;
+      }
+
     });
 
     var oldWrite = req.write;

tests/test_recorder.js

@@ -543,24 +543,87 @@ tap.test('records request headers except user-agent if enable_reqheaders_recordi
 });
 
 tap.test('includes query parameters from superagent', function(t) {
-    nock.restore();
-    nock.recorder.clear();
-    t.equal(nock.recorder.play().length, 0);
+  nock.restore();
+  nock.recorder.clear();
+  t.equal(nock.recorder.play().length, 0);
 
-    nock.recorder.rec({
-        dont_print: true,
-        output_objects: true
+  nock.recorder.rec({
+    dont_print: true,
+    output_objects: true
+  });
+
+  superagent.get('http://google.com')
+    .query({q: 'test search' })
+    .end(function(res) {
+      nock.restore();
+      var ret = nock.recorder.play();
+      t.true(ret.length >= 1);
+      t.equal(ret[0].path, '/?q=test%20search');
+      t.end();
     });
+});
+
+tap.test('works with clients listening for readable', function(t) {
+  nock.restore();
+  nock.recorder.clear();
+  t.equal(nock.recorder.play().length, 0);
+
+  var REQUEST_BODY = 'ABCDEF';
+  var RESPONSE_BODY = '012345';
+
+  //  Create test http server and perform the tests while it's up.
+  var testServer = http.createServer(function (req, res) {
+    res.write(RESPONSE_BODY);
+    res.end();
+  }).listen(8081, function(err) {
+
+    t.equal(err, undefined);
+
+    var options = { host:'localhost'
+                  , port:testServer.address().port
+                  , path:'/' }
+    ;
+
+    var rec_options = {
+      dont_print: true,
+      output_objects: true
+    };
+
+    nock.recorder.rec(rec_options);
+
+    var req = http.request(options, function(res) {
+      var readableCount = 0;
+      var chunkCount = 0;
+      res.on('readable', function() {
+        ++readableCount;
+        var chunk;
+        while (null !== (chunk = res.read())) {
+          ++chunkCount;
+        }
+      });
+      res.once('end', function() {
+        nock.restore();
+        var ret = nock.recorder.play();
+        t.equal(ret.length, 1);
+        ret = ret[0];
+        t.type(ret, 'object');
+        t.equal(readableCount, 1);
+        t.equal(chunkCount, 1);
+        t.equal(ret.scope, "http://localhost:" + options.port);
+        t.equal(ret.method, "GET");
+        t.equal(ret.body, REQUEST_BODY);
+        t.equal(ret.status, 200);
+        t.equal(ret.response, RESPONSE_BODY);
+        t.end();
+
+        //  Close the test server, we are done with it.
+        testServer.close();
+      });
+    });
+
+    req.end(REQUEST_BODY);
+  });
 
-    superagent.get('http://google.com')
-        .query({q: 'test search' })
-        .end(function(res) {
-            nock.restore();
-            var ret = nock.recorder.play();
-            t.true(ret.length >= 1);
-            t.equal(ret[0].path, '/?q=test%20search');
-            t.end();
-        });
 });
 
 tap.test("teardown", function(t) {
@@ -572,4 +635,4 @@ tap.test("teardown", function(t) {
   }
   t.deepEqual(leaks, [], 'No leaks');
   t.end();
-});
+});