// LICENSE_CODE ZON
'use strict'; /*jslint node:true, browser:true, es6:true*/
(function(){
let define;
let is_node = typeof module=='object' && module.exports && module.children;
var is_rn = typeof global=='object' && !!global.nativeRequire ||
    typeof navigator=='object' && navigator.product=='ReactNative';
if (is_rn)
{
    define = require('./require_node.js').define(module, '../',
        require('/util/conv.js'), require('/util/etask.js'),
        require('/util/events.js'), require('/util/string.js'),
        require('/util/zerr.js'));
}
else if (!is_node)
    define = self.define;
else
    define = require('./require_node.js').define(module, '../');
define(['/util/conv.js', '/util/etask.js', '/util/events.js',
    '/util/string.js', '/util/zerr.js'],
    function(conv, etask, events, string, zerr){

const ef = etask.ef;
const zjson_opt = {date: true, func: true, re: true};
const is_win = /^win/.test(process.platform);
const default_user_agent = is_node ? (()=>{
    const zconf = require('./config.js');
    const conf = require('./conf.js');
    return `Hola ${conf.app}/${zconf.ZON_VERSION}`;
})() : undefined;
const debug_str_len = 4096;
let zcounter; // has to be lazy because zcounter.js itself uses this module

function noop(){}

class WS extends events.EventEmitter {
    constructor(opt){
        super();
        this.ws = undefined;
        this.data = opt.data;
        this.connected = false;
        this.reason = undefined;
        this.zc_rx = opt.zcounter=='rx' || opt.zcounter=='all';
        this.zc_tx = opt.zcounter=='tx' || opt.zcounter=='all';
        this.zc = opt.zcounter
            ? opt.label ? `${opt.label}_ws` : 'ws' : undefined;
        this.label = opt.label;
        this.remote_label = undefined;
        this.local_addr = undefined;
        this.local_port = undefined;
        this.remote_addr = undefined;
        this.remote_port = undefined;
        this.remote_forwarded = false;
        this.status = 'disconnected';
        this.ping = is_node && opt.ping!=false;
        this.ping_interval = typeof opt.ping_interval=='function'
            ? opt.ping_interval() : opt.ping_interval || 60000;
        this.ping_timeout = typeof opt.ping_timeout=='function'
            ? opt.ping_timeout() : opt.ping_timeout || 10000;
        this.ping_timer = undefined;
        this.ping_last = undefined;
        this.ipc = opt.ipc_client
            ? new IPC_client(this, opt.ipc_client, {zjson: opt.ipc_zjson})
            : undefined;
        this.time_parse = opt.time_parse;
        if (opt.ipc_server)
        {
            new IPC_server(this, opt.ipc_server,
                {zjson: opt.ipc_zjson, sync: opt.ipc_sync});
        }
        this.mux = opt.mux ? new Mux(this) : undefined;
        if (this.zc && !zcounter)
            zcounter = require('./zcounter.js');
    }
    send(msg){
        if (zerr.is.info())
        {
            zerr.info(typeof msg=='string'
                ? `${this}> str: ${string.trunc(msg, debug_str_len)}`
                : `${this}> buf: ${msg.length} bytes`);
        }
        if (!this.connected)
        {
            if (zerr.is.info())
                zerr.info(`${this}: sending failed: disconnected`);
            return false;
        }
        // workaround for ws library: the socket is already closing,
        // but a notification has not yet been emitted
        if (this.ws.readyState==2) // ws.CLOSING
        {
            if (zerr.is.info())
                zerr.info(`${this}: sending failed: socket closing`);
            return false;
        }
        this.ws.send(msg);
        if (this.zc_tx)
        {
            zcounter.inc(`${this.zc}_tx_msg`);
            zcounter.inc(`${this.zc}_tx_bytes`, msg.length);
            zcounter.avg(`${this.zc}_tx_bytes_per_msg`, msg.length);
        }
        return true;
    }
    json(data){ return this.send(JSON.stringify(data)); }
    zjson(data, ttl){ return this.send(conv.JSON_stringify(data, zjson_opt)); }
    _check_status(){
        let prev = this.status;
        this.status = this.ws
            ? this.connected ? 'connected' : 'connecting'
            : 'disconnected';
        if (this.status!=prev)
        {
            this.emit(this.status);
            this.emit('status', this.status);
            if (this.status=='disconnected')
                this._on_disconnected();
        }
    }
    _on_disconnected(){ this.emit('destroyed'); }
    _assign(ws){
        this.ws = ws;
        this.ws.onopen = this._on_open.bind(this);
        this.ws.onclose = this._on_close.bind(this);
        this.ws.onmessage = this._on_message.bind(this);
        this.ws.onerror = this._on_error.bind(this);
        if (this.ping)
            this.ws.on('pong', this._on_pong.bind(this));
    }
    abort(code, reason){
        this.reason = reason||code;
        let msg = `${this}: closed locally`;
        if (this.reason)
            msg += ` (${this.reason})`;
        this._close(true, code, reason);
        zerr.warn(msg);
        if (this.zc && code)
            zcounter.inc(`${this.zc}_err_${code}`);
        this._check_status();
    }
    _close(close, code, reason){
        if (!this.ws)
            return;
        if (this.ping)
        {
            clearTimeout(this.ping_timer);
            this.ping_timer = undefined;
            this.ping_last = undefined;
            this.ws.removeAllListeners('pong');
        }
        this.ws.onopen = undefined;
        this.ws.onclose = undefined;
        this.ws.onmessage = undefined;
        this.ws.onerror = noop;
        this.local_addr = undefined;
        this.local_port = undefined;
        this.remote_addr = undefined;
        this.remote_port = undefined;
        if (close)
        {
            if (this.ws.terminate && !this.connected)
                this.ws.terminate();
            else
                this.ws.close(code, reason);
        }
        this.ws = undefined;
        this.connected = false;
    }
    toString(){
        let res = this.label ? `${this.label} WS` : 'WS';
        if (this.remote_label || this.remote_addr)
            res += ' (';
        if (this.remote_label)
            res += this.remote_label;
        if (this.remote_label && this.remote_addr)
            res += ' ';
        if (this.remote_addr)
            res += this.remote_addr;
        if (this.remote_label || this.remote_addr)
            res += ')';
        return res;
    }
    _on_open(){
        if (this.connected)
            return;
        this.connected = true;
        let sock = this.ws._socket||{};
        // XXX pavlo: uws lib doesn't have these properties in _socket:
        // https://github.com/hola/uWebSockets-bindings/blob/master/nodejs/src/uws.js#L276
        // get them from upgrade request
        this.local_addr = sock.localAddress;
        this.local_port = sock.localPort;
        if (this.remote_addr==sock.remoteAddress)
            this.remote_forwarded = false;
        if (!this.remote_forwarded)
        {
            this.remote_addr = sock.remoteAddress;
            this.remote_port = sock.remotePort;
        }
        zerr.notice(`${this}: connected`);
        if (this.ping)
        {
            this.ping_timer = setTimeout(this._ping.bind(this),
                this.ping_interval);
        }
        this._check_status();
    }
    _on_close(event){
        this.reason = event.reason||event.code;
        zerr.notice(`${this}: closed by remote (${this.reason})`);
        this._close();
        this._check_status();
    }
    _on_error(event){
        this.reason = event.message||'Network error';
        zerr(`${this}: ${this.reason}`);
        if (this.zc)
            zcounter.inc(`${this.zc}_err`);
        this._close();
        this._check_status();
    }
    _on_message(event){
        let msg = event.data, handled = false;
        if (msg instanceof ArrayBuffer)
            msg = Buffer.from(Buffer.from(msg)); // make a copy
        if (zerr.is.info())
        {
            zerr.info(typeof msg=='string'
                ? `${this}< str: ${string.trunc(msg, debug_str_len)}`
                : `${this}< buf: ${msg.length} bytes`);
        }
        if (this.zc_rx)
        {
            zcounter.inc(`${this.zc}_rx_msg`);
            zcounter.inc(`${this.zc}_rx_bytes`, msg.length);
            zcounter.avg(`${this.zc}_rx_bytes_per_msg`, msg.length);
        }
        try {
            if (typeof msg=='string')
            {
                if (this._events.zjson)
                {
                    let parsed;
                    if (this.zc && this.time_parse)
                    {
                        let started = Date.now();
                        parsed = conv.JSON_parse(msg);
                        zcounter.inc(`${this.zc}_parse_zjson_ms`,
                            Date.now()-started);
                    }
                    else
                        parsed = conv.JSON_parse(msg);
                    this.emit('zjson', parsed);
                    handled = true;
                }
                if (this._events.json)
                {
                    let parsed, started;
                    if (this.zc && this.time_parse)
                        started = Date.now();
                    parsed = JSON.parse(msg);
                    if (this.zc && this.time_parse)
                    {
                        zcounter.inc(`${this.zc}_parse_json_ms`,
                            Date.now()-started);
                    }
                    this.emit('json', parsed);
                    handled = true;
                }
                if (this._events.text)
                {
                    this.emit('text', msg);
                    handled = true;
                }
            }
            else
            {
                if (this._events.bin)
                {
                    this.emit('bin', msg);
                    handled = true;
                }
            }
            if (this._events.raw)
            {
                this.emit('raw', msg);
                handled = true;
            }
        } catch(e){ ef(e);
            zerr(`${this}: ${zerr.e2s(e)}`);
            return this.abort(1011, e.message);
        }
        if (!handled)
            this.abort(1003, 'Unexpected message');
        if (this.ping_last)
        {
            clearTimeout(this.ping_timer);
            this.ping_timer = setTimeout(this._ping_expire.bind(this),
                this.ping_timeout);
        }
    }
    _on_pong(){
        clearTimeout(this.ping_timer);
        let rtt = Date.now()-this.ping_last;
        this.ping_last = undefined;
        this.ping_timer = setTimeout(this._ping.bind(this),
            Math.max(this.ping_interval-rtt, 0));
        if (zerr.is.debug())
            zerr.debug(`${this}< pong (rtt ${rtt}ms)`);
        if (this.zc)
            zcounter.avg(`${this.zc}_ping_ms`, rtt);
    }
    _ping(){
        // workaround for ws library: the socket is already closing,
        // but a notification has not yet been emitted
        if (this.ws.readyState==2) // ws.CLOSING
            return;
        this.ws.ping();
        this.ping_timer = setTimeout(this._ping_expire.bind(this),
            this.ping_timeout);
        this.ping_last = Date.now();
        if (zerr.is.debug())
            zerr.debug(`${this}> ping (max ${this.ping_timeout}ms)`);
    }
    _ping_expire(){ this.abort(1002, 'Ping timeout'); }
    close(code, reason){
        if (!this.ws)
            return;
        let msg = `${this}: closed locally`;
        if (reason||code)
            msg += ` (${reason||code})`;
        this._close(true, code, reason);
        zerr.notice(msg);
        if (this.zc && code)
            zcounter.inc(`${this.zc}_err_${code}`);
        this._check_status();
    }
    inspect(){
        return {
            class: this.constructor.name,
            label: this.toString(),
            status: this.status,
            reason: this.reason,
            local_addr: this.local_addr,
            local_port: this.local_port,
            remote_addr: this.remote_addr,
            remote_port: this.remote_port,
        };
    }
}

class Client extends WS {
    constructor(url, opt={}){
        super(opt);
        this.status = 'connecting';
        this.impl = client_impl(opt);
        this.url = url;
        this.retry_interval = opt.retry_interval||10000;
        this.retry_max = opt.retry_max||this.retry_interval;
        this.next_retry = this.retry_interval;
        if (is_node)
        {
            this.headers = Object.assign(
                {'User-Agent': opt.user_agent||default_user_agent},
                opt.headers);
        }
        this.deflate = !!opt.deflate;
        if (opt.proxy)
        {
            let _lib = require('https-proxy-agent');
            this.agent = new _lib(opt.proxy);
        }
        else
            this.agent = opt.agent;
        this.reconnect_timer = undefined;
        this.handshake_timeout = opt.handshake_timeout===undefined
            ? 10000 : opt.handshake_timeout;
        this.handshake_timer = undefined;
        if (this.zc)
            zcounter.inc_level(`level_${this.zc}_online`, 0, 'sum');
        this._connect();
    }
    // we don't want WS to emit 'destroyed', Client controls it by itself,
    // because it supports reconnects
    _on_disconnected(){}
    _assign(ws){
        super._assign(ws);
        if (this.zc)
            zcounter.inc_level(`level_${this.zc}_online`, 1, 'sum');
    }
    _close(close, code, reason){
        if (this.zc && this.ws)
            zcounter.inc_level(`level_${this.zc}_online`, -1, 'sum');
        if (this.handshake_timer)
            this.handshake_timer = clearTimeout(this.handshake_timer);
        super._close(close, code, reason);
    }
    _connect(){
        this.reason = undefined;
        this.reconnect_timer = undefined;
        let opt = {headers: this.headers};
        if (!is_rn)
        {
            opt.agent = this.agent;
            opt.perMessageDeflate = this.deflate;
        }
        this._assign(new this.impl(this.url, undefined, opt));
        if (this.handshake_timeout)
        {
            this.handshake_timer = setTimeout(
                ()=>this.abort(1002, 'Handshake timeout'),
                this.handshake_timeout);
        }
        this._check_status();
    }
    _reconnect(){
        let delay = this.next_retry;
        if (typeof delay=='function')
            delay = delay();
        else
        {
            this.next_retry = Math.min(delay*2,
                typeof this.retry_max=='function'
                    ? this.retry_max() : this.retry_max);
        }
        if (zerr.is.info())
            zerr.info(`${this}: will retry in ${delay}ms`);
        this.reconnect_timer = setTimeout(()=>this._connect(), delay);
    }
    _on_open(){
        if (this.handshake_timer)
            this.handshake_timer = clearTimeout(this.handshake_timer);
        this.next_retry = this.retry_interval;
        super._on_open();
    }
    _on_close(event){
        this._reconnect();
        super._on_close(event);
    }
    _on_error(event){
        this._reconnect();
        super._on_error(event);
    }
    abort(code, reason){
        this._reconnect();
        super.abort(code, reason);
    }
    close(code, reason){
        if (this.ws)
            this.emit('destroyed');
        super.close(code, reason);
        if (this.reconnect_timer)
        {
            clearTimeout(this.reconnect_timer);
            this.reconnect_timer = undefined;
            this.emit('destroyed');
        }
    }
    inspect(){
        return Object.assign(super.inspect(), {
            url: this.url,
            retry_interval: this.retry_interval,
            retry_max: this.retry_max,
            next_retry: this.next_retry,
            handshake_timeout: this.handshake_timeout,
            deflate: this.deflate,
            reconnecting: !!this.reconnect_timer,
        });
    }
}

class Server {
    constructor(opt={}, handler=undefined){
        this.handler = handler;
        let ws_opt = {
            server: opt.http_server,
            host: opt.host||'0.0.0.0',
            port: opt.port,
            noServer: !opt.http_server && !opt.port,
            path: opt.path,
            clientTracking: false,
            perMessageDeflate: !!opt.deflate,
        };
        if (opt.max_payload)
            ws_opt.maxPayload = opt.max_payload;
        if (opt.verify)
            ws_opt.verifyClient = opt.verify;
        this.ws_server = new (server_impl(opt))(ws_opt);
        this.opt = opt;
        this.label = opt.label;
        this.connections = new Set();
        this.zc = opt.label ? `${opt.label}_ws` : 'ws';
        this.ws_server.addListener('connection', this.accept.bind(this));
        if (opt.port)
            zerr.notice(`${this}: listening on port ${opt.port}`);
        if (!zcounter)
            zcounter = require('./zcounter.js');
        // ensure the metric exists, even if 0
        zcounter.inc_level(`level_${this.zc}_conn`, 0, 'sum');
    }
    toString(){ return this.label ? `${this.label} WS server` : 'WS server'; }
    upgrade(req, socket, head){
        this.ws_server.handleUpgrade(req, socket, head,
            ws=>this.accept(ws, req));
    }
    accept(ws, req=ws.upgradeReq){
        if (!ws._socket.remoteAddress)
        {
            ws.onerror = noop;
            return zerr.warn(`${this}: dead incoming connection`);
        }
        let zws = new WS(this.opt);
        let headers = req && req.headers || {};
        if (this.opt.trust_forwarded)
        {
            let forwarded = headers['x-real-ip'];
            if (!forwarded)
            {
                forwarded = headers['x-forwarded-for'];
                if (forwarded)
                {
                    let ips = forwarded.split(',');
                    forwarded = ips[ips.length-1];
                }
            }
            if (forwarded)
            {
                zws.remote_addr = forwarded;
                zws.remote_forwarded = true;
            }
        }
        let ua = headers['user-agent'];
        let m = /^Hola (.+)$/.exec(ua);
        zws.remote_label = m ? m[1] : ua ? 'web' : undefined;
        zws._assign(ws);
        zws._on_open();
        this.connections.add(zws);
        zcounter.inc(`${this.zc}_conn`);
        zcounter.inc_level(`level_${this.zc}_conn`, 1, 'sum');
        zws.addListener('disconnected', ()=>{
            this.connections.delete(zws);
            zcounter.inc_level(`level_${this.zc}_conn`, -1, 'sum');
        });
        if (this.handler)
        {
            try {
                zws.data = this.handler(zws, req);
            } catch(e){ ef(e);
                zerr(zerr.e2s(e));
                return zws.close(1011, String(e));
            }
        }
        return zws;
    }
    broadcast(msg){
        zerr.info(typeof msg=='string'
            ? `${this}> broadcast str: ${string.trunc(msg, debug_str_len)}`
            : `${this}> broadcast buf: ${msg.length} bytes`);
        for (let zws of this.connections)
            zws.send(msg);
    }
    broadcast_json(data){
        if (this.connections.size)
            this.broadcast(JSON.stringify(data));
    }
    broadcast_zjson(data){
        if (this.connections.size)
            this.broadcast(conv.JSON_stringify(data, zjson_opt));
    }
    close(code, reason){
        zerr.notice(`${this}: closed`);
        this.ws_server.close();
        for (let zws of this.connections)
            zws.close(code, reason);
    }
    inspect(){
        let connections = [];
        for (let c of this.connections)
            connections.push(c.inspect());
        return {
            class: this.constructor.name,
            label: this.toString(),
            opt: this.opt,
            connections,
        };
    }
}

class IPC_client {
    constructor(zws, names, opt={}){
        this._ws = zws;
        this._pending = new Map();
        this._ws.addListener(opt.zjson ? 'zjson' : 'json',
            this._on_resp.bind(this));
        this._ws.addListener('status', this._on_status.bind(this));
        this._ws.addListener('destroyed',
            this._on_status.bind(this, 'destroyed'));
        if (Array.isArray(names))
        {
            for (let name of names)
                this[name] = this._call.bind(this, opt, name);
        }
        else
        {
            for (let name in names)
            {
                let spec = names[name];
                if (typeof spec=='string')
                    spec = {type: spec};
                let _opt = Object.assign({}, opt, spec);
                switch (_opt.type||'call')
                {
                case 'call':
                    this[name] = this._call.bind(this, _opt, name);
                    break;
                case 'post':
                    this[name] = this._post.bind(this, _opt, name);
                    break;
                default:
                    zerr.zexit(
                        `${this._ws}: ${name}: Invalid IPC client spec`);
                }
            }
        }
    }
    _call(opt, cmd, ...arg){
        let _this = this;
        return etask(function*IPC_client_call(){
            let req = {type: 'ipc_call', cmd, cookie: ++IPC_client._cookie};
            if (arg.length==1)
                req.msg = arg[0];
            else if (arg)
                req.arg = arg;
            this.info.label = ()=>_this._ws.toString();
            this.info.cmd = cmd;
            this.info.cookie = req.cookie;
            _this._pending.set(req.cookie, this);
            this.finally(()=>_this._pending.delete(req.cookie));
            if (opt.timeout)
                this.alarm(opt.timeout, ()=>this.throw(new Error('Timeout')));
            let res = {status: _this._ws.status}, prev;
            while (res.status)
            {
                switch (res.status)
                {
                case 'disconnected':
                    if (opt.retry==false || !_this._ws.reconnect_timer)
                        throw new Error(_this._ws.reason||'Connection closed');
                    break;
                case 'connecting':
                    if (opt.retry==false)
                        throw new Error('Connection not ready');
                    break;
                case 'destroyed':
                    throw new Error(_this._ws.reason||'Connection closed');
                case 'connected':
                    if (opt.zjson)
                        _this._ws.zjson(req);
                    else
                        _this._ws.json(req);
                    break;
                }
                do {
                    prev = res.status;
                    res = yield this.wait();
                } while (prev==res.status);
            }
            return res.value;
        });
    }
    _post(opt, cmd, ...arg){
        let req = {type: 'ipc_post', cmd};
        if (arg.length==1)
            req.msg = arg[0];
        else if (arg)
            req.arg = arg;
        if (opt.zjson)
            this._ws.zjson(req);
        else
            this._ws.json(req);
    }
    _on_resp(msg){
        if (!msg || msg.type!='ipc_result' && msg.type!='ipc_error')
            return;
        let task = this._pending.get(msg.cookie);
        if (!task)
        {
            return zerr.info(`${this._ws}: `
                +`unexpected IPC cookie ${msg.cookie}`);
        }
        if (msg.type=='ipc_result')
            task.continue({value: msg.msg});
        else
            task.throw(new Error(msg.msg));
    }
    _on_status(status){
        for (let task of this._pending.values())
            task.continue({status});
    }
}
IPC_client._cookie = 0;

class IPC_server {
    constructor(zws, methods, opt={}){
        this.ws = zws;
        if (Array.isArray(methods))
        {
            this.methods = {};
            for (let m of methods)
                this.methods[m] = true;
        }
        else
            this.methods = methods;
        this.zjson = !!opt.zjson;
        this.sync = !!opt.sync;
        this.pending = this.sync ? undefined : new Set();
        this.ws.addListener(this.zjson ? 'zjson' : 'json',
            this._on_call.bind(this));
        if (!this.sync)
        {
            this.ws.addListener('disconnected',
                this._on_disconnected.bind(this));
        }
    }
    _on_call(msg){
        if (!msg || !msg.cmd)
            return;
        let type = msg.type||'ipc_call';
        if (type!='ipc_call' && type!='ipc_post')
            return;
        let method = this.methods[msg.cmd];
        if (method==true)
            method = this.ws.data[msg.cmd];
        if (!method)
        {
            let err = `Method ${msg.cmd} not defined`;
            if (type=='ipc_post')
                return zerr(`${this.ws}: ${err}`);
            return this.ws.json({
                type: 'ipc_error',
                cmd: msg.cmd,
                cookie: msg.cookie,
                msg: err,
            });
        }
        let arg = msg.arg || [msg.msg];
        let _this = this;
        etask(function*IPC_server_on_call(){
            if (!_this.sync && type=='ipc_post')
            {
                _this.pending.add(this);
                this.finally(()=>_this.pending.delete(this));
            }
            this.info.label = ()=>_this.ws.toString();
            this.info.cmd = msg.cmd;
            this.info.cookie = msg.cookie;
            try {
                let rv = method.apply(_this.ws.data||_this.ws, arg);
                if (!_this.sync)
                    rv = yield rv;
                if (type=='ipc_post')
                    return;
                let res = {
                    type: 'ipc_result',
                    cmd: msg.cmd,
                    cookie: msg.cookie,
                    msg: rv,
                };
                if (_this.zjson)
                    _this.ws.zjson(res);
                else
                    _this.ws.json(res);
            } catch(e){
                if (type=='ipc_post')
                    return zerr(`${_this.ws}: ${msg.cmd}: ${zerr.e2s(e)}`);
                _this.ws.json({
                    type: 'ipc_error',
                    cmd: msg.cmd,
                    cookie: msg.cookie,
                    msg: e.message || String(e),
                });
            }
        });
    }
    _on_disconnected(){
        for (let task of this.pending)
            task.return();
    }
}

class Mux {
    constructor(zws){
        this.ws = zws;
        this.streams = new Map();
        this.ws.on('bin', this._on_bin.bind(this));
        this.ws.on('disconnected', this._on_disconnected.bind(this));
    }
    open(id, bytes_allowed=Infinity){
        let stream = this.streams.get(id);
        if (stream)
            return stream;
        const _lib = require('stream');
        let _this = this, suspended;
        stream = new _lib.Duplex({
            read(size){},
            write(data, encoding, cb){
                if (bytes_allowed<=0)
                {
                    suspended = ()=>this._write(data, encoding, cb);
                    return;
                }
                if (zerr.is.info())
                    zerr.info(`${_this.ws}> vfd ${id}`);
                bytes_allowed -= data.length;
                let buf = Buffer.allocUnsafe(data.length+8);
                buf.writeUInt32BE(id, 0);
                buf.writeUInt32BE(0, 4);
                data.copy(buf, 8);
                cb(_this.ws.send(buf) ? undefined
                    : new Error(_this.ws.reason || _this.ws.status));
            },
            destroy(err, cb){
                // XXX viktor: fix once it is clear what happens. ignore this
                // error and let real error from _http_client.js:441 throws
                try {
                    stream.push(null);
                } catch(e){
                    zerr.notice('DEBUG RECURSIVE DESTROY '+e);
                }
                stream.end();
                this.emit('_close');
                process.nextTick(cb, err);
            }
        });
        let _stacktrace = (new Error()).stack;
        stream.prependListener('data', function(chunk){
            if (!this._httpMessage || this.parser && this.parser.socket===this
                || is_rn)
            {
                return;
            }
            // XXX sergey: in case we have a bug and parser freed before data
            // fully consumed, replace damaged parser with new one, but
            // instead of processing data return error, this will close socket
            // and emit error on request object
            const {parsers} = require('_http_common');
            zcounter.inc(`mux_no_parser`);
            zerr('\n--- assert failed, socketinfo:\n'+
                `DEBUG HEADER: ${this._httpMessage._header}\n`+
                `DEBUG mux.open(): ${_stacktrace}\n`+
                `DEBUG WS: ${_this.ws.remote_addr}:${_this.ws.remote_port}`);
            const old_parser = this.parser;
            const parser = this.parser = parsers.alloc();
            const old_execute = parser.execute;
            parser.socket = this;
            parser.execute = buf=>{
                parser.execute = old_execute;
                this.parser = old_parser;
                return new Error('Mux Duplex parser removed before data '+
                    'consumed');
            };
        });
        stream.allow = (bytes=Infinity)=>{
            bytes_allowed = bytes;
            if (bytes_allowed>0 && suspended)
            {
                suspended();
                suspended = undefined;
            }
        };
        // XXX vladimir: rm custom '_close' event
        // not using 'close' event due to confusion with socket close event
        // which is emitted async after handle closed
        stream.on('_close', ()=>{
            if (this.streams.delete(id))
            {
                zerr.info(`${this.ws}: vfd ${id} closed`);
                if (this.ws.zc)
                    zcounter.inc_level(`level_${this.ws.zc}_mux_vfd`, -1);
            }
        });
        this.streams.set(id, stream);
        zerr.info(`${this.ws}: vfd ${id} open`);
        if (this.ws.zc)
            zcounter.inc_level(`level_${this.ws.zc}_mux_vfd`, 1);
        return stream;
    }
    close(id){
        let stream = this.streams.get(id);
        if (!stream)
            return false;
        if (stream.destroy)
            stream.destroy();
        else
        {
            // XXX vladimir: no destroy only for embedded node v6.10.0
            stream.push(null);
            stream.end();
            stream.emit('_close');
        }
        return true;
    }
    _on_bin(buf){
        if (buf.length<8)
            return zerr(`${this.ws}: malformed binary message`);
        let id = buf.readUInt32BE(0);
        if (zerr.is.info())
            zerr.info(`${this.ws}< vfd ${id}`);
        let stream = this.streams.get(id);
        if (!stream)
            return zerr(`${this.ws}: unexpected stream ID ${id}`);
        stream.push(buf.slice(8)); // XXX vladimir: pause WS stream on false
    }
    _on_disconnected(){
        let err = new Error(this.ws.reason || 'disconnected');
        for (let stream of this.streams.values())
        {
            if (stream.destroy)
                stream.destroy(err);
            else
            {
                stream.emit('error', err);
                stream.emit('_close');
            }
        }
        this.streams.clear();
    }
}

function lib(impl){
    if (impl=='ws')
        return require('ws');
    if (impl=='uws' && !is_win)
        return require('uws');
    zerr.zexit(`WS library ${impl} is not available`);
}

function client_impl(opt){
    // WebSocket is global in react native
    if (is_rn)
        return WebSocket;
    if (!is_node && !opt.impl)
        return self.WebSocket;
    return lib(opt.impl || 'ws');
}

function server_impl(opt){
    if (!is_node)
        throw new Error(`WS server is not available`);
    return lib(opt.impl || (is_win ? 'ws' : 'uws')).Server;
}

return {Client, Server, IPC_client, IPC_server, Mux};

}); }());
