Class OTelChannelSubscribers

packages/client/lib/opentelemetry/metrics.ts:195–491

// ---------------------------------------------------------------------------

class OTelChannelSubscribers {
  readonly #instruments: MetricInstruments;
  readonly #options: MetricOptions;
  readonly #unsubscribers: Array<() => void> = [];

  constructor(
    options: MetricOptions,
    instruments: MetricInstruments,
    enabledGroups: MetricGroup[],
  ) {
    this.#options = options;
    this.#instruments = instruments;

    const hasBasic = enabledGroups.includes(METRIC_GROUP.CONNECTION_BASIC);
    const hasAdvanced = enabledGroups.includes(METRIC_GROUP.CONNECTION_ADVANCED);
    if (hasBasic) {
      this.#subscribeConnectionBasic();
    }
    if (hasAdvanced) {
      this.#subscribeConnectionAdvanced();
    }
    if (hasBasic || hasAdvanced) {
      this.#subscribeConnectionClosed(hasBasic, hasAdvanced);
    }
    if (enabledGroups.includes(METRIC_GROUP.RESILIENCY)) {
      this.#subscribeResiliency();
    }
    if (enabledGroups.includes(METRIC_GROUP.CLIENT_SIDE_CACHING)) {
      this.#subscribeClientSideCache();
    }
    if (enabledGroups.includes(METRIC_GROUP.PUBSUB)) {
      this.#subscribePubSub();
    }
    if (enabledGroups.includes(METRIC_GROUP.PUBSUB) || enabledGroups.includes(METRIC_GROUP.STREAMING)) {
      this.#subscribeCommandReply(enabledGroups);
    }
  }

  #sub(name: string, handler: (ctx: any) => void) {
    const ch = getChannel(name);
    if (!ch) return;
    ch.subscribe(handler);
    this.#unsubscribers.push(() => ch.unsubscribe(handler));
  }

  destroy() {
    this.#unsubscribers.forEach(fn => fn());
  }

  // -- Connection Basic --

  #subscribeConnectionBasic() {
    this.#sub(CHANNELS.CONNECTION_READY, (ctx: any) => {
      const clientAttributes = resolveClientAttributes(ctx.clientId);
      this.#instruments.dbClientConnectionCreateTime.record(
        ctx.createTimeMs / 1000,
        {
          ...this.#options.attributes,