import { StreamMessageEvent } from '@libp2p/interface';
import delay from 'delay';
import { TypedEventEmitter } from 'main-event';
import { raceSignal } from 'race-signal';
import { isUint8ArrayList, Uint8ArrayList } from 'uint8arraylist';
import { Queue } from "./queue/index.js";
const DEFAULT_CHUNK_SIZE = 1024 * 64;
/**
 * Accepts events to emit after a short delay, and with a configurable maximum
 * queue capacity after which the send method will return false to let us
 * simulate write backpressure.
 */
export class MessageQueue extends TypedEventEmitter {
    needsDrain;
    queue;
    capacity;
    delay;
    log;
    chunkSize;
    constructor(init) {
        super();
        this.needsDrain = false;
        this.queue = new Queue({
            concurrency: 1
        });
        this.capacity = init.capacity ?? 5;
        this.delay = init.delay ?? 0;
        this.log = init.log;
        this.chunkSize = init.chunkSize ?? DEFAULT_CHUNK_SIZE;
        this.queue.addEventListener('idle', () => {
            if (this.needsDrain) {
                this.log('network send queue drained');
                this.needsDrain = false;
                this.safeDispatchEvent('drain');
            }
            else {
                this.log('network send queue idle');
            }
        });
    }
    send(evt) {
        if (isMessageEvent(evt)) {
            // chunk outgoing messages to match TCP packet sizes
            const data = new Uint8ArrayList(evt.data);
            while (data.byteLength > 0) {
                const end = Math.min(this.chunkSize, data.byteLength);
                const chunk = data.sublist(0, end);
                data.consume(chunk.byteLength);
                const chunkEvent = new StreamMessageEvent(chunk);
                this.queue.add(async (opts) => {
                    if (this.delay > 0) {
                        await raceSignal(delay(this.delay), opts.signal);
                    }
                    this.dispatchEvent(opts.evt);
                }, {
                    evt: chunkEvent
                });
            }
        }
        else {
            this.queue.add(async (opts) => {
                if (this.delay > 0) {
                    await raceSignal(delay(this.delay), opts.signal);
                }
                this.dispatchEvent(opts.evt);
            }, {
                evt
            });
        }
        if (this.queue.size >= this.capacity) {
            this.log('network send queue full');
            this.needsDrain = true;
            return false;
        }
        return true;
    }
    pause() {
        this.queue.pause();
    }
    resume() {
        this.queue.resume();
    }
    onIdle() {
        return this.queue.onIdle();
    }
    size() {
        return this.queue.size;
    }
}
function isMessageEvent(evt) {
    return evt?.data instanceof Uint8Array || isUint8ArrayList(evt?.data);
}
//# sourceMappingURL=message-queue.js.map