kannon/classes/AudioServer.js

const sleep = require('../libs/util.js').sleep;

const net = require('net');
const fs = require('fs');
const stat = require('fs/promises').stat;

const Message = require('./Message.js');

const { CLIENT_STATE_READY, CLIENT_STATE_PLAYING, CLIENT_STATE_PAUSED, CLIENT_STATE_STOPPED, CLIENT_STATE_ERROR } = require('../libs/constants.js');

class AudioServer {

    constructor(file) {
        this.listen = config?.server?.listen || '0.0.0.0';
        this.port = 0;
        this.buffer = {
            file: file,
            stream: fs.createReadStream(file),
            limit: (config?.audio.bufferlimit || 256) * 1048576
        };
        this.clients = [];
        this.sockets = [];
        this.playback = {
            position: undefined
        };
        this.server = net.createServer();
        this.#prepare();
    }

    async #prepare() {
        if (server?.clients === undefined || server.clients.length === 0) {
            logger.warn('there are currently no clients connected, aborting preparation of audio server...')
            this.aborted = true;
            return;
        }
        await new Promise((resolve, reject) => {
            this.server.listen(this.port, this.listen).on('listening', () => {
                this.port = this.server.address().port;
                logger.info('audio server listening on ' + this.listen + ':' + this.port + '...');
                this.#handleEvents();
                resolve();
            });
            this.server.on('connection', (socket) => {
                this.sockets.push(socket);
            });
            this.server.on('error', (err) => {
                reject('an error occured preparing the audio server for file \'' + this.file + '\' > ' + err);
            });
        });
        const stats = await stat(this.buffer.file);
        this.buffer.size = stats.size;
        this.buffer.threshold = (this.buffer.size / 100) / (!isNaN(config.audio?.threshold) || 30);
        this.#announceAudioServer();
    }

    #handleEvents() {
        eventparser.on('audio:register', (data) => {
            if (data?.clientId === undefined || data?.socket === undefined) {
                return;
            }
            let client = server.getClientById(data.clientId);
            if (client === undefined) {
                return;
            }
            let socket;
            for (let index = 0; index < this.sockets.length; index++) {
                if (this.sockets[index].remotePort === data.socket) {
                    socket = this.sockets[index];
                    this.sockets.splice(index, 1);
                    break;
                }
            }
            if (socket === undefined) {
                return;
            }
            client.audiosocket = socket;
            this.clients.push(client);
            logger.debug(client.getTag() + ' connected to audio server...');
            // this.#sendData(client, data);
            // TEST ONLY
            const timestamp = Date.now();
            const stream = fs.createReadStream(this.buffer.file);
            stream.on('data', (data) => {
                const backpressure = false;
                for (let index = 0; index < this.clients.length; index++) {
                    if (!this.clients[index].audiosocket.write(data)) {
                        backpressure = true;
                    }
                }
                if (backpressure === true) {
                    logger.debug('BACKPRESSURE DETECTED - PAUSING STREAM!');
                    stream.pause();
                }
            });
            stream.on('drain', () => {
                logger.debug('STREAM IS DRAINED - RESUMING!');
                stream.resume();
            });
            stream.on('close', () => {
                for (let index = 0; index < this.clients.length; index++) {
                    this.clients[index].audiosocket.end();
                }
                logger.debug('buffering file \'' + this.buffer.file + '\' took ' + (Date.now() - timestamp) + 'ms (size: ' + stream.bytesRead + ' bytes)');
            });
            stream.on('error', (error) => {
                logger.error('')
            });

           
        });
        eventparser.on('audio:state', (data) => {
            this.#handleStateChange(data);
        });
    }

    #handleStateChange(data) {
        if (data?.clientId === undefined || data?.state === undefined) {
            return;
        }
        let client = this.#getClientById(data.clientId);
        if (client === undefined) {
            return;
        }
        logger.debug(client.getTag() + ' state changed to \'' + data.state + '\'');
        client.state = data.state;
        switch (client.state) {
            case CLIENT_STATE_READY:
                return this.#handleStateReady(client);
            case CLIENT_STATE_PLAYING:
                return this.#handleStatePlaying(client);
            case CLIENT_STATE_PAUSED:
                return this.#handleStatePaused(client, data);
            case CLIENT_STATE_STOPPED:
                return this.#handleStateStopped(client, data);
            case CLIENT_STATE_ERROR:
                return this.#handleStateError(client, data);
        }
    }

    async #handleStateReady(client) {
        logger.debug(client.getTag() + ' is ready for playback...');
        for (let index = 0; index < this.clients.length; index++) {
            if (this.clients[index].state !== CLIENT_STATE_READY) {
                return;
            }
        }
        this.#startPlayback();
    }

    async #handleStatePlaying(client) {
        logger.debug(client.getTag() + ' has started playback...');
        // TODO: remove - test only
        await sleep(5000);
        this.#pausePlayback();
    }

    async #handleStatePaused(client, data) {
        if (client === undefined || data === undefined) {
            return;
        }
        logger.debug(client.getTag() + ' paused playback at position \'' + data.position + '\'...');
        for (let index = 0; index < this.playback.paused.length; index++) {
            if (this.playback.paused[index] === client.id) {
                if (this.playback.position === 0 || this.playback.position > data.position) {
                    this.playback.position = data.position;
                }
            }
        }
        // TODO: remove - test only
        await sleep(1);
        this.#startPlayback();
    }

    async #handleStateStopped(client, data) {
        logger.debug(client.getTag() + ' stopped playback at position \'' + data.position + '\'...');
    }

    async #handleStateError(client, data) {
        logger.error(client.getTag() + ' experienced an error during playback at position \'' + data.position + '\': ' + data.error);
    }

    async #sendData(client) {
        const timestamp = Date.now();
        // const buffer = await this.#bufferFile();
        return new Promise(async (resolve, reject) => {
            this.buffer.stream.on('data', (data) => {
                this.buffer.buffered += data.length;
                buffer.push(data);
                if (resolved !== true && this.buffer.stream.bytesRead >= this.buffer.threshold) {
                    resolved = true;
                    logger.debug('buffering threshold of ' + this.buffer.threshold + ' bytes for file \'' + this.buffer.file + '\' took ' + (Date.now() - timestamp) + 'ms');
                    resolve(buffer);
                }
                if (this.buffer.buffered >= this.buffer.limit) {
                    logger.debug('pausing read stream');
                    this.buffer.stream.pause();
                    if (resolved !== true) {
                        resolved = true;
                        resolve(buffer);
                    }
                }
            });
            this.buffer.stream.on('close', () => {
                // this.buffer.data = Buffer.concat(buffer);
                logger.debug('buffering file \'' + this.buffer.file + '\' took ' + (Date.now() - timestamp) + 'ms (size: ' + this.buffer.stream.bytesRead + ' bytes)');
                // resolve();
            });
            this.buffer.stream.on('error', (error) => {
                // TODO: handle with try/catch
                reject(error);
            });
            // this.buffer.written = 0;
            // while (true) {
            //     if (buffer[0] === undefined) {
            //         await sleep(1);
            //         continue;
            //     }
            //     const tmp = buffer[0];
            //     buffer.shift();
            //     if ((client.audiosocket.bytesWritten + tmp.length) >= this.buffer.size) {
            //         logger.debug(client.getTag() + ' sent audio file \'' + this.buffer.file + '\' after ' + (Date.now() - timestamp) + 'ms...');
            //         client.audiosocket.end(tmp);
            //         break;
            //     }
            //     client.audiosocket.write(tmp);
            //     this.buffer.buffered -= tmp.length;
            //     if (this.buffer.stream.isPaused() && this.buffer.buffered < this.buffer.limit) {
            //         this.buffer.stream.resume();
            //         logger.debug('resuming read stream');
            //     }
            // }
            client.audiosocket.on('error', (error) => {
                logger.error(client.getTag() + ' encountered an error: ' + error);
            });
            client.audiosocket.on('end', () => {
                logger.debug(client.getTag() + ' ended audio socket');
            });
            client.audiosocket.on('close', (hadError) => {
                let fn = resolve;
                let msg = client.getTag() + ' closed audio socket';
                if (hadError === true) {
                    msg += ' after an error';
                    fn = reject;
                }
                logger.debug(msg);
                fn(msg);
            });
        });
    }

    // async #waitForBuffer() {
    //     while (this.buffer.data === undefined || this.buffer.data.length < this.buffer.size) {
    //         await sleep(1);
    //     }
    //     return this.buffer.data;
    // }

    // async #bufferFile() {
    //     // const stream = fs.createReadStream(this.buffer.file);
    //     return new Promise((resolve, reject) => {
    //         const timestamp = Date.now();
    //         const buffer = [];
    //         let resolved = false;
    //         this.buffer.buffered = 0;
    //         this.buffer.stream.on('data', (data) => {
    //             this.buffer.buffered += data.length;
    //             buffer.push(data);
    //             if (resolved !== true && this.buffer.stream.bytesRead >= this.buffer.threshold) {
    //                 resolved = true;
    //                 logger.debug('buffering threshold of ' + this.buffer.threshold + ' bytes for file \'' + this.buffer.file + '\' took ' + (Date.now() - timestamp) + 'ms');
    //                 resolve(buffer);
    //             }
    //             if (this.buffer.buffered >= this.buffer.limit) {
    //                 logger.debug('pausing read stream');
    //                 this.buffer.stream.pause();
    //                 if (resolved !== true) {
    //                     resolved = true;
    //                     resolve(buffer);
    //                 }
    //             }
    //         });
    //         this.buffer.stream.on('close', () => {
    //             // this.buffer.data = Buffer.concat(buffer);
    //             logger.debug('buffering file \'' + this.buffer.file + '\' took ' + (Date.now() - timestamp) + 'ms (size: ' + this.buffer.stream.bytesRead + ' bytes)');
    //             // resolve();
    //         });
    //         this.buffer.stream.on('error', (error) => {
    //             // TODO: handle with try/catch
    //             reject(error);
    //         });
    //     });
    // }

    #getClientById(clientId) {
        if (clientId === undefined) {
            return;
        }
        for (let index = 0; index < this.clients.length; index++) {
            const client = this.clients[index];
            if (client.id !== clientId) {
                continue;
            }
            return client;
        }
    }

    async #announceAudioServer() {
        const broadcasted = await new Message('audio:initialize', {
            port: this.server.address().port,
            size: this.buffer.size,
            threshold: this.buffer.threshold
        }).broadcast(true);
        logger.debug('sent broadcast for audio server to client(s) \'' + broadcasted + '\'...');
    }

    async #startPlayback() {
        const broadcasted = await new Message('audio:play', { position: this.playback.position }).broadcast();
        logger.debug('sent broadcast to start playback to client(s) \'' + broadcasted + '\'...');
        this.playback.started = broadcasted;
    }

    async #stopPlayback() {
        const broadcasted = await new Message('audio:stop').broadcast();
        logger.debug('sent broadcast to stop playback to client(s) \'' + broadcasted + '\'...');
        this.playback.stopped = broadcasted;
    }

    async #pausePlayback() {
        const broadcasted = await new Message('audio:pause').broadcast();
        logger.debug('sent broadcast to pause playback to client(s) \'' + broadcasted + '\'...');
        this.playback.paused = broadcasted;
    }

    async destroy() {
        eventparser.removeAllListeners('audio:ready');
        for (let index = 0; index < this.clients.length; index++) {
            const audiosocket = this.clients[index].audiosocket;
            if (audiosocket.destroyed === true) {
                continue;
            }
            audiosocket.destroy();
        }
        await new Promise((resolve, reject) => {
            this.server.close((err) => {
                if (err !== undefined) {
                    reject(err);
                }
                resolve();
            });
        });

    }

}

module.exports = AudioServer;