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
        };
        this.clients = [];
        this.sockets = [];
        this.playback = {
            position: 0
        };
        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();
        this.#bufferFile();
    }

    #handleEvents(socket) {
        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);
        });
        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.#waitForBuffer();
        return new Promise((resolve, reject) => {
            client.audiosocket.end(buffer, () => {
                logger.debug(client.getTag() + ' sent audio file \'' + this.buffer.file + '\' after ' + (Date.now() - timestamp) + 'ms...');
            });
            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() {
        return new Promise((resolve, reject) => {
            const timestamp = Date.now();
            const buffer = [];
            const stream = fs.createReadStream(this.buffer.file);
            stream.on('data', (data) => {
                buffer.push(data);
            });
            stream.on('close', () => {
                this.buffer.data = Buffer.concat(buffer);
                logger.debug('buffering file \'' + this.buffer.file + '\' took ' + (Date.now() - timestamp) + 'ms (size: ' + this.buffer.data.length + ' bytes)');
                resolve();
            });
            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;