import { MuxerClosedError, TypedEventEmitter } from '@libp2p/interface';
import { raceSignal } from 'race-signal';
import { MaxEarlyStreamsError } from "./errors.js";
import { isPromise } from "./is-promise.js";
export class AbstractStreamMuxer extends TypedEventEmitter {
    streams;
    protocol;
    status;
    log;
    maConn;
    streamOptions;
    earlyStreams;
    maxEarlyStreams;
    metrics;
    constructor(maConn, init) {
        super();
        this.maConn = maConn;
        this.protocol = init.protocol;
        this.streams = [];
        this.earlyStreams = [];
        this.status = 'open';
        this.log = maConn.log.newScope(init.name);
        this.streamOptions = init.streamOptions;
        this.maxEarlyStreams = init.maxEarlyStreams ?? 10;
        this.metrics = init.metrics;
        // read/write all data from/to underlying maConn
        const muxerMaConnOnMessage = (evt) => {
            try {
                this.onData(evt.data);
            }
            catch (err) {
                this.abort(err);
                this.maConn.abort(err);
            }
        };
        this.maConn.addEventListener('message', muxerMaConnOnMessage);
        // signal stream writers when the underlying connection can accept more data
        const muxerMaConnOnDrain = () => {
            this.log('underlying stream drained, signal %d streams to continue writing', this.streams.length);
            this.streams.forEach(stream => {
                stream.onMuxerDrain();
            });
        };
        this.maConn.addEventListener('drain', muxerMaConnOnDrain);
        const muxerOnMaConnClose = () => {
            this.log('underlying stream closed with status %s and %d streams', this.status, this.streams.length);
            this.onTransportClosed();
        };
        this.maConn.addEventListener('close', muxerOnMaConnClose);
    }
    send(data) {
        const result = this.maConn.send(data);
        if (result === false) {
            this.log('underlying stream saturated, signal %d streams to pause writing', this.streams.length);
            this.streams.forEach(stream => {
                stream.onMuxerNeedsDrain();
            });
        }
        return result;
    }
    async close(options) {
        if (this.status === 'closed' || this.status === 'closing') {
            return;
        }
        this.status = 'closing';
        await raceSignal(Promise.all([...this.streams].map(async (s) => {
            await s.close(options);
        })), options?.signal);
        this.status = 'closed';
    }
    abort(err) {
        if (this.status === 'closed') {
            return;
        }
        this.status = 'closing';
        [...this.streams].forEach(s => {
            s.abort(err);
        });
        this.status = 'closed';
    }
    onTransportClosed(err) {
        this.status = 'closing';
        try {
            [...this.streams].forEach(stream => {
                stream.onTransportClosed(err);
            });
        }
        catch (err) {
            this.abort(err);
        }
        this.status = 'closed';
    }
    async createStream(options) {
        if (this.status !== 'open') {
            throw new MuxerClosedError();
        }
        let stream = this.onCreateStream({
            ...this.streamOptions,
            ...options
        });
        if (isPromise(stream)) {
            stream = await stream;
        }
        this.streams.push(stream);
        this.cleanUpStream(stream);
        return stream;
    }
    /**
     * Extending classes should invoke this method when a new stream was created
     * by the remote muxer
     */
    onRemoteStream(stream) {
        this.streams.push(stream);
        this.cleanUpStream(stream);
        if (this.listenerCount('stream') === 0) {
            // no listener has been added for the stream event yet, store the stream
            // to emit it later
            this.earlyStreams.push(stream);
            if (this.earlyStreams.length > this.maxEarlyStreams) {
                this.abort(new MaxEarlyStreamsError(`Too many early streams were opened - ${this.earlyStreams.length}/${this.maxEarlyStreams}`));
            }
            return;
        }
        this.safeDispatchEvent('stream', {
            detail: stream
        });
    }
    cleanUpStream(stream) {
        const muxerOnStreamEnd = (evt) => {
            const index = this.streams.findIndex(s => s === stream);
            if (index !== -1) {
                this.streams.splice(index, 1);
            }
            if (evt.error != null) {
                if (evt.local) {
                    this.metrics?.increment({ [`${stream.direction}_stream_reset`]: true });
                }
                else {
                    this.metrics?.increment({ [`${stream.direction}_stream_abort`]: true });
                }
            }
            else {
                this.metrics?.increment({ [`${stream.direction}_stream_end`]: true });
            }
        };
        stream.addEventListener('close', muxerOnStreamEnd);
        this.metrics?.increment({ [`${stream.direction}_stream`]: true });
    }
    addEventListener(...args) {
        // @ts-expect-error cannot ensure args has enough members
        super.addEventListener.apply(this, args);
        // if a 'stream' listener is being added and we have early streams, emit
        // them
        if (args[0] === 'stream' && this.earlyStreams.length > 0) {
            // event listeners can be added in constructors and often use object
            // properties - if this the case we can access a class member before it
            // has been initialized so dispatch the message in the microtask queue
            queueMicrotask(() => {
                this.earlyStreams.forEach(stream => {
                    this.safeDispatchEvent('stream', {
                        detail: stream
                    });
                });
                this.earlyStreams = [];
            });
        }
    }
}
//# sourceMappingURL=abstract-stream-muxer.js.map