kannon-client/classes/Player.js

const EventEmitter = require('events');
const { sleep } = require('../libs/util.js');
const { spawn } = require('child_process');
const createWriteStream = require('fs').createWriteStream;
const unlink = require('fs/promises').unlink;
const resolve = require('path').resolve;

const STATE_SPAWNED = 'spawned';
const STATE_READY = 'ready';
const STATE_PLAYING = 'playing';
const STATE_PAUSED = 'paused';
const STATE_STOPPED = 'stopped';
const STATE_ERROR = 'error';

class Player extends EventEmitter {

    constructor() {
        super();
        this.position = 0;
        this.events = [];
        this.buffer = [];
        this.tmp = {
            file: resolve('/tmp/kannon.tmp')
        }
        this.buffersize = 0;
    }

    async prepare(size, threshold) {
        logger.debug('preparing audio player...');
        this.size = size;
        this.threshold = threshold;
        this.tmp.stream = createWriteStream(this.tmp.file);
        this.#reset();
        this.#spawnProcess();
    }

    feed(buffer) {
        this.buffer.push(buffer);
        this.buffersize += buffer.length;
        if (this.isSpawned() && this.buffersize >= this.threshold) {
            this.#setState(STATE_READY);
        }
    }

    async play() {
        if (this.isPlaying()) {
            this.stop();
        }
        await this.#spawnProcess();
        this.process.stderr.on('data', (data) => {
            this.#setState(STATE_PLAYING);
            data = data.toString();
            const position = data.toString().trim().split(' ')[0];
            if (position.length === 0 || isNaN(position)) {
                return;
            }
            this.position = position;
        });
        this.process.stdin.on('error', (error) => {
            this.#setState(STATE_ERROR, error);
        });
        if (this.process.spawnargs[this.process.spawnargs.length] === this.tmp.file) {
            return;
        }
        if (this.buffer === undefined || this.buffer.length === 0) {
            logger.warn('aborting playback of an empty buffer...');
            return;
        }
        const timestamp = Date.now();
        while (true) {
            if (this.buffer.length === 0 && this.buffersize !== this.size) {
                await sleep(1);
                continue;
            }
            const tmp = this.buffer[0];
            this.buffer.shift();
            if (this.buffer.length === 0 && this.buffersize === this.size) {
                logger.warn('BUFFER EMPTIED AFTER ' + ( Date.now() - timestamp) + 'ms')
                this.process.stdin.end(tmp);
                this.tmp.stream.end(tmp);
                break;
            }
            this.tmp.stream.write(tmp);
            this.process.stdin.write(tmp);
        }
    }

    async pause() {
        this.#setState(STATE_PAUSED);
        this.#reset();
        this.#spawnProcess();
    }

    async stop() {
        this.#setState(STATE_STOPPED);
        this.#reset();
    }

    isSpawned() {
        return this.state === STATE_SPAWNED;
    }

    isReady() {
        return this.state === STATE_READY;
    }

    isPlaying() {
        return this.state === STATE_PLAYING;
    }

    isPaused() {
        return this.state === STATE_PAUSED;
    }

    isFinished() {
        return this.state === STATE_STOPPED;
    }

    hasError() {
        return this.state === STATE_ERROR;
    }

    async #spawnProcess() {
        if (this.process !== undefined) {
            return;
        }
        return new Promise((resolve, reject) => {
            const args = [
                '-vn',
                '-nodisp'
            ];
            if (this.isPaused()) {
                args.unshift('-ss', this.position);
                args.push(this.tmp.file);
            } else {
                args.push('-');
            }
            this.process = spawn("ffplay", args);
            this.process.on('error', (error) => {
                this.#reset();
                // TODO: try/catch error
                reject('error spawning process \'ffplay\': ' + error);
            });
            this.process.on('exit', (code, signal) => {
                let msg = 'process \'ffplay\' exited with';
                if (code !== undefined) {
                    msg += ' code \'' + code + '\'';
                } else {
                    msg += ' signal \'' + signal + '\'';
                }
                logger.debug(msg);
                this.#closeStdIO();
            });
            this.process.on('close', (code, signal) => {
                let msg = 'process \'ffplay\' closed with';
                if (code !== undefined) {
                    msg += ' code \'' + code + '\'';
                } else {
                    msg += ' signal \'' + signal + '\'';
                }
                logger.debug(msg);
                this.process = undefined;
            });
            this.process.on('spawn', () => {
                logger.info('spawned process \'ffplay\' (pid: ' + this.process.pid + ')...');
                this.#setState(STATE_SPAWNED);
                resolve();
            });
        });

    }

    #setState(state, data) {
        if (this.state === state) {
            return;
        }
        this.state = state;
        logger.debug('setting state of audio player to \'' + state + '\'...');
        if (this.events.includes(state)) {
            return;
        }
        logger.debug('emitting state \'' + state + '\' of audio player...');
        this.emit(state, data);
        this.events.push(state);
    }

    #killProcess() {
        this.#closeStdIO();
        if (this.process?.killed === false) {
            this.process.kill('SIGTERM');
        }
    }

    #closeStdIO() {
        if (this.process?.stdio === undefined) {
            return;
        }
        logger.debug('closing all stdio streams of process \'ffplay\' (pid: ' + this.process.pid + ')...');
        for (let index = 0; index < this.process.stdio.length; index++) {
            this.process.stdio[index].destroy();
        }
    }

    async #reset() {
        this.#killProcess();
        this.events = [];
        this.buffer = [];
        if (!this.isPaused()) {
            this.position = 0;
            this.backupbuffer = [];
            this.buffersize = 0;
            try {
                await unlink(this.tmp.file);
            } catch (error) {
                logger.error('no file to unlink ' + this.tmp.file);
            }
        }
    }

}

module.exports = Player;