import { AsyncIterableX } from '../asynciterablex.js'; import { OperatorAsyncFunction } from '../../interfaces.js'; import { wrapWithAbort } from './withabort.js'; import { throwIfAborted } from '../../aborterror.js'; /** @ignore */ export interface Timestamp { time: number; value: TSource; } /** @ignore */ export class TimestampAsyncIterable extends AsyncIterableX> { private _source: AsyncIterable; constructor(source: AsyncIterable) { super(); this._source = source; } async *[Symbol.asyncIterator](signal?: AbortSignal) { throwIfAborted(signal); for await (const item of wrapWithAbort(this._source, signal)) { yield { time: Date.now(), value: item }; } } } /** * Timestamps each element in an async-iterable sequence using the local system clock. * * @template TSource The type of the elements in the source sequence. * @returns {OperatorAsyncFunction>} An async-iterable sequence with timestamp information on elements. */ export function timestamp(): OperatorAsyncFunction> { return function timestampOperatorFunction( source: AsyncIterable ): AsyncIterableX> { return new TimestampAsyncIterable(source); }; }