lib: add tracing channel to diagnostics_channel

lib/diagnostics_channel.js

@@ -7,6 +7,8 @@ const {
   ObjectCreate,
   ObjectGetPrototypeOf,
   ObjectSetPrototypeOf,
+  PromisePrototypeThen,
+  ReflectApply,
   SymbolHasInstance,
 } = primordials;
 
@@ -136,10 +138,135 @@ function hasSubscribers(name) {
   return channel.hasSubscribers;
 }
 
+class TracingChannel {
+  constructor(name) {
+    this.name = name;
+    this.channels = {
+      start: channel(`${name}.start`),
+      end: channel(`${name}.end`),
+      asyncEnd: channel(`${name}.asyncEnd`),
+      error: channel(`${name}.error`)
+    };
+  }
+
+  get hasSubscribers() {
+    const { channels } = this;
+    for (const key in channels) {
+      if (channels[key].hasSubscribers) return true;
+    }
+    return false;
+  }
+
+  subscribe(handlers) {
+    let subscribed = true;
+    for (const key in handlers) {
+      if (!subscribe(`${this.name}.${key}`, handlers[key])) {
+        subscribed = false;
+      }
+    }
+    return subscribed;
+  }
+
+  unsubscribe(handlers) {
+    let unsubscribed = true;
+    for (const key in handlers) {
+      if (!unsubscribe(`${this.name}.${key}`, handlers[key])) {
+        unsubscribed = false;
+      }
+    }
+    return unsubscribed;
+  }
+
+  traceSync(fn, ctx = {}, thisArg, ...args) {
+    const { start, end, error } = this.channels;
+    start.publish(ctx);
+    try {
+      const result = ReflectApply(fn, thisArg, args);
+      ctx.result = result;
+      return result;
+    } catch (err) {
+      ctx.error = err;
+      error.publish(ctx);
+      throw err;
+    } finally {
+      end.publish(ctx);
+    }
+  }
+
+  tracePromise(fn, ctx = {}, thisArg, ...args) {
+    const { asyncEnd, start, end, error } = this.channels;
+    start.publish(ctx);
+
+    const reject = (err) => {
+      ctx.error = err;
+      error.publish(ctx);
+      asyncEnd.publish(ctx);
+      throw err;
+    };
+
+    const resolve = (result) => {
+      ctx.result = result;
+      asyncEnd.publish(ctx);
+      return result;
+    };
+
+    try {
+      const promise = ReflectApply(fn, thisArg, args);
+      return PromisePrototypeThen(promise, resolve, reject);
+    } catch (err) {
+      ctx.error = err;
+      error.publish(ctx);
+      throw err;
+    } finally {
+      end.publish(ctx);
+    }
+  }
+
+  traceCallback(fn, position = 0, ctx = {}, thisArg, ...args) {
+    const { start, end, asyncEnd, error } = this.channels;
+    start.publish(ctx);
+
+    function wrap(fn) {
+      return function wrappedCallback (err, res) {
+        if (err) {
+          ctx.error = err;
+          error.publish(ctx);
+        } else {
+          ctx.result = res;
+        }
+
+        asyncEnd.publish(ctx);
+        if (fn) {
+          return ReflectApply(fn, this, arguments);
+        }
+      }
+    }
+
+    if (position >= 0) {
+      args.splice(position, 1, wrap(args.at(position)));
+    }
+
+    try {
+      return ReflectApply(fn, thisArg, args);
+    } catch (err) {
+      ctx.error = err;
+      error.publish(ctx);
+      throw err;
+    } finally {
+      end.publish(ctx);
+    }
+  }
+}
+
+function tracingChannel(name) {
+  return new TracingChannel(name);
+}
+
 module.exports = {
   channel,
   hasSubscribers,
   subscribe,
+  tracingChannel,
   unsubscribe,
   Channel
 };

test/parallel/test-diagnostics-channel-tracing-channel-async-error.js

@@ -0,0 +1,45 @@
+'use strict';
+
+const common = require('../common');
+const dc = require('diagnostics_channel');
+const assert = require('assert');
+
+const channel = dc.tracingChannel('test');
+
+const expectedError = new Error('test');
+const input = { foo: 'bar' };
+const thisArg = { baz: 'buz' };
+
+function check(found) {
+  assert.deepStrictEqual(found, input);
+}
+
+const handlers = {
+  start: common.mustCall(check, 2),
+  end: common.mustCall(check, 2),
+  asyncEnd: common.mustCall(check, 2),
+  error: common.mustCall((found) => {
+    check(found);
+    assert.deepStrictEqual(found.error, expectedError);
+  }, 2)
+};
+
+channel.subscribe(handlers);
+
+channel.traceCallback(function (cb, err) {
+  assert.deepStrictEqual(this, thisArg);
+  setImmediate(cb, err);
+}, 0, input, thisArg, common.mustCall((err, res) => {
+  assert.strictEqual(err, expectedError);
+  assert.deepStrictEqual(res, undefined);
+}), expectedError);
+
+channel.tracePromise(function (value) {
+  assert.deepStrictEqual(this, thisArg);
+  return Promise.reject(value);
+}, input, thisArg, expectedError).then(
+  common.mustNotCall(),
+  common.mustCall(value => {
+    assert.deepStrictEqual(value, expectedError);
+  })
+);

test/parallel/test-diagnostics-channel-tracing-channel-async.js

@@ -0,0 +1,46 @@
+'use strict';
+
+const common = require('../common');
+const dc = require('diagnostics_channel');
+const assert = require('assert');
+
+const channel = dc.tracingChannel('test');
+
+const expectedResult = { foo: 'bar' };
+const input = { foo: 'bar' };
+const thisArg = { baz: 'buz' };
+
+function check(found) {
+  assert.deepStrictEqual(found, input);
+}
+
+const handlers = {
+  start: common.mustCall(check, 2),
+  end: common.mustCall(check, 2),
+  asyncEnd: common.mustCall((found) => {
+    check(found);
+    assert.strictEqual(found.error, undefined);
+    assert.deepStrictEqual(found.result, expectedResult);
+  }, 2),
+  error: common.mustNotCall()
+};
+
+channel.subscribe(handlers);
+
+channel.traceCallback(function (cb, err, res) {
+  assert.deepStrictEqual(this, thisArg);
+  setImmediate(cb, err, res);
+}, 0, input, thisArg, common.mustCall((err, res) => {
+  assert.strictEqual(err, null);
+  assert.deepStrictEqual(res, expectedResult);
+}), null, expectedResult);
+
+channel.tracePromise(function (value) {
+  assert.deepStrictEqual(this, thisArg);
+  return Promise.resolve(value);
+}, input, thisArg, expectedResult).then(
+  common.mustCall(value => {
+    assert.deepStrictEqual(value, expectedResult);
+  }),
+  common.mustNotCall()
+);

test/parallel/test-diagnostics-channel-tracing-channel-sync-error.js

@@ -0,0 +1,38 @@
+'use strict';
+
+const common = require('../common');
+const dc = require('diagnostics_channel');
+const assert = require('assert');
+
+const channel = dc.tracingChannel('test');
+
+const expectedError = new Error('test');
+const input = { foo: 'bar' };
+const thisArg = { baz: 'buz' };
+
+function check(found) {
+  assert.deepStrictEqual(found, input);
+}
+
+const handlers = {
+  start: common.mustCall(check),
+  end: common.mustCall(check),
+  asyncEnd: common.mustNotCall(),
+  error: common.mustCall((found) => {
+    check(found);
+    assert.deepStrictEqual(found.error, expectedError);
+  })
+};
+
+channel.subscribe(handlers);
+try {
+  channel.traceSync(function (err) {
+    assert.deepStrictEqual(this, thisArg);
+    assert.strictEqual(err, expectedError);
+    throw err;
+  }, input, thisArg, expectedError);
+
+  throw new Error('It should not reach this error');
+} catch (error) {
+  assert.deepStrictEqual(error, expectedError);
+}

test/parallel/test-diagnostics-channel-tracing-channel-sync.js

@@ -0,0 +1,37 @@
+'use strict';
+
+const common = require('../common');
+const dc = require('diagnostics_channel');
+const assert = require('assert');
+
+const channel = dc.tracingChannel('test');
+
+const expectedResult = { foo: 'bar' };
+const input = { foo: 'bar' };
+
+function check(found) {
+  assert.deepStrictEqual(found, input);
+}
+
+const handlers = {
+  start: common.mustCall(check),
+  end: common.mustCall((found) => {
+    check(found);
+    assert.deepStrictEqual(found.result, expectedResult);
+  }),
+  asyncEnd: common.mustNotCall(),
+  error: common.mustNotCall()
+};
+
+assert.strictEqual(channel.hasSubscribers, false);
+channel.subscribe(handlers);
+assert.strictEqual(channel.hasSubscribers, true);
+channel.traceSync(() => {
+  return expectedResult;
+}, input);
+
+channel.unsubscribe(handlers);
+assert.strictEqual(channel.hasSubscribers, false);
+channel.traceSync(() => {
+  return expectedResult;
+}, input);