// LICENSE_CODE ZON
'use strict'; /*jslint node:true*/
require('./config.js');
const Duplex = require('stream').Duplex;
const events = require('events');

class Ws_demuxer extends events.EventEmitter {
    constructor(conn){
        super();
        conn.on('message', (msg, flags)=>{
            if (!flags.binary)
                return;
            let vfd = msg.readUInt32BE();
            let b = msg.slice(8);
            let ev = 'msg_'+vfd;
            if (!this.listenerCount(ev))
                return this.emit('error', vfd);
            this.emit(ev, b);
        });
    }
}

class Ws_stream extends Duplex {
    constructor(opt){
        super(opt);
        this.vfd = opt.vfd;
        this.conn = opt.conn;
        this.demuxer = opt.demuxer;
        let on_error, on_close, on_data;
        this.conn.on('error', on_error = ()=>this.emit('error'));
        this.conn.on('close', on_close = ()=>this.emit('close'));
        this.demuxer.on('msg_'+opt.vfd, on_data = buf=>{
            if (!this.push(buf))
            {
                // XXX vladimir: handle pausing ws stream
                return;
            }
        });
        this.on('close', ()=>{
            this.conn.removeListener('error', on_error);
            this.conn.removeListener('close', on_close);
            this.demuxer.removeListener('msg_'+this.vfd, on_data);
        });
    }
    _write(chunk, encoding, cb){
        let b = Buffer.allocUnsafe(chunk.length+8);
        b.writeUInt32BE(this.vfd);
        b.writeUInt32BE(0, 4);
        chunk.copy(b, 8);
        this.conn.send(b, cb);
    }
    _read(size){
        // we begin reading in the constructor
        // XXX vladimir: handle restarting reading when pausing
    }
    // XXX vladimir: node v8 readable stream implements destroy
    destroy(){}
}

module.exports = {Ws_stream, Ws_demuxer};
