repo_description: Node.js JavaScript runtime ✨🐢🚀✨
repo_topics: javascript js linux macos mit node nodejs runtime windows
code: 'use strict';
  
  const {
    ArrayPrototypeAt,
    ArrayPrototypeIndexOf,
    ArrayPrototypePush,
    ArrayPrototypePushApply,
    ArrayPrototypeSlice,
    ArrayPrototypeSplice,
    ObjectDefineProperty,
    ObjectGetPrototypeOf,
    ObjectSetPrototypeOf,
    PromisePrototypeThen,
    PromiseReject,
    ReflectApply,
    SafeFinalizationRegistry,
    SafeMap,
    SymbolDispose,
    SymbolHasInstance,
  } = primordials;
  
  const {
    codes: {
      ERR_INVALID_ARG_TYPE,
    },
  } = require('internal/errors');
  const {
    validateFunction,
  } = require('internal/validators');
  
  const { triggerUncaughtException } = internalBinding('errors');
  
  const dc_binding = internalBinding('diagnostics_channel');
  const { subscribers: subscriberCounts } = dc_binding;
  
  const { WeakReference } = require('internal/util');
  const { isPromise } = require('internal/util/types');
  
  // Can't delete when weakref count reaches 0 as it could increment again.
  // Only GC can be used as a valid time to clean up the channels map.
  class WeakRefMap extends SafeMap {
    #finalizers = new SafeFinalizationRegistry((key) => {
      // Check that the key doesn't have any value before deleting, as the WeakRef for the key
      // may have been replaced since finalization callbacks aren't synchronous with GC.
      if (!this.has(key)) this.delete(key);
    });
  
    set(key, value) {
      this.#finalizers.register(value, key);
      return super.set(key, new WeakReference(value));
    }
  
    get(key) {
      return super.get(key)?.get();
    }
  
    has(key) {
      return !!this.get(key);
    }
  
    incRef(key) {
      return super.get(key)?.incRef();
    }
  
    decRef(key) {
      return super.get(key)?.decRef();
    }
  }
  
  function markActive(channel) {
    // eslint-disable-next-line no-use-before-define
    ObjectSetPrototypeOf(channel, ActiveChannel.prototype);
    channel._subscribers = [];
    channel._stores = new SafeMap();
  }
  
  function maybeMarkInactive(channel) {
    // When there are no more active subscribers or bound, restore to fast prototype.
    if (!channel._subscribers.length && !channel._stores.size) {
      // eslint-disable-next-line no-use-before-define
      ObjectSetPrototypeOf(channel, Channel.prototype);
      channel._subscribers = undefined;
      channel._stores = undefined;
    }
  }
  
  class RunStoresScope {
    #stack;
  
    constructor(activeChannel, data) {
      // eslint-disable-next-line no-restricted-globals
      using stack = new DisposableStack();
  
      // Enter stores using withScope
      if (activeChannel._stores) {
        for (const entry of activeChannel._stores.entries()) {
          const store = entry[0];
          const transform = entry[1];
  
          let newContext = data;
          if (transform) {
            try {
              newContext = transform(data);
            } catch (err) {
              process.nextTick(() => {
                triggerUncaughtException(err, false);
              });
              continue;
            }
          }
  
          stack.use(store.withScope(newContext));
        }
      }
  
      // Publish data
      activeChannel.publish(data);
  
      // Transfer ownership of the stack
      this.#stack = stack.move();
    }
  
    [SymbolDispose]() {
      this.#stack[SymbolDispose]();
    }
  }
  
  // TODO(qard): should there be a C++ channel interface?
  class ActiveChannel {
    subscribe(subscription) {
      validateFunction(subscription, 'subscription');
      this._subscribers = ArrayPrototypeSlice(this._subscribers);
      ArrayPrototypePush(this._subscribers, subscription);
      channels.incRef(this.name);
      if (this._index !== undefined) subscriberCounts[this._index]++;
    }
  
    unsubscribe(subscription) {
      const index = ArrayPrototypeIndexOf(this._subscribers, subscription);
      if (index === -1) return false;
  
      const before = ArrayPrototypeSlice(this._subscribers, 0, index);
      const after = ArrayPrototypeSlice(this._subscribers, index + 1);
      this._subscribers = before;
      ArrayPrototypePushApply(this._subscribers, after);
  
      channels.decRef(this.name);
      if (this._index !== undefined) subscriberCounts[this._index]--;
      maybeMarkInactive(this);
  
      return true;
    }
  
    bindStore(store, transform) {
      const replacing = this._stores.has(store);
      if (!replacing) {
        channels.incRef(this.name);
        if (this._index !== undefined) subscriberCounts[this._index]++;
      }
      this._stores.set(store, transform);
    }
  
    unbindStore(store) {
      if (!this._stores.has(store)) {
        return false;
      }
  
      this._stores.delete(store);
  
      channels.decRef(this.name);
      if (this._index !== undefined) subscriberCounts[this._index]--;
      maybeMarkInactive(this);
  
      return true;
    }
  
    get hasSubscribers() {
      return true;
    }
  
    publish(data) {
      const subscribers = this._subscribers;
      for (let i = 0; i < (subscribers?.length || 0); i++) {
        try {
          const onMessage = subscribers[i];
          onMessage(data, this.name);
        } catch (err) {
          process.nextTick(() => {
            triggerUncaughtException(err, false);
          });
        }
      }
    }
  
    withStoreScope(data) {
      return new RunStoresScope(this, data);
    }
  
    runStores(data, fn, thisArg, ...args) {
      // eslint-disable-next-line no-unused-vars
      using scope = this.withStoreScope(data);
      return ReflectApply(fn, thisArg, args);
    }
  }
  
  class Channel {
    constructor(name) {
      this._subscribers = undefined;
      this._stores = undefined;
      this.name = name;
      if (typeof name === 'string') {
        this._index = dc_binding.getOrCreateChannelIndex(name);
      }
  
      channels.set(name, this);
    }
  
    static [SymbolHasInstance](instance) {
      const prototype = ObjectGetPrototypeOf(instance);
      return prototype === Channel.prototype ||
             prototype === ActiveChannel.prototype;
    }
  
    subscribe(subscription) {
      markActive(this);
      this.subscribe(subscription);
    }
  
    unsubscribe() {
      return false;
    }
  
    bindStore(store, transform) {
      markActive(this);
      this.bindStore(store, transform);
    }
  
    unbindStore() {
      return false;
    }
  
    get hasSubscribers() {
      return false;
    }
  
    publish() {}
  
    runStores(data, fn, thisArg, ...args) {
      return ReflectApply(fn, thisArg, args);
    }
  
    withStoreScope() {
      // Return no-op disposable for inactive channels
      return {
        [SymbolDispose]() {},
      };
    }
  }
  
  const channels = new WeakRefMap();
  
  function channel(name) {
    const channel = channels.get(name);
    if (channel) return channel;
  
    if (typeof name !== 'string' && typeof name !== 'symbol') {
      throw new ERR_INVALID_ARG_TYPE('channel', ['string', 'symbol'], name);
    }
  
    return new Channel(name);
  }
  
  function subscribe(name, subscription) {
    return channel(name).subscribe(subscription);
  }
  
  function unsubscribe(name, subscription) {
    return channel(name).unsubscribe(subscription);
  }
  
  function hasSubscribers(name) {
    const channel = channels.get(name);
    if (!channel) return false;
  
    return channel.hasSubscribers;
  }
  
  const boundedEvents = [
    'start',
    'end',
  ];
  
  function assertChannel(value, name) {
    if (!(value instanceof Channel)) {
      throw new ERR_INVALID_ARG_TYPE(name, ['Channel'], value);
    }
  }
  
  function emitNonThenableWarning(fn) {
    process.emitWarning(`tracePromise was called with the function '${fn.name || '<anonymous>'}', ` +
                        'which returned a non-thenable.');
  }
  
  function channelFromMap(nameOrChannels, name, className) {
    if (typeof nameOrChannels === 'string') {
      return channel(`tracing:${nameOrChannels}:${name}`);
    }
  
    if (typeof nameOrChannels === 'object' && nameOrChannels !== null) {
      const channel = nameOrChannels[name];
      assertChannel(channel, `nameOrChannels.${name}`);
      return channel;
    }
  
    throw new ERR_INVALID_ARG_TYPE('nameOrChannels',
                                   ['string', 'object', className],
                                   nameOrChannels);
  }
  
  class BoundedChannelScope {
    #context;
    #end;
    #scope;
  
    constructor(boundedChannel, context) {
      // Only proceed if there are subscribers
      if (!boundedChannel.hasSubscribers) {
        return;
      }
  
      const { start, end } = boundedChannel;
      this.#context = context;
      this.#end = end;
  
      // Use RunStoresScope for the start channel
      this.#scope = new RunStoresScope(start, context);
    }
  
    [SymbolDispose]() {
      if (!this.#scope) {
        return;
      }
  
      // Publish end event
      this.#end.publish(this.#context);
  
      // Dispose the start scope to restore stores
      this.#scope[SymbolDispose]();
      this.#scope = undefined;
    }
  }
  
  class BoundedChannel {
    constructor(nameOrChannels) {
      for (let i = 0; i < boundedEvents.length; ++i) {
        const eventName = boundedEvents[i];
        ObjectDefineProperty(this, eventName, {
          __proto__: null,
          value: channelFromMap(nameOrChannels, eventName, 'BoundedChannel'),
        });
      }
    }
  
    get hasSubscribers() {
      return this.start?.hasSubscribers ||
        this.end?.hasSubscribers;
    }
  
    subscribe(handlers) {
      for (let i = 0; i < boundedEvents.length; ++i) {
        const name = boundedEvents[i];
        if (!handlers[name]) continue;
  
        this[name]?.subscribe(handlers[name]);
      }
    }
  
    unsubscribe(handlers) {
      let done = true;
  
      for (let i = 0; i < boundedEvents.length; ++i) {
        const name = boundedEvents[i];
        if (!handlers[name]) continue;
  
        if (!this[name]?.unsubscribe(handlers[name])) {
          done = false;
        }
      }
  
      return done;
    }
  
    withScope(context = {}) {
      return new BoundedChannelScope(this, context);
    }
  
    run(context, fn, thisArg, ...args) {
      context ??= {};
      // eslint-disable-next-line no-unused-vars
      using scope = this.withScope(context);
      return ReflectApply(fn, thisArg, args);
    }
  }
  
  function boundedChannel(nameOrChannels) {
    return new BoundedChannel(nameOrChannels);
  }
  
  class TracingChannel {
    #callWindow;
    #continuationWindow;
  
    constructor(nameOrChannels) {
      // Create a BoundedChannel for start/end (call window)
      if (typeof nameOrChannels === 'string') {
        this.#callWindow = new BoundedChannel(nameOrChannels);
        this.#continuationWindow = new BoundedChannel({
          start: channel(`tracing:${nameOrChannels}:asyncStart`),
          end: channel(`tracing:${nameOrChannels}:asyncEnd`),
        });
      } else if (typeof nameOrChannels === 'object') {
        this.#callWindow = new BoundedChannel({
          start: nameOrChannels.start,
          end: nameOrChannels.end,
        });
        this.#continuationWindow = new BoundedChannel({
          start: nameOrChannels.asyncStart,
          end: nameOrChannels.asyncEnd,
        });
      }
  
      // Create individual channel for error
      ObjectDefineProperty(this, 'error', {
        __proto__: null,
        value: channelFromMap(nameOrChannels, 'error', 'TracingChannel'),
      });
    }
  
    get start() {
      return this.#callWindow.start;
    }
  
    get end() {
      return this.#callWindow.end;
    }
  
    get asyncStart() {
      return this.#continuationWindow.start;
    }
  
    get asyncEnd() {
      return this.#continuationWindow.end;
    }
  
    get hasSubscribers() {
      return this.#callWindow.hasSubscribers ||
        this.#continuationWindow.hasSubscribers ||
        this.error?.hasSubscribers;
    }
  
    subscribe(handlers) {
      // Subscribe to call window (start/end)
      if (handlers.start || handlers.end) {
        this.#callWindow.subscribe({
          start: handlers.start,
          end: handlers.end,
        });
      }
  
      // Subscribe to continuation window (asyncStart/asyncEnd)
      if (handlers.asyncStart || handlers.asyncEnd) {
        this.#continuationWindow.subscribe({
          start: handlers.asyncStart,
          end: handlers.asyncEnd,
        });
      }
  
      // Subscribe to error channel
      if (handlers.error) {
        this.error.subscribe(handlers.error);
      }
    }
  
    unsubscribe(handlers) {
      let done = true;
  
      // Unsubscribe from call window
      if (handlers.start || handlers.end) {
        if (!this.#callWindow.unsubscribe({
          start: handlers.start,
          end: handlers.end,
        })) {
          done = false;
        }
      }
  
      // Unsubscribe from continuation window
      if (handlers.asyncStart || handlers.asyncEnd) {
        if (!this.#continuationWindow.unsubscribe({
          start: handlers.asyncStart,
          end: handlers.asyncEnd,
        })) {
          done = false;
        }
      }
  
      // Unsubscribe from error channel
      if (handlers.error) {
        if (!this.error.unsubscribe(handlers.error)) {
          done = false;
        }
      }
  
      return done;
    }
  
    traceSync(fn, context = {}, thisArg, ...args) {
      if (!this.hasSubscribers) {
        return ReflectApply(fn, thisArg, args);
      }
  
      const { error } = this;
  
      // eslint-disable-next-line no-unused-vars
      using scope = this.#callWindow.withScope(context);
      try {
        const result = ReflectApply(fn, thisArg, args);
        context.result = result;
        return result;
      } catch (err) {
        context.error = err;
        error.publish(context);
        throw err;
      }
    }
  
    tracePromise(fn, context = {}, thisArg, ...args) {
      if (!this.hasSubscribers) {
        const result = ReflectApply(fn, thisArg, args);
        if (typeof result?.then !== 'function') {
          emitNonThenableWarning(fn);
        }
        return result;
      }
  
      const { error } = this;
      const continuationWindow = this.#continuationWindow;
  
      function reject(err) {
        context.error = err;
        error.publish(context);
        // Use continuation window for asyncStart/asyncEnd
        // eslint-disable-next-line no-unused-vars
        using scope = continuationWindow.withScope(context);
        // TODO: Is there a way to have asyncEnd _after_ the continuation?
        return PromiseReject(err);
      }
  
      function resolve(result) {
        context.result = result;
        // Use continuation window for asyncStart/asyncEnd
        // eslint-disable-next-line no-unused-vars
        using scope = continuationWindow.withScope(context);
        // TODO: Is there a way to have asyncEnd _after_ the continuation?
        return result;
      }
  
      // eslint-disable-next-line no-unused-vars
      using scope = this.#callWindow.withScope(context);
      try {
        const result = ReflectApply(fn, thisArg, args);
        // If the return value is not a thenable, return it directly with a warning.
        // Do not publish to asyncStart/asyncEnd.
        if (typeof result?.then !== 'function') {
          emitNonThenableWarning(fn);
          context.result = result;
          return result;
        }
        // For native Promises use PromisePrototypeThen to avoid user overrides.
        if (isPromise(result)) {
          return PromisePrototypeThen(result, resolve, reject);
        }
        // For custom thenables, call .then() directly to preserve the thenable type.
        return result.then(resolve, reject);
      } catch (err) {
        context.error = err;
        error.publish(context);
        throw err;
      }
    }
  
    traceCallback(fn, position = -1, context = {}, thisArg, ...args) {
      if (!this.hasSubscribers) {
        return ReflectApply(fn, thisArg, args);
      }
  
      const { error } = this;
      const continuationWindow = this.#continuationWindow;
  
      function wrappedCallback(err, res) {
        if (err) {
          context.error = err;
          error.publish(context);
        } else {
          context.result = res;
        }
  
        // Use continuation window for asyncStart/asyncEnd around callback
        // eslint-disable-next-line no-unused-vars
        using scope = continuationWindow.withScope(context);
        return ReflectApply(callback, this, arguments);
      }
  
      const callback = ArrayPrototypeAt(args, position);
      validateFunction(callback, 'callback');
      ArrayPrototypeSplice(args, position, 1, wrappedCallback);
  
      // eslint-disable-next-line no-unused-vars
      using scope = this.#callWindow.withScope(context);
      try {
        return ReflectApply(fn, thisArg, args);
      } catch (err) {
        context.error = err;
        error.publish(context);
        throw err;
      }
    }
  }
  
  function tracingChannel(nameOrChannels) {
    return new TracingChannel(nameOrChannels);
  }
  
  dc_binding.linkNativeChannel((name) => channel(name));
  
  module.exports = {
    channel,
    hasSubscribers,
    subscribe,
    tracingChannel,
    unsubscribe,
    boundedChannel,
    Channel,
    BoundedChannel,
  };
  