<commit_before>'use strict'

const ioEvts          = require('./ioEvts');
const ctx             = require('chalk');
const Rx              = require('rx');
const chalk           = new ctx.constructor({enabled: true});

class SocketServer {
    constructor(pmpEngineInstance, ioConfig) {
        this._pmpEngine                 = pmpEngineInstance;
        this._config                    = ioConfig;
        this._superviseSocket           = require('express')();
        this._httpSocket                = require('http').Server(this._superviseSocket);
        this._io                        = require('socket.io')(this._httpSocket);
        this._inputStream               = null;
        this._inputStreamSubscription   = null;

        //setup
        this._setupInput();
        this._setupOutput();
        this._setupServer();
    }

    //setup server and listeners
    _setupServer() {
        //connect/disconnect behavior
        this._io.on(ioEvts.utils.connect, socket => {
            console.log(chalk.green(`Socket service CONNECTED`));

            //handle disconnects
            socket.once(ioEvts.utils.disconnect, () => {
                //clean input stream
                if(this._inputStreamSubscription) this._inputStreamSubscription.unsubscribe();
                if(this._inputStream) this._inputStream = null;
                console.log(chalk.yellow(`Socket service DISCONNECTED`));
            });
        });

        //socket server listening
        this._httpSocket.listen(this._config.port, this._config.host, () => {
            console.log(chalk.blue(`Socket service listening to port ${this._config.port}`)); 
        });
    }

    _setupInput() {
        this._io.on(ioEvts.utils.connect, socket => {
            this._inputStream = Rx.Observable.fromEvent(socket, 'input');

            //INPUTS SUBSCRIPTION
            this._inputStreamSubscription = this._inputStream.subscribe(inputCmd => {
                console.log('received input', inputCmd);
            });
        });
    }

    _setupOutput() {

    }
}

module.exports = {
    SocketServer: SocketServer
}<commit_msg>added output handling<commit_after>'use strict'

const ioEvts          = require('./ioEvts');
const ctx             = require('chalk');
const Rx              = require('rx');
const chalk           = new ctx.constructor({enabled: true});

class SocketServer {
    constructor(pmpEngineInstance, ioConfig) {
        this._pmpEngine                 = pmpEngineInstance;
        this._config                    = ioConfig;
        this._superviseSocket           = require('express')();
        this._httpSocket                = require('http').Server(this._superviseSocket);
        this._io                        = require('socket.io')(this._httpSocket);
        this._inputStream               = null;
        this._subscriptions             = [];

        //setup
        this._setupInput();
        this._setupOutput();
        this._setupServer();
    }

    //setup server and listeners
    _setupServer() {
        //connect/disconnect behavior
        this._io.on(ioEvts.utils.connect, socket => {
            console.log(chalk.green(`Socket service CONNECTED`));

            //handle disconnects
            socket.once(ioEvts.utils.disconnect, () => {
                //clean subscriptions
                if(this._subscriptions.length > 0) {
                    this._subscriptions.map(subscription => { subscription.unsubscribe(); });
                    this._subscriptions = [];
                }
                if(this._inputStream) this._inputStream = null;
                console.log(chalk.yellow(`Socket service DISCONNECTED`));
            });
        });

        //socket server listening
        this._httpSocket.listen(this._config.port, this._config.host, () => {
            console.log(chalk.blue(`Socket service listening to port ${this._config.port}`)); 
        });
    }

    _setupInput() {
        this._io.on(ioEvts.utils.connect, socket => {
            this._inputStream = Rx.Observable.fromEvent(socket, 'input');

            //INPUTS SUBSCRIPTION
            this._subscriptions.push(this._inputStream.subscribe(inputCmd => {
                console.log('received input', inputCmd);
            }));
        });
    }

    _setupOutput() {
        this._io.on(ioEvts.utils.connect, socket => {
            //status streaming
            this._subscriptions.push(this._pmpEngine.pmpEngineStatusStream.subscribe(status => {
                let evt = ioEvts.outputs.engineStatusLog(status);
                socket.broadcast.emit(evt);
            }));
            //log streaming
            this._subscriptions.push(this._pmpEngine.pmpEngineLogsStream.subscribe(log => {
                let evt = ioEvts.outputs.log(log);
                socket.broadcast.emit(evt);
            }));
            //errors streaming
            this._subscriptions.push(this._pmpEngine.pmpEngineErrorsStream.subscribe(err => {
                let evt = ioEvts.outputs.error(err);
                socket.broadcast.emit(evt);
            }));
        });
    }
}

module.exports = {
    SocketServer: SocketServer
}
{'fraction_non_alphanumeric': 0.1210710128055879, 'fraction_numerical': 0.00019402405898331392, 'mean_word_length': 1.778975741239892, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2931072', 'n_tokens_mistral': 1366, 'n_tokens_neox': 1322, 'n_words': 209}