Merge pull request #321 from willarmiros/emitters

replaced error listeners with errorMonitor

Author: shengxil

packages/core/lib/patchers/http_p.js

@@ -12,6 +12,7 @@ var contextUtils = require('../context_utils');
 var Utils = require('../utils');
 
 var logger = require('../logger');
+var events = require('events');
 
 /**
  * Wraps the http/https.request() and .get() calls to automatically capture information for the segment.
@@ -138,7 +139,11 @@ function enableCapture(module, downstreamXRayEnabled, subsegmentCallback) {
         subsegment.close(e);
       }
 
-      if (this._events && this._events.error && this._events.error.length === 1) {
+      // Only need to remove our listener & re-emit if we're not listening using the errorMonitor,
+      // e.g. the app is running on Node 10. Otherwise the errorMonitor will re-emit automatically.
+      // See: https://github.com/aws/aws-xray-sdk-node/issues/318
+      // TODO: Remove this logic once node 12 support is deprecated
+      if (!events.errorMonitor && this.listenerCount('error') <= 1) {
         this.removeListener('error', errorCapturer);
         this.emit('error', e);
       }
@@ -174,12 +179,17 @@ function enableCapture(module, downstreamXRayEnabled, subsegmentCallback) {
         } else {
           callback(res);
         }
-        // if no callback provided and there is only SDK added response listener,
-        // we consume the response so the actual end can fire.
-      } else if (res && res.listenerCount('end') === 1) {
+        // if no callback provided by user application, AND no explicit response listener
+        // added by user application, then we consume the response so the 'end' event fires
+        // See: https://nodejs.org/api/http.html#http_class_http_clientrequest
+      } else if (res.req && res.req.listenerCount('response') === 0) {
         res.resume();
       }
-    }).on('error', errorCapturer);
+    });
+
+    // Use errorMonitor if available (in Node 12.17+), otherwise fall back to standard error listener
+    // See: https://nodejs.org/dist/latest-v12.x/docs/api/events.html#events_eventemitter_errormonitor
+    req.on(events.errorMonitor || 'error', errorCapturer);
 
     return req;
   }

packages/core/test/unit/patchers/http_p.test.js

@@ -4,6 +4,7 @@ var chai = require('chai');
 var sinon = require('sinon');
 var sinonChai = require('sinon-chai');
 var URL = require('url');
+var events = require('events');
 
 var captureHTTPs = require('../../../lib/patchers/http_p').captureHTTPs;
 var captureHTTPsGlobal = require('../../../lib/patchers/http_p').captureHTTPsGlobal;
@@ -128,14 +129,15 @@ describe('HTTP/S', function() {
     });
 
     describe('on invocation', function() {
-      var capturedHttp, fakeRequest, fakeResponse, httpClient, requestSpy, sandbox;
+      var capturedHttp, fakeRequest, fakeResponse, httpClient, requestSpy, resumeSpy, sandbox;
 
       beforeEach(function() {
         sandbox = sinon.sandbox.create();
         segment = new Segment('test', traceId);
 
         fakeRequest = buildFakeRequest();
         fakeResponse = buildFakeResponse();
+        fakeResponse.req = fakeRequest;
 
         httpClient = { request: function(...args) {
           const callback = args[typeof args[1] === 'object' ? 2 : 1];
@@ -144,8 +146,10 @@ describe('HTTP/S', function() {
         }};
         httpClient.get = httpClient.request;
 
+        resumeSpy = sandbox.spy(fakeResponse, 'resume');
         requestSpy = sandbox.spy(httpClient, 'request');
-        capturedHttp = captureHTTPs(httpClient, true); });
+        capturedHttp = captureHTTPs(httpClient, true);
+      });
 
       afterEach(function() {
         sandbox.restore();
@@ -158,6 +162,22 @@ describe('HTTP/S', function() {
         resolveManualStub.should.have.been.calledWith(options);
       });
 
+      it('should consume the response if no callback is provided by user', function() {
+        capturedHttp.request(httpOptions);  // no callback
+        resumeSpy.should.have.been.calledOnce;
+      });
+
+      it('should not consume the response if a callback is provided by user', function() {
+        capturedHttp.request(httpOptions, () => {});
+        resumeSpy.should.not.have.been.called;
+      });
+
+      it('should not consume the response if a response listener is provided by user', function() {
+        fakeRequest.on('response', () => {});
+        capturedHttp.request(httpOptions);
+        resumeSpy.should.not.have.been.called;
+      });
+
       it('should create a new subsegment with name as hostname', function() {
         var options = {hostname: 'hostname', path: '/'};
         capturedHttp.request(options);
@@ -206,16 +226,14 @@ describe('HTTP/S', function() {
         assert.match(options.headers['X-Amzn-Trace-Id'], xAmznTraceId);
       });
 
-      if (process.version.startsWith('v') && process.version >= 'v10') {
-        it('should inject the tracing headers into the options if a URL is also provided', function() {
-          capturedHttp.request(`http://${httpOptions.host}${httpOptions.path}`, httpOptions);
-  
-          // example: 'Root=1-59138384-82ff54d5ba9282f0c680adb3;Parent=53af362e4e4efeb8;Sampled=1'
-          var xAmznTraceId = new RegExp('^Root=' + traceId + ';Parent=([a-f0-9]{16});Sampled=1$');
-          var options = requestSpy.firstCall.args[1];
-          assert.match(options.headers['X-Amzn-Trace-Id'], xAmznTraceId);
-        });
-      }
+      it('should inject the tracing headers into the options if a URL is also provided', function() {
+        capturedHttp.request(`http://${httpOptions.host}${httpOptions.path}`, httpOptions);
+
+        // example: 'Root=1-59138384-82ff54d5ba9282f0c680adb3;Parent=53af362e4e4efeb8;Sampled=1'
+        var xAmznTraceId = new RegExp('^Root=' + traceId + ';Parent=([a-f0-9]{16});Sampled=1$');
+        var options = requestSpy.firstCall.args[1];
+        assert.match(options.headers['X-Amzn-Trace-Id'], xAmznTraceId);
+      });
 
       it('should return the request object', function() {
         var request = capturedHttp.request(httpOptions);
@@ -385,6 +403,15 @@ describe('HTTP/S', function() {
           done();
         }, 50);
       });
+
+      if (process.version.startsWith('v') && process.version >= 'v12.17') {
+        it('should still re-emit if there are multiple errorMonitors attached', function() {
+          fakeRequest.on(events.errorMonitor, function() {});
+          fakeRequest.on(events.errorMonitor, function() {});
+
+          assert.throws(function() { fakeRequest.emitter.emit('error', error); });
+        });
+      }
     });
   });
 });

packages/mysql/lib/mysql_p.js

@@ -3,6 +3,7 @@
  */
 
 var AWSXRay = require('aws-xray-sdk-core');
+var events = require('events');
 var SqlData = AWSXRay.database.SqlData;
 
 var DATABASE_VERS = process.env.MYSQL_DATABASE_VERSION;
@@ -214,13 +215,14 @@ function captureOperation(name) {
       var errorCapturer = function (err) {
         subsegment.close(err);
 
-        if (this._events && this._events.error && this._events.error.length === 1) {
+        // TODO: Remove this logic once Node 10 is deprecated
+        if (!events.errorMonitor && this.listenerCount('error') <= 1) {
           this.removeListener('error', errorCapturer);
           this.emit('error', err);
         }
       };
 
-      command.on('error', errorCapturer);
+      command.on(events.errorMonitor || 'error', errorCapturer);
     }
 
     subsegment.addSqlData(createSqlData(config, command));

packages/postgres/lib/postgres_p.js

@@ -3,6 +3,7 @@
  */
 
 var AWSXRay = require('aws-xray-sdk-core');
+var events = require('events');
 var SqlData = AWSXRay.database.SqlData;
 
 var DATABASE_VERS = process.env.POSTGRES_DATABASE_VERSION;
@@ -106,13 +107,14 @@ function captureQuery() {
       var errorCapturer = function (err) {
         subsegment.close(err);
 
-        if (this._events && this._events.error && this._events.error.length === 1) {
+        // TODO: Remove this logic once Node 10 is deprecated
+        if (!events.errorMonitor && this.listenerCount('error') <= 1) {
           this.removeListener('error', errorCapturer);
           this.emit('error', err);
         }
       };
 
-      query.on('error', errorCapturer);
+      query.on(events.errorMonitor || 'error', errorCapturer);
     }
 
     subsegment.addSqlData(createSqlData(this.connectionParameters, query));