events,lib: don't require EE#listenerCount()

Now parts of our public and public-ish APIs fall back to old-style
listenerCount() if the emitter does not have a listenerCount function.

Fixes: nodejs#2655
Refs: 8f58fb9

PR-URL: nodejs#2661
Reviewed-By: Sakthipriyan Vairamani <thechargingvolcano@gmail.com>
Reviewed-By: Colin Ihrig <cjihrig@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>

lib/_http_server.js

@@ -374,7 +374,7 @@ function connectionListener(socket) {
       parser = null;
 
       var eventName = req.method === 'CONNECT' ? 'connect' : 'upgrade';
-      if (self.listenerCount(eventName) > 0) {
+      if (EventEmitter.listenerCount(self, eventName) > 0) {
         debug('SERVER have listener for %s', eventName);
         var bodyHead = d.slice(bytesParsed, d.length);
 
@@ -497,7 +497,7 @@ function connectionListener(socket) {
         (req.httpVersionMajor == 1 && req.httpVersionMinor == 1) &&
         continueExpression.test(req.headers['expect'])) {
       res._expect_continue = true;
-      if (self.listenerCount('checkContinue') > 0) {
+      if (EventEmitter.listenerCount(self, 'checkContinue') > 0) {
         self.emit('checkContinue', req, res);
       } else {
         res.writeContinue();

lib/_stream_readable.js

@@ -537,7 +537,7 @@ Readable.prototype.pipe = function(dest, pipeOpts) {
     debug('onerror', er);
     unpipe();
     dest.removeListener('error', onerror);
-    if (dest.listenerCount('error') === 0)
+    if (EE.listenerCount(dest, 'error') === 0)
       dest.emit('error', er);
   }
   // This is a brutally ugly hack to make sure that our error handler
@@ -586,7 +586,7 @@ function pipeOnDrain(src) {
     debug('pipeOnDrain', state.awaitDrain);
     if (state.awaitDrain)
       state.awaitDrain--;
-    if (state.awaitDrain === 0 && src.listenerCount('data')) {
+    if (state.awaitDrain === 0 && EE.listenerCount(src, 'data')) {
       state.flowing = true;
       flow(src);
     }

lib/events.js

@@ -395,10 +395,15 @@ EventEmitter.prototype.listeners = function listeners(type) {
 };
 
 EventEmitter.listenerCount = function(emitter, type) {
-  return emitter.listenerCount(type);
+  if (typeof emitter.listenerCount === 'function') {
+    return emitter.listenerCount(type);
+  } else {
+    return listenerCount.call(emitter, type);
+  }
 };
 
-EventEmitter.prototype.listenerCount = function listenerCount(type) {
+EventEmitter.prototype.listenerCount = listenerCount;
+function listenerCount(type) {
   const events = this._events;
 
   if (events) {

lib/stream.js

@@ -70,7 +70,7 @@ Stream.prototype.pipe = function(dest, options) {
   // don't leave dangling pipes when there are errors.
   function onerror(er) {
     cleanup();
-    if (this.listenerCount('error') === 0) {
+    if (EE.listenerCount(this, 'error') === 0) {
       throw er; // Unhandled stream error in pipe.
     }
   }

test/parallel/test-stream-pipe-without-listenerCount.js

@@ -0,0 +1,19 @@
+'use strict';
+const common = require('../common');
+const stream = require('stream');
+
+const r = new stream.Stream();
+r.listenerCount = undefined;
+
+const w = new stream.Stream();
+w.listenerCount = undefined;
+
+w.on('pipe', function() {
+  r.emit('error', new Error('Readable Error'));
+  w.emit('error', new Error('Writable Error'));
+});
+r.on('error', common.mustCall(noop));
+w.on('error', common.mustCall(noop));
+r.pipe(w);
+
+function noop() {};