'use strict';

// New Streams API - Share
//
// Pull-model multi-consumer streaming. Shares a single source among
// multiple consumers with explicit buffering.

const {
  ArrayPrototypePush,
  FunctionPrototypeCall,
  ObjectSetPrototypeOf,
  PromisePrototypeThen,
  PromiseResolve,
  PromiseWithResolvers,
  SafePromiseRace,
  SafeSet,
  Symbol,
  SymbolAsyncIterator,
  SymbolDispose,
  SymbolIterator,
} = primordials;

const {
  shareProtocol,
  shareSyncProtocol,
} = require('internal/streams/iter/types');

const {
  from,
  fromSync,
  isAsyncIterable,
  isSyncIterable,
} = require('internal/streams/iter/from');

const {
  pullSync: pullSyncWithTransforms,
  pullWithConsumerCleanup,
} = require('internal/streams/iter/pull');

const {
  IterResult,
  kMultiConsumerDefaultBudget,
  createBatchEntry,
  getProtocolMethod,
  getMinCursor,
  onSignalAbort,
  parsePullArgs,
  splitBatchEntry,
  validateBatchEntry,
  validateBudget,
} = require('internal/streams/iter/utils');
const {
  converters,
} = require('internal/streams/iter/webidl');

const {
  RingBuffer,
} = require('internal/streams/iter/ringbuffer');

const {
  codes: {
    ERR_INVALID_ARG_TYPE,
    ERR_INVALID_ARG_VALUE,
    ERR_INVALID_RETURN_VALUE,
    ERR_INVALID_STATE,
    ERR_OUT_OF_RANGE,
  },
} = require('internal/errors');

const { markPromiseAsHandled } = internalBinding('util');

// =============================================================================
// Async Share Implementation
// =============================================================================

const kNoShareError = Symbol('kNoShareError');
const kShareCancelled = Symbol('kShareCancelled');
const kSetFactorySignal = Symbol('kSetFactorySignal');

class ShareImpl {
  #source;
  #options;
  #buffer = new RingBuffer();
  #bufferStart = 0;
  #consumers = new SafeSet();
  #sourceIterator = null;
  #sourceExhausted = false;
  #sourceError = kNoShareError;
  #cancelled = false;
  #pulling = false;
  #pullWaiters = [];
  #cancelPromise;
  #resolveCancel;
  #cancelError = kNoShareError;
  #cachedMinCursor = 0;
  #cachedMinCursorConsumers = 0;
  #abortHandler;
  /** Cumulative byte size of buffered entries */
  #bufferedBytes = 0;

  constructor(source, options) {
    this.#source = source;
    this.#options = options;
    const { promise, resolve } = PromiseWithResolvers();
    this.#cancelPromise = promise;
    this.#resolveCancel = resolve;
  }

  get consumerCount() {
    return this.#consumers.size;
  }

  [kSetFactorySignal](signal) {
    this.#abortHandler = () => this.cancel(signal.reason);
    onSignalAbort(signal, this.#abortHandler);
  }

  pull(...args) {
    const parsed = parsePullArgs(args);
    const { transforms } = parsed;
    const options = converters.PullOptions(parsed.options, {
      __proto__: null,
      context: 'options',
    });
    const { signal } = options;

    // Avoid registering a consumer that the pre-aborted pipeline will never
    // read or detach.
    if (signal?.aborted) {
      return {
        __proto__: null,
        // eslint-disable-next-line require-yield
        async *[SymbolAsyncIterator]() {
          throw signal.reason;
        },
      };
    }

    const rawConsumer = this.#createRawConsumer();

    if (transforms.length > 0 || signal) {
      return pullWithConsumerCleanup(rawConsumer, transforms, signal);
    }
    return rawConsumer;
  }

  #createRawConsumer() {
    const state = ObjectSetPrototypeOf({
      cursor: this.#bufferStart,
      resolve: null,
      reject: null,
      detached: false,
      error: kNoShareError,
      pendingNext: PromiseResolve(),
    }, null);

    this.#consumers.add(state);
    if (this.#consumers.size === 1) {
      this.#cachedMinCursor = state.cursor;
      this.#cachedMinCursorConsumers = 1;
    } else if (state.cursor === this.#cachedMinCursor) {
      this.#cachedMinCursorConsumers++;
    } else {
      this.#recomputeMinCursor();
    }
    const self = this;

    return {
      __proto__: null,
      [SymbolAsyncIterator]() {
        const getNext = async () => {
          // Loop until we get data, source is exhausted, or
          // consumer is detached. Multiple consumers may be woken
          // after a single pull - those that find no data at their
          // cursor must re-pull rather than terminating prematurely.
          for (;;) {
            if (state.detached) {
              if (state.error !== kNoShareError) throw state.error;
              return new IterResult(true, undefined);
            }

            if (self.#cancelled) {
              state.detached = true;
              state.error = self.#cancelError;
              self.#deleteConsumer(state);
              if (state.error !== kNoShareError) throw state.error;
              return new IterResult(true, undefined);
            }

            // Check if data is available in buffer
            const bufferIndex = state.cursor - self.#bufferStart;
            if (bufferIndex < self.#buffer.length) {
              const chunk = self.#readEntry(self.#buffer.get(bufferIndex));
              const cursor = state.cursor;
              state.cursor++;
              if (cursor === self.#cachedMinCursor &&
                  --self.#cachedMinCursorConsumers === 0) {
                self.#tryTrimBuffer();
              }
              return new IterResult(false, chunk);
            }

            if (self.#sourceExhausted) {
              state.detached = true;
              self.#deleteConsumer(state);
              if (self.#sourceError !== kNoShareError) {
                state.error = self.#sourceError;
                throw state.error;
              }
              return new IterResult(true, undefined);
            }

            // Need to pull from source - check buffer limit
            let shouldBuffer;
            try {
              shouldBuffer = await self.#waitForBufferSpace();
            } catch (error) {
              state.detached = true;
              if (self.#deleteConsumer(state)) {
                self.#tryTrimBuffer();
              }
              throw error;
            }
            if (shouldBuffer === null) {
              state.detached = true;
              state.error = self.#cancelError;
              self.#deleteConsumer(state);
              if (state.error !== kNoShareError) throw state.error;
              return new IterResult(true, undefined);
            }

            await self.#pullFromSource(!shouldBuffer);
            if (!shouldBuffer) {
              await self.#waitForBufferSpaceAfterDrop();
            }
          }
        };

        return ObjectSetPrototypeOf({
          next() {
            const next = PromisePrototypeThen(
              state.pendingNext,
              getNext,
              getNext);
            state.pendingNext = next;
            markPromiseAsHandled(state.pendingNext);
            return next;
          },

          async return() {
            state.detached = true;
            state.resolve = null;
            state.reject = null;
            if (self.#deleteConsumer(state)) {
              self.#tryTrimBuffer();
            }
            return new IterResult(true, undefined);
          },

          async throw() {
            state.detached = true;
            state.resolve = null;
            state.reject = null;
            if (self.#deleteConsumer(state)) {
              self.#tryTrimBuffer();
            }
            return new IterResult(true, undefined);
          },
        }, null);
      },
    };
  }

  cancel(reason) {
    if (this.#cancelled) return;
    const hasReason = arguments.length > 0;
    this.#cancelled = true;

    if (hasReason) {
      this.#cancelError = reason;
    }

    this.#resolveCancel(kShareCancelled);
    this.#resolveCancel = null;

    try {
      const returnMethod = this.#sourceIterator?.return;
      if (typeof returnMethod === 'function') {
        const cleanup = PromisePrototypeThen(
          PromiseResolve(FunctionPrototypeCall(
            returnMethod, this.#sourceIterator)),
          undefined,
          undefined);
        markPromiseAsHandled(cleanup);
      }
    } catch {
      // Cancellation has precedence over source cleanup errors.
    }

    for (const consumer of this.#consumers) {
      consumer.error = this.#cancelError;
      if (consumer.resolve) {
        if (hasReason) {
          consumer.reject?.(reason);
        } else {
          consumer.resolve(new IterResult(true, undefined));
        }
        consumer.resolve = null;
        consumer.reject = null;
      }
      consumer.detached = true;
    }
    this.#consumers.clear();
    this.#buffer.clear();
    this.#bufferedBytes = 0;
    this.#cleanupFactorySignal();

    for (let i = 0; i < this.#pullWaiters.length; i++) {
      this.#pullWaiters[i]();
    }
    this.#pullWaiters = [];
  }

  [SymbolDispose]() {
    this.cancel();
  }

  // Internal methods

  async #waitForBufferSpace() {
    while (this.#bufferedBytes >= this.#options.budget) {
      if (this.#cancelled ||
          this.#sourceError !== kNoShareError ||
          this.#sourceExhausted) {
        return this.#cancelled ? null : true;
      }

      switch (this.#options.backpressure) {
        case 'strict':
          throw new ERR_OUT_OF_RANGE(
            'buffered bytes', `< ${this.#options.budget}`,
            this.#bufferedBytes);
        case 'unbounded': {
          const { promise, resolve } = PromiseWithResolvers();
          ArrayPrototypePush(this.#pullWaiters, resolve);
          await promise;
          break;
        }
        case 'drop-oldest':
          while (this.#bufferedBytes >= this.#options.budget &&
                 this.#buffer.length > 0) {
            const evicted = this.#buffer.shift();
            this.#bufferedBytes -= evicted.byteLength;
            this.#bufferStart++;
          }
          for (const consumer of this.#consumers) {
            if (consumer.cursor < this.#bufferStart) {
              this.#deleteConsumerFromMin(consumer);
              consumer.cursor = this.#bufferStart;
            }
          }
          this.#recomputeMinCursor();
          return true;
        case 'drop-newest':
          return false;
      }
    }
    return true;
  }

  async #waitForBufferSpaceAfterDrop() {
    while (this.#bufferedBytes >= this.#options.budget &&
           !this.#cancelled &&
           this.#sourceError === kNoShareError &&
           !this.#sourceExhausted) {
      const { promise, resolve } = PromiseWithResolvers();
      ArrayPrototypePush(this.#pullWaiters, resolve);
      await promise;
    }
  }

  #pullFromSource(discard = false) {
    if (this.#sourceExhausted || this.#cancelled) {
      return PromiseResolve();
    }

    if (this.#pulling) {
      const { promise, resolve } = PromiseWithResolvers();
      ArrayPrototypePush(this.#pullWaiters, resolve);
      return promise;
    }

    this.#pulling = true;

    return (async () => {
      try {
        if (!this.#sourceIterator) {
          if (isAsyncIterable(this.#source)) {
            this.#sourceIterator =
              this.#source[SymbolAsyncIterator]();
          } else if (isSyncIterable(this.#source)) {
            const syncIterator =
              this.#source[SymbolIterator]();
            this.#sourceIterator = ObjectSetPrototypeOf({
              async next() {
                return syncIterator.next();
              },
              async return() {
                return syncIterator.return?.() ??
                       new IterResult(true, undefined);
              },
            }, null);
          } else {
            throw new ERR_INVALID_ARG_TYPE(
              'source', ['AsyncIterable', 'Iterable'], this.#source);
          }
        }

        const result = await SafePromiseRace([
          this.#sourceIterator.next(),
          this.#cancelPromise,
        ]);

        if (this.#cancelled || result === kShareCancelled) return;

        if (result.done) {
          this.#sourceExhausted = true;
        } else if (!discard) {
          this.#bufferBatch(result.value);
        }
      } catch (error) {
        this.#sourceError = error;
        this.#sourceExhausted = true;
      } finally {
        if (this.#sourceExhausted && this.#consumers.size === 0) {
          this.#cleanupFactorySignal();
        }
        this.#pulling = false;
        for (let i = 0; i < this.#pullWaiters.length; i++) {
          this.#pullWaiters[i]();
        }
        this.#pullWaiters = [];
      }
    })();
  }

  #bufferBatch(batch) {
    const entry = createBatchEntry(batch);
    // 'drop-oldest' evicts whole entries. A single pulled batch can be much
    // larger than the budget (for example when from() combines many values
    // of a sync source), and evicting it would discard every chunk in it,
    // including ones that no consumer has read. Split such batches so that
    // eviction keeps the newest chunks that fit the budget.
    const entries = this.#options.backpressure === 'drop-oldest' ?
      splitBatchEntry(entry, this.#options.budget) : undefined;
    if (entries === undefined) {
      this.#buffer.push(entry);
    } else {
      for (let i = 0; i < entries.length; i++) {
        this.#buffer.push(entries[i]);
      }
    }
    this.#bufferedBytes += entry.byteLength;
  }

  #tryTrimBuffer() {
    // Retain buffered data for consumers that attach while none are active.
    // Without this, the last consumer detaching would discard entries that
    // a late-joining consumer has not read and cannot get back from the
    // source.
    if (this.#consumers.size === 0) return;
    if (this.#cachedMinCursorConsumers === 0) {
      this.#recomputeMinCursor();
    }
    const trimCount = this.#cachedMinCursor - this.#bufferStart;
    if (trimCount > 0) {
      for (let i = 0; i < trimCount; i++) {
        const evicted = this.#buffer.get(i);
        this.#bufferedBytes -= evicted.byteLength;
      }
      this.#buffer.trimFront(trimCount);
      this.#bufferStart = this.#cachedMinCursor;
      for (let i = 0; i < this.#pullWaiters.length; i++) {
        this.#pullWaiters[i]();
      }
      this.#pullWaiters = [];
    }
  }

  #readEntry(entry) {
    try {
      return validateBatchEntry(entry);
    } catch (error) {
      this.cancel(error);
      this.#buffer.clear();
      this.#bufferedBytes = 0;
      throw error;
    }
  }

  #recomputeMinCursor() {
    const { minCursor, minCursorConsumers } = getMinCursor(
      this.#consumers, this.#bufferStart + this.#buffer.length);
    this.#cachedMinCursor = minCursor;
    this.#cachedMinCursorConsumers = minCursorConsumers;
  }

  #cleanupFactorySignal() {
    if (this.#abortHandler !== undefined) {
      this.#options.signal.removeEventListener('abort', this.#abortHandler);
      this.#abortHandler = undefined;
    }
  }

  #deleteConsumerFromMin(consumer) {
    if (consumer.cursor === this.#cachedMinCursor) {
      this.#cachedMinCursorConsumers--;
      return this.#cachedMinCursorConsumers === 0;
    }
    return false;
  }

  #deleteConsumer(consumer) {
    if (this.#consumers.delete(consumer)) {
      const shouldTrimBuffer = this.#deleteConsumerFromMin(consumer);
      if (this.#sourceExhausted && this.#consumers.size === 0) {
        this.#cleanupFactorySignal();
      }
      return shouldTrimBuffer;
    }
    return false;
  }
}

// =============================================================================
// Sync Share Implementation
// =============================================================================

class SyncShareImpl {
  #source;
  #options;
  #buffer = new RingBuffer();
  #bufferStart = 0;
  #consumers = new SafeSet();
  #sourceIterator = null;
  #sourceExhausted = false;
  #sourceError = kNoShareError;
  #cancelled = false;
  #pulling = false;
  #cachedMinCursor = 0;
  #cachedMinCursorConsumers = 0;
  /** Cumulative byte size of buffered entries */
  #bufferedBytes = 0;

  constructor(source, options) {
    this.#source = source;
    this.#options = options;
  }

  get consumerCount() {
    return this.#consumers.size;
  }

  pull(...transforms) {
    const rawConsumer = this.#createRawConsumer();

    if (transforms.length > 0) {
      return pullSyncWithTransforms(rawConsumer, ...transforms);
    }
    return rawConsumer;
  }

  #createRawConsumer() {
    const state = ObjectSetPrototypeOf({
      cursor: this.#bufferStart,
      detached: false,
      error: kNoShareError,
    }, null);

    this.#consumers.add(state);
    if (this.#consumers.size === 1) {
      this.#cachedMinCursor = state.cursor;
      this.#cachedMinCursorConsumers = 1;
    } else if (state.cursor === this.#cachedMinCursor) {
      this.#cachedMinCursorConsumers++;
    } else {
      this.#recomputeMinCursor();
    }
    const self = this;

    return {
      __proto__: null,
      [SymbolIterator]() {
        return ObjectSetPrototypeOf({
          next() {
            if (state.detached) {
              if (state.error !== kNoShareError) throw state.error;
              return new IterResult(true, undefined);
            }
            if (self.#sourceError !== kNoShareError) {
              state.detached = true;
              state.error = self.#sourceError;
              self.#deleteConsumer(state);
              throw state.error;
            }
            if (self.#cancelled) {
              state.detached = true;
              self.#deleteConsumer(state);
              return new IterResult(true, undefined);
            }

            const bufferIndex = state.cursor - self.#bufferStart;
            if (bufferIndex < self.#buffer.length) {
              const chunk = self.#readEntry(self.#buffer.get(bufferIndex));
              const cursor = state.cursor;
              state.cursor++;
              if (cursor === self.#cachedMinCursor &&
                  --self.#cachedMinCursorConsumers === 0) {
                self.#tryTrimBuffer();
              }
              return new IterResult(false, chunk);
            }

            if (self.#sourceExhausted) {
              state.detached = true;
              self.#deleteConsumer(state);
              return new IterResult(true, undefined);
            }

            // Check buffer limit. 'unbounded' and 'drop-newest' are rejected
            // by shareSync().
            if (self.#bufferedBytes >= self.#options.budget) {
              switch (self.#options.backpressure) {
                case 'strict': {
                  const error = new ERR_OUT_OF_RANGE(
                    'buffered bytes', `< ${self.#options.budget}`,
                    self.#bufferedBytes);
                  // Detach before throwing, as the async share does. Neither
                  // for...of nor a transform pipeline calls return() when
                  // next() throws, so a consumer left registered here would
                  // keep its cursor forever and wedge the other consumers.
                  state.detached = true;
                  if (self.#deleteConsumer(state)) {
                    self.#tryTrimBuffer();
                  }
                  throw error;
                }
                case 'drop-oldest':
                  while (self.#bufferedBytes >= self.#options.budget &&
                         self.#buffer.length > 0) {
                    const evicted = self.#buffer.shift();
                    self.#bufferedBytes -= evicted.byteLength;
                    self.#bufferStart++;
                  }
                  for (const consumer of self.#consumers) {
                    if (consumer.cursor < self.#bufferStart) {
                      self.#deleteConsumerFromMin(consumer);
                      consumer.cursor = self.#bufferStart;
                    }
                  }
                  self.#recomputeMinCursor();
                  break;
              }
            }

            self.#pullFromSource();

            if (self.#sourceError !== kNoShareError) {
              state.detached = true;
              state.error = self.#sourceError;
              self.#deleteConsumer(state);
              throw state.error;
            }

            const newBufferIndex = state.cursor - self.#bufferStart;
            if (newBufferIndex < self.#buffer.length) {
              const chunk = self.#readEntry(self.#buffer.get(newBufferIndex));
              const cursor = state.cursor;
              state.cursor++;
              if (cursor === self.#cachedMinCursor &&
                  --self.#cachedMinCursorConsumers === 0) {
                self.#tryTrimBuffer();
              }
              return new IterResult(false, chunk);
            }

            if (self.#sourceExhausted) {
              state.detached = true;
              self.#deleteConsumer(state);
              return new IterResult(true, undefined);
            }

            return new IterResult(true, undefined);
          },

          return() {
            state.detached = true;
            if (self.#deleteConsumer(state)) {
              self.#tryTrimBuffer();
            }
            return new IterResult(true, undefined);
          },

          throw() {
            state.detached = true;
            if (self.#deleteConsumer(state)) {
              self.#tryTrimBuffer();
            }
            return new IterResult(true, undefined);
          },
        }, null);
      },
    };
  }

  cancel(reason) {
    if (this.#cancelled) return;
    const hasReason = arguments.length > 0;
    this.#cancelled = true;

    if (hasReason) {
      this.#sourceError = reason;
    }

    try {
      const returnMethod = this.#sourceIterator?.return;
      if (typeof returnMethod === 'function') {
        FunctionPrototypeCall(returnMethod, this.#sourceIterator);
      }
    } catch {
      // Cancellation has precedence over source cleanup errors.
    }

    for (const consumer of this.#consumers) {
      if (hasReason) consumer.error = reason;
      consumer.detached = true;
    }
    this.#consumers.clear();
  }

  [SymbolDispose]() {
    this.cancel();
  }

  #pullFromSource() {
    if (this.#sourceExhausted || this.#cancelled) return;

    // A source that reads from its own share would re-enter its own
    // iterator. Fail that read without touching the share's state; the
    // source sees the error and may handle it.
    if (this.#pulling) {
      throw new ERR_INVALID_STATE(
        'shareSync() source cannot be read while it is producing a value');
    }

    this.#pulling = true;
    try {
      this.#sourceIterator ||= this.#source[SymbolIterator]();

      const result = this.#sourceIterator.next();

      if (result.done) {
        this.#sourceExhausted = true;
      } else {
        this.#bufferBatch(result.value);
      }
    } catch (error) {
      this.#sourceError = error;
      this.#sourceExhausted = true;
    } finally {
      this.#pulling = false;
    }
  }

  #bufferBatch(batch) {
    const entry = createBatchEntry(batch);
    // 'drop-oldest' evicts whole entries. A single pulled batch can be much
    // larger than the budget (for example when from() combines many values
    // of a sync source), and evicting it would discard every chunk in it,
    // including ones that no consumer has read. Split such batches so that
    // eviction keeps the newest chunks that fit the budget.
    const entries = this.#options.backpressure === 'drop-oldest' ?
      splitBatchEntry(entry, this.#options.budget) : undefined;
    if (entries === undefined) {
      this.#buffer.push(entry);
    } else {
      for (let i = 0; i < entries.length; i++) {
        this.#buffer.push(entries[i]);
      }
    }
    this.#bufferedBytes += entry.byteLength;
  }

  #tryTrimBuffer() {
    // Retain buffered data for consumers that attach while none are active.
    // Without this, the last consumer detaching would discard entries that
    // a late-joining consumer has not read and cannot get back from the
    // source.
    if (this.#consumers.size === 0) return;
    if (this.#cachedMinCursorConsumers === 0) {
      this.#recomputeMinCursor();
    }
    const trimCount = this.#cachedMinCursor - this.#bufferStart;
    if (trimCount > 0) {
      for (let i = 0; i < trimCount; i++) {
        const evicted = this.#buffer.get(i);
        this.#bufferedBytes -= evicted.byteLength;
      }
      this.#buffer.trimFront(trimCount);
      this.#bufferStart = this.#cachedMinCursor;
    }
  }

  #readEntry(entry) {
    try {
      return validateBatchEntry(entry);
    } catch (error) {
      this.cancel(error);
      this.#buffer.clear();
      this.#bufferedBytes = 0;
      throw error;
    }
  }

  #recomputeMinCursor() {
    const { minCursor, minCursorConsumers } = getMinCursor(
      this.#consumers, this.#bufferStart + this.#buffer.length);
    this.#cachedMinCursor = minCursor;
    this.#cachedMinCursorConsumers = minCursorConsumers;
  }

  #deleteConsumerFromMin(consumer) {
    if (consumer.cursor === this.#cachedMinCursor) {
      this.#cachedMinCursorConsumers--;
      return this.#cachedMinCursorConsumers === 0;
    }
    return false;
  }

  #deleteConsumer(consumer) {
    if (this.#consumers.delete(consumer)) {
      return this.#deleteConsumerFromMin(consumer);
    }
    return false;
  }
}

// =============================================================================
// Public API
// =============================================================================

function share(source, options = { __proto__: null }) {
  // Normalize source via from() - accepts strings, ArrayBuffers, protocols, etc.
  const normalized = from(source);
  options = converters.ShareOptions(options, {
    __proto__: null,
    context: 'options',
  });
  const {
    budget = kMultiConsumerDefaultBudget,
    backpressure = 'strict',
    signal,
  } = options;
  validateBudget(budget);

  const opts = {
    __proto__: null,
    budget,
    backpressure,
    signal,
  };

  const shareImpl = new ShareImpl(normalized, opts);

  if (signal) {
    shareImpl[kSetFactorySignal](signal);
  }

  return shareImpl;
}

function shareSync(source, options = { __proto__: null }) {
  // Normalize source via fromSync() - accepts strings, ArrayBuffers, protocols, etc.
  const normalized = fromSync(source);
  options = converters.ShareSyncOptions(options, {
    __proto__: null,
    context: 'options',
  });
  const {
    budget = kMultiConsumerDefaultBudget,
    backpressure = 'strict',
  } = options;
  validateBudget(budget);
  // A synchronous consumer can neither wait for the slowest consumer to
  // release budget ('unbounded') nor keep pulling and discarding until it
  // does ('drop-newest'): the slowest consumer cannot advance while the
  // call is running.
  if (backpressure === 'unbounded' || backpressure === 'drop-newest') {
    throw new ERR_INVALID_ARG_VALUE(
      'options.backpressure', backpressure,
      `${backpressure} is not supported by shareSync()`);
  }

  const opts = {
    __proto__: null,
    budget,
    backpressure,
  };

  return new SyncShareImpl(normalized, opts);
}

const Share = {
  __proto__: null,
  from(input, options) {
    const protocol = getProtocolMethod(input, shareProtocol);
    if (protocol !== undefined) {
      const result = FunctionPrototypeCall(protocol, input, options);
      if (result === null || typeof result !== 'object') {
        throw new ERR_INVALID_RETURN_VALUE(
          'an object', '[Symbol.for(\'Stream.shareProtocol\')]', result);
      }
      return result;
    }
    if (isAsyncIterable(input) || isSyncIterable(input)) {
      return share(input, options);
    }
    throw new ERR_INVALID_ARG_TYPE(
      'input', ['Shareable', 'AsyncIterable', 'Iterable'], input);
  },
};

const SyncShare = {
  __proto__: null,
  fromSync(input, options) {
    const protocol = getProtocolMethod(input, shareSyncProtocol);
    if (protocol !== undefined) {
      const result = FunctionPrototypeCall(protocol, input, options);
      if (result === null || typeof result !== 'object') {
        throw new ERR_INVALID_RETURN_VALUE(
          'an object', '[Symbol.for(\'Stream.shareSyncProtocol\')]', result);
      }
      return result;
    }
    if (isSyncIterable(input)) {
      return shareSync(input, options);
    }
    throw new ERR_INVALID_ARG_TYPE(
      'input', ['SyncShareable', 'Iterable'], input);
  },
};

module.exports = {
  Share,
  SyncShare,
  share,
  shareSync,
};