stan.js

/*
 * Copyright 2016-2018 The NATS Authors
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

/* jshint esversion: 6 */
/* jshint node: true */
'use strict';

/**
 * Module Dependencies
 */
const util = require('util'),
    nats = require('nats'),
    timers = require('timers'),
    events = require('events'),
    nuid = require('nuid'),
    url = require('url'),
    proto = require('./pb');


/**
 * Constants
 */
const VERSION = '0.2.6',
    DEFAULT_PORT = 4222,
    DEFAULT_PRE = 'nats://localhost:',
    DEFAULT_URI = DEFAULT_PRE + DEFAULT_PORT,
    DEFAULT_DISCOVER_PREFIX = '_STAN.discover',
    DEFAULT_ACK_PREFIX = '_STAN.acks',
    DEFAULT_CONNECT_WAIT = 1000 * 2,

    DEFAULT_MAX_IN_FLIGHT = 16384,
    DEFAULT_ACK_WAIT = 30 * 1000,

    BAD_SUBJECT = 'stan: subject must be supplied',
    BAD_CLUSTER_ID = 'stan: cluster ID must be supplied',
    BAD_CLIENT_ID = 'stan: client ID must be supplied',
    MAX_FLIGHT_LIMIT_REACHED = 'stan: max in flight reached.',
    CONN_CLOSED = 'stan: Connection closed',
    BAD_SUBSCRIPTION = 'stan: invalid subscription',
    BINARY_ENCODING_REQUIRED = 'stan: NATS connection encoding must be \'binary\'.',
    NO_SERVER_SUPPORT = 'stan: not supported by server',
    ACK_TIMEOUT = 'stan: publish ack timeout',
    CONNECT_REQ_TIMEOUT = 'stan: connect request timeout',
    CLOSE_REQ_TIMEOUT = 'stan: close request timeout',
    SUB_REQ_TIMEOUT = 'stan: subscribe request timeout',
    UNSUB_REQ_TIMEOUT = 'stan: unsubscribe request timeout',

    PROTOCOL_ONE = 1,
    DEFAULT_PING_INTERVAL = 5 * 1000,
    DEFAULT_PING_MAXOUT = 3,
    MAX_PINGS_EXCEEDED = 'stan: connection lost due to PING failure';


/**
 * Library Version
 * @type {string}
 */
exports.version = VERSION;


function Stan(clusterID, clientID, opts) {
    events.EventEmitter.call(this);
    if (typeof clusterID !== 'string' || clusterID.length < 1) {
        throw new Error(BAD_CLUSTER_ID);
    }
    if (typeof clientID !== 'string' || clientID.length < 1) {
        throw new Error(BAD_CLIENT_ID);
    }
    this.clusterID = clusterID;
    this.clientID = clientID;
    this.ackSubject = DEFAULT_ACK_PREFIX + "." + nuid.next(); // publish acks

    // these are set by stan
    this.pubPrefix = null; // publish prefix appended to subject
    this.subRequests = null; // subject for subscription requests
    this.unsubRequests = null; // subject for unsubscribe requests
    this.subCloseRequests = null; // subject for subscription close requests
    this.closeRequests = null; // subject for close requests

    this.parseOptions(opts);
    this.initState();
    this.createConnection();
    return this;
}


util.inherits(Stan, events.EventEmitter);

/**
 * Connect to a nats-streaming-server and return the client.
 * @param {string} clusterID
 * @param {string} [clientID] - must be unique
 * @param {object} [opts] - object with NATS/STAN options
 * @return {Stan}
 * @public
 */
exports.connect = function(clusterID, clientID, opts) {
    return new Stan(clusterID, clientID, opts);
};

/**
 * Returns true if the connection to NATS is closed.
 * @returns {boolean}
 * @private
 */
Stan.prototype.isClosed = function() {
    return this.nc === undefined;
};

/**
 * Parses the provided options
 * @param {number|string|object} opts
 * @private
 */
Stan.prototype.parseOptions = function(opts) {
    const options = this.options = {
        url: DEFAULT_URI,
        connectTimeout: DEFAULT_CONNECT_WAIT,
        ackTimeout: DEFAULT_ACK_WAIT,
        discoverPrefix: DEFAULT_DISCOVER_PREFIX,
        maxPubAcksInflight: DEFAULT_MAX_IN_FLIGHT,
        stanEncoding: 'utf8',
        stanPingInterval: DEFAULT_PING_INTERVAL,
        stanMaxPingOut: DEFAULT_PING_MAXOUT,
        maxReconnectAttempts: -1
    };

    if (opts === undefined) {
        options.url = DEFAULT_URI;
    } else if ('number' === typeof opts) {
        options.url = DEFAULT_PRE + opts;
    } else if ('string' === typeof opts) {
        options.url = sanitizeUrl(opts);
    } else if ('object' === typeof opts) {
        if (opts.port !== undefined) {
            options.url = DEFAULT_PRE + opts.port;
        }

        this.assignOption(opts, 'discoverPrefix');
        this.assignOption(opts, 'nc');
        this.assignOption(opts, 'connectTimeout');
        this.assignOption(opts, 'ackTimeout');
        this.assignOption(opts, 'maxPubAcksInflight');
        this.assignOption(opts, 'stanEncoding');
        this.assignOption(opts, 'stanPingInterval');
        this.assignOption(opts, 'stanMaxPingOut');

        // node-nats does takes a bunch of other options
        // we simply forward them, as node-nats is used
        // underneath.
        this.assignOption(opts, 'url');
        this.assignOption(opts, 'uri', 'url');
        this.assignOption(opts, 'user');
        this.assignOption(opts, 'pass');
        this.assignOption(opts, 'token');
        this.assignOption(opts, 'tokenHandler');
        this.assignOption(opts, 'password', 'pass');
        this.assignOption(opts, 'verbose');
        this.assignOption(opts, 'pedantic');
        this.assignOption(opts, 'reconnect');
        this.assignOption(opts, 'maxReconnectAttempts');
        this.assignOption(opts, 'reconnectTimeWait');
        this.assignOption(opts, 'servers');
        this.assignOption(opts, 'urls', 'servers');
        this.assignOption(opts, 'noRandomize');
        this.assignOption(opts, 'NoRandomize', 'noRandomize');
        this.assignOption(opts, 'dontRandomize', 'noRandomize');
        this.assignOption(opts, 'encoding');
        this.assignOption(opts, 'tls');
        this.assignOption(opts, 'secure', 'tls');
        this.assignOption(opts, 'name');
        this.assignOption(opts, 'client', 'name');
        this.assignOption(opts, 'yieldTime');
        this.assignOption(opts, 'waitOnFirstConnect');
        this.assignOption(opts, 'preserveBuffers');
        this.assignOption(opts, 'pingInterval');
        this.assignOption(opts, 'maxPingOut');
        this.assignOption(opts, 'useOldRequestStyle');
    }
};


function sanitizeUrl(host) {
    if ((/^.*:\/\/.*/).exec(host) === null) {
        // Does not have a scheme.
        host = 'nats://' + host;
    }
    const u = url.parse(host);
    if (u.port === null || u.port == '') {
        host += ":" + DEFAULT_PORT;
    }
    return host;
}


/**
 * Updates the internal option to the value from opts.
 * @param {object} opts
 * @param {string} prop - the property name
 * @param {string} [assign] is an alternate name for prop name in the target
 */
Stan.prototype.assignOption = function(opts, prop, assign) {
    if (assign === undefined) {
        assign = prop;
    }
    if (opts[prop] !== undefined) {
        this.options[assign] = opts[prop];
    }
};


/**
 * Internal initializer
 */
Stan.prototype.initState = function() {
    this.pubAckMap = {};
    this.pubAckOutstanding = 0;
    this.subMap = {};
};

/**
 * Connect event - emitted when the streaming protocol connection sequence  has
 * completed and the client is ready to process requests.
 *
 * @event Stan#connect
 * @type {Stan}
 */

/**
 * Close event - emitted when Stan#close() is called or its underlying NATS connection
 * closes
 *
 * @event Stan#close
 */

/**
 * Reconnecting event - emitted with the underlying NATS connection emits reconnecting
 *
 * @Event Stan#reconnecting
 */

/**
 * Error event - emitted when there's an error
 * @type {Error|object}
 *
 * Stan#error
 */

/**
 * Connect to a NATS Streaming subsystem
 * @fires Stan#connect, Stan#close, Stan#reconnecting, Stan#error
 */
Stan.prototype.createConnection = function() {
    if (typeof this.options.nc === 'object') {
        if (this.options.nc.encoding !== 'binary') {
            throw new Error(BINARY_ENCODING_REQUIRED);
        } else {
            this.nc = this.options.nc;
        }
    }
    if (this.nc === undefined) {
        const encoding = this.options.encoding;
        if (encoding && encoding !== 'binary') {
            throw new Error(BINARY_ENCODING_REQUIRED);
        } else {
            this.options.encoding = 'binary';
        }
        this.nc = nats.connect(this.options);
        this.ncOwned = true;
    }


    this.nc.on('connect', () => {
        // heartbeat processing
        const hbInbox = nats.createInbox();
        this.hbSubscription = this.nc.subscribe(hbInbox, (msg, reply) => {
            this.nc.publish(reply);
        });

        this.pingInbox = nats.createInbox();
        this.pingSubscription = this.nc.subscribe(this.pingInbox, (msg) => {
            if (msg) {
                const pingResponse = proto.pb.PingResponse.deserializeBinary(Buffer.from(msg, 'binary'));
                const err = pingResponse.getError();
                if (err) {
                    this.closeWithError('connection_lost', err);
                    return;
                }
            }
            this.pingOut = 0;
        });

        this.ackSubscription = this.nc.subscribe(this.ackSubject, this.processAck());

        const discoverSubject = this.options.discoverPrefix + '.' + this.clusterID;
        //noinspection JSUnresolvedFunction
        this.connId = Buffer.from(nuid.next(), "utf8");
        const req = new proto.pb.ConnectRequest();
        req.setClientId(this.clientID);
        req.setHeartbeatInbox(hbInbox);
        req.setProtocol(PROTOCOL_ONE);
        req.setConnId(this.connId);
        req.setPingInterval(Math.ceil(this.options.stanPingInterval / 1000));
        req.setPingMaxOut(this.options.stanMaxPingOut);


        this.nc.requestOne(discoverSubject, Buffer.from(req.serializeBinary()), this.options.connectTimeout, (msg) => {
            if (msg instanceof nats.NatsError) {
                let err = msg;
                if (msg.code === nats.REQ_TIMEOUT) {
                    err = new nats.NatsError(CONNECT_REQ_TIMEOUT, CONNECT_REQ_TIMEOUT, err);
                }
                this.closeWithError('error', err);
                return;
            }

            const cr = proto.pb.ConnectResponse.deserializeBinary(Buffer.from(msg, 'binary'));
            if (cr.getError() !== "") {
                this.closeWithError('error', cr.getError());
                return;
            }
            this.pubPrefix = cr.getPubPrefix();
            this.subRequests = cr.getSubRequests();
            this.unsubRequests = cr.getUnsubRequests();
            this.subCloseRequests = cr.getSubCloseRequests();
            this.closeRequests = cr.getCloseRequests();

            let unsubPingSub = true;
            if (cr.getProtocol() >= PROTOCOL_ONE) {
                if (cr.getPingInterval() !== 0) {
                    unsubPingSub = false;

                    this.pingRequests = cr.getPingRequests();
                    this.stanPingInterval = cr.getPingInterval() * 1000;
                    this.stanMaxPingOut = cr.getPingMaxOut();

                    const ping = new proto.pb.Ping();
                    ping.setConnId(this.connId);
                    this.pingBytes = Buffer.from(ping.serializeBinary());

                    this.pingOut = 0;
                    const that = this;
                    this.pingTimer = setTimeout(function pingFun() {
                        that.pingOut++;
                        if (that.pingOut > that.stanMaxPingOut) {
                            that.closeWithError('connection_lost', new Error(MAX_PINGS_EXCEEDED));
                            return;
                        }
                        that.nc.publish(that.pingRequests, that.pingBytes, that.pingInbox);
                        that.pingTimer = setTimeout(pingFun, that.stanPingInterval);
                    }, this.stanPingInterval);
                }
            }
            if (unsubPingSub) {
                this.nc.unsubscribe(this.pingSubscription);
                this.pingSubscription = null;
            }

            this.emit('connect', this);
        });
    });

    this.nc.on('close', () => {
        // insure we cleaned up
        this.cleanupOnClose();
        this.emit('close');
    });

    this.nc.on('disconnect', () => {
        this.emit('disconnect');
    });

    this.nc.on('reconnect', () => {
        this.emit('reconnect', this);
    });

    this.nc.on('reconnecting', () => {
        this.emit('reconnecting');
    });

    this.nc.on('error', (msg) => {
        this.emit('error', msg);
    });
};


/**
 * Close stan invoking the event notification with the
 * specified error, followed by a close notification.
 * @param event
 * @param error
 * @private
 */
Stan.prototype.closeWithError = function(event, error) {
    if (this.nc === undefined || this.clientID === undefined) {
        return;
    }
    this.cleanupOnClose(error);
    if (this.ncOwned) {
        this.nc.close();
    }
    this.emit(event, error);
    this.emit('close');
};


/**
 * Cleanup stan protocol subscriptions, pings and pending acks
 * @param err
 * @private
 */
Stan.prototype.cleanupOnClose = function(err) {
    // remove the ping timer
    if (this.pingTimer) {
        timers.clearTimeout(this.pingTimer);
        delete this.pingTimer;
    }

    // if we don't own the connection, we unsub to insure
    // that a subsequent reconnect will properly clean up.
    // Otherwise the close() will take care of it.
    if (!this.ncOwned && this.nc) {
        if (this.ackSubscription) {
            this.nc.unsubscribe(this.ackSubscription);
            this.ackSubscription = null;
        }
        if (this.pingSubscription) {
            this.nc.unsubscribe(this.pingSubscription);
            this.pingSubscription = null;
        }
        if (this.hbSubscription) {
            this.nc.unsubscribe(this.hbSubscription);
            this.hbSubscription = null;
        }
    }
    for (const guid in this.pubAckMap) {
        if (this.pubAckMap.hasOwnProperty(guid)) {
            const a = this.removeAck(guid);
            if (a && a.ah && typeof a.ah === 'function') {
                a.ah(err, guid);
            }
        }
    }
};

/**
 * Closes the NATS streaming server connection, or returns if already closed.
 * @fire Stan.close, Stan.error
 *
 */
Stan.prototype.close = function() {
    if (this.nc === undefined || this.clientID === undefined) {
        return;
    }
    this.cleanupOnClose(new Error(CONN_CLOSED));
    //noinspection JSUnresolvedFunction
    if (this.nc && this.closeRequests) {
        const req = new proto.pb.CloseRequest();
        req.setClientId(this.clientID);
        this.nc.requestOne(this.closeRequests, Buffer.from(req.serializeBinary()), {}, this.options.connectTimeout, (msgOrError) => {
            const nc = this.nc;
            delete this.nc;
            let closeError = null;
            //noinspection JSUnresolvedVariable
            if (msgOrError instanceof nats.NatsError) {
                // if we get an error here, we simply show it in the close notification as there's not much we can do here.
                closeError = msgOrError;
            } else {
                const cr = proto.pb.CloseResponse.deserializeBinary(Buffer.from(msgOrError, 'binary'));
                const err = cr.getError();
                if (err && err.length > 0) {
                    // if the protocol returned an error there's nothing for us to handle, pass it as an arg to close notification.
                    closeError = new Error(err);
                }
            }
            if (nc && this.ncOwned) {
                nc.close();
            }
            this.emit('close', closeError);
        });
    } else {
        if (this.nc && this.ncOwned) {
            this.nc.close();
            delete this.nc;
        }
        this.emit('close');
    }
};


/**
 * @return {Function} for processing acks associated with the protocol
 * @protected
 */
Stan.prototype.processAck = function() {
    return (msg) => {
        //noinspection JSUnresolvedVariable
        const pa = proto.pb.PubAck.deserializeBinary(Buffer.from(msg, 'binary'));
        const guid = pa.getGuid();
        const a = this.removeAck(guid);
        if (a && a.ah) {
            const err = pa.getError();
            a.ah(err === '' ? undefined : err, guid);
        }
    };
};

/**
 * Removes Ack for the specified guid from the outstanding ack list
 * @param {string} guid
 * @return {object}
 * @protected
 */
Stan.prototype.removeAck = function(guid) {
    const a = this.pubAckMap[guid];
    if (a !== undefined) {
        delete this.pubAckMap[guid];
        this.pubAckOutstanding--;
        if (a.t !== undefined) {
            //noinspection JSUnresolvedFunction
            timers.clearTimeout(a.t);
        }
    }
    return a;
};

/**
 * Publishes a message to the streaming server with the specified subject and data.
 * Data can be {Uint8Array|string|Buffer}. The ackHandler is called with any errors or
 * empty string, and the guid for the published message.
 *
 * Note that if the maxPubAcksInflight option is exceeded, the ackHandler will be called
 * with an error. If no ackHandler was provided, an exception is thrown.
 * @param subject
 * @param data {Uint8Array|string|Buffer}
 * @param ackHandler(err,guid)
 */
Stan.prototype.publish = function(subject, data, ackHandler) {
    if (this.nc === undefined) {
        if (util.isFunction(ackHandler)) {
            ackHandler(new Error(CONN_CLOSED));
            return;
        } else {
            throw new Error(CONN_CLOSED);
        }
    }

    if (this.pubAckOutstanding > this.options.maxPubAcksInflight) {
        // we have many pending publish messages, fail it.
        if (util.isFunction(ackHandler)) {
            ackHandler(new Error(MAX_FLIGHT_LIMIT_REACHED));
        } else {
            throw new Error(MAX_FLIGHT_LIMIT_REACHED);
        }
    }

    const subj = this.pubPrefix + '.' + subject;
    const peGUID = nuid.next();
    //noinspection JSUnresolvedFunction
    const pe = new proto.pb.PubMsg();
    pe.setClientId(this.clientID);
    pe.setConnId(this.connId);
    pe.setGuid(peGUID);
    pe.setSubject(subject);
    let buf;
    if (typeof data === 'string') {
        buf = Buffer.from(data, 'utf8');
        data = new Uint8Array(buf);
    } else if (Buffer.prototype.isPrototypeOf(data)) {
        buf = Buffer.from(data, 'utf8');
        data = new Uint8Array(buf);
    } else if (Buffer.prototype.isPrototypeOf(Uint8Array)) {
        // we already handle this
    }

    pe.setData(data);

    const ack = {};
    ack.ah = ackHandler;
    this.pubAckMap[peGUID] = ack;

    const bytes = Buffer.from(pe.serializeBinary());
    this.nc.publish(subj, bytes, this.ackSubject);
    this.pubAckOutstanding++;

    // all acks are received in ackSubject, so not possible to reuse nats.timeout
    //noinspection JSUnresolvedFunction
    ack.t = timers.setTimeout(() => {
        const a = this.removeAck(peGUID);
        if (a && a.ah !== undefined) {
            a.ah(new Error(ACK_TIMEOUT), peGUID);
        }
    }, this.options.ackTimeout);

    return peGUID;
};


/**
 * Creates a NATS streaming server subscription on the specified subject. If qGroup
 * is provided, the subscription will be distributed between all subscribers using
 * the same qGroup name.
 * @param {String} subject
 * @param {String} [qGroup]
 * @param {SubscriptionOptions} [options]
 * @throws err if the subject is not provided
 * @fires Stan#error({Error})
 * @returns Subscription
 */
Stan.prototype.subscribe = function(subject, qGroup, options) {
    const args = {};
    if (typeof qGroup === 'string') {
        args.qGroup = qGroup;
    } else if (typeof qGroup === 'object') {
        args.options = qGroup;
    }
    if (typeof options === 'object') {
        args.options = options;
    }
    if (!args.options) {
        args.options = new SubscriptionOptions();
    }

    // in node-nats there's no Subscription object...
    const retVal = new Subscription(this, subject, args.qGroup, nats.createInbox(), args.options, args.callback);

    if (typeof subject !== 'string' || subject.length === 0) {
        process.nextTick(() => {
            retVal.emit('error', new Error(BAD_SUBJECT));
        });
        return retVal;
    }

    if (this.isClosed()) {
        process.nextTick(() => {
            retVal.emit('error', new Error(CONN_CLOSED));
        });
        return retVal;
    }

    this.subMap[retVal.inbox] = retVal;
    retVal.inboxSub = this.nc.subscribe(retVal.inbox, this.processMsg());
    const sr = new proto.pb.SubscriptionRequest();
    sr.setClientId(this.clientID);
    sr.setSubject(subject);
    sr.setQGroup(retVal.qGroup || '');
    sr.setInbox(retVal.inbox);
    sr.setMaxInFlight(retVal.opts.maxInFlight);
    sr.setAckWaitInSecs(retVal.opts.ackWait / 1000);
    sr.setStartPosition(retVal.opts.startPosition);
    sr.setDurableName(retVal.opts.durableName || '');

    switch (sr.getStartPosition()) {
        case proto.pb.StartPosition.TIME_DELTA_START:
            sr.setStartTimeDelta(retVal.opts.startTime);
            break;
        case proto.pb.StartPosition.SEQUENCE_START:
            sr.setStartSequence(retVal.opts.startSequence);
            break;
    }

    this.nc.requestOne(this.subRequests, Buffer.from(sr.serializeBinary()), this.options.connectTimeout, (msg) => {
        if (msg instanceof nats.NatsError) {
            if (msg.code === nats.REQ_TIMEOUT) {
                const err = new nats.NatsError(SUB_REQ_TIMEOUT, SUB_REQ_TIMEOUT, msg);
                retVal.emit('timeout', err);
            } else {
                retVal.emit('error', msg);
            }
            return;
        }
        //noinspection JSUnresolvedVariable
        const r = proto.pb.SubscriptionResponse.deserializeBinary(Buffer.from(msg, 'binary'));
        const err = r.getError();
        if (err && err.length !== 0) {
            retVal.emit('error', new Error(err));
            this.nc.unsubscribe(retVal.inboxSub);
            retVal.emit('unsubscribed');
            return;
        }
        retVal.ackInbox = r.getAckInbox();
        retVal.emit('ready');
    });

    return retVal;
};

/**
 * A NATS streaming subscription is an {event.EventEmitter} representing a subscription to the
 * server. The subscription will be ready to receive messages after the Subscription#ready notification.
 * fires. Messages are delivered on the Subscription#message(msg) notificatication.
 * @param stanConnection
 * @param subject
 * @param qGroup
 * @param inbox
 * @param opts
 * @constructor
 * @fires Subscription#error({Error}), Subscription#unsubscribed, Subscription#ready, Subscription#timeout({Error})
 *    Subscription#message({Message})
 */
function Subscription(stanConnection, subject, qGroup, inbox, opts) {
    this.stanConnection = stanConnection;
    this.subject = subject;
    this.qGroup = qGroup;
    this.inbox = inbox;
    this.opts = opts;
    this.ackInbox = undefined;
    this.inboxSub = undefined;
}

/**
 * Error event - if there's an error with setting up the subscription, such
 * as the connection is closed or the server returns an error.
 *
 * @event Subscription#Error
 * @type Error
 */

/**
 * Timeout event - An error notification indicating that the operation timeout.
 *
 * @event Subscription#Timeout
 * @type Error
 */

/**
 * Unsubscribed event - notification that the unsubscribe request was processed by the server
 *
 * @event Subscription#unsubscribed
 */

/**
 * Ready event - notification that the subscription request was processed by the server
 *
 * @event Subscription#ready
 */

/**
 * Message event - notification that the subscription received a message from the server
 * @event Subscription#message
 * @type {Message}
 */

util.inherits(Subscription, events.EventEmitter);

/**
 * Returns true if the subscription has been closed or unsubscribed from.
 * @returns {boolean}
 */
Subscription.prototype.isClosed = function() {
    return this.stanConnection === undefined;
};

/**
 * Unregisters the subscription from the streaming server. You cannot unsubscribe
 * from the server unless the Subscription#ready notification has already fired.
 * @fires Subscription#error({Error}, Subscription#unsubscribed, Subscription#timeout({Error}
 */
Subscription.prototype.unsubscribe = function() {
    this.closeOrUnsubscribe(false);
};


/**
 * Close removes the subscriber from the server, but unlike the Subscription#unsubscribe(),
 * the durable interest is not removed. If the client has connected to a server
 * for which this feature is not available, Subscription#Close() will emit a
 * Subscription#error(NO_SERVER_SUPPORT) error. Note that this affects durable clients only.
 * If called on a non-durable subscriber, this is equivalent to Subscription#close()
 *
 * @fires Subscription#error({Error}, Subscription#closed
 */
Subscription.prototype.close = function() {
    this.closeOrUnsubscribe(true);
};


/**
 * @param doClose
 * @private
 */
Subscription.prototype.closeOrUnsubscribe = function(doClose) {
    if (this.isClosed()) {
        this.emit('error', new Error(BAD_SUBSCRIPTION));
        return;
    }

    const sc = this.stanConnection;
    delete this.stanConnection;
    delete sc.subMap[this.inbox];

    if (sc.isClosed()) {
        this.emit('error', new Error(CONN_CLOSED));
        return;
    }

    let reqSubject = sc.unsubRequests;
    if (doClose) {
        reqSubject = sc.subCloseRequests;
        if (!reqSubject) {
            this.emit('error', new Error(NO_SERVER_SUPPORT));
        }
    }

    sc.nc.unsubscribe(this.inboxSub);
    //noinspection JSUnresolvedFunction
    const ur = new proto.pb.UnsubscribeRequest();
    ur.setClientId(sc.clientID);
    ur.setSubject(this.subject);
    ur.setInbox(this.ackInbox);

    sc.nc.requestOne(reqSubject, Buffer.from(ur.serializeBinary()), sc.options.connectTimeout, (msg) => {
        let err;
        if (msg instanceof nats.NatsError) {
            const type = doClose ? CLOSE_REQ_TIMEOUT : UNSUB_REQ_TIMEOUT;
            err = new nats.NatsError(type, type, msg);
            if (msg.code === nats.REQ_TIMEOUT) {
                this.emit('timeout', err);
            } else {
                this.emit('error', err);
            }
            return;
        }
        //noinspection JSUnresolvedVariable
        const r = proto.pb.SubscriptionResponse.deserializeBinary(Buffer.from(msg, 'binary'));
        err = r.getError();
        if (err && err.length > 0) {
            this.emit('error', new Error(r.getError()));
        } else {
            this.emit(doClose ? 'closed' : 'unsubscribed');
        }
    });
};


/**
 * Internal function to process in-bound messages.
 * @return {Function}
 * @private
 */
Stan.prototype.processMsg = function() {
    // curry
    return (rawMsg, reply, subject, sid) => {
        const sub = this.subMap[subject];
        try {
            //noinspection JSUnresolvedVariable
            const m = proto.pb.MsgProto.deserializeBinary(Buffer.from(rawMsg, 'binary'));
            if (sub === undefined || !this.nc) {
                return;
            }
            const msg = new Message(this, m, sub);
            sub.emit('message', msg);
            msg.maybeAutoAck();
        } catch (error) {
            sub.emit('error', error);
        }
    };
};

/**
 * Represents a message received from the streaming server.
 * @param stanClient
 * @param msg
 * @param subscription
 * @constructor
 */
function Message(stanClient, msg, subscription) {
    this.stanClient = stanClient;
    this.msg = msg;
    this.subscription = subscription;
}

/**
 * Returns the sequence number of the message.
 * @returns {number}
 */
Message.prototype.getSequence = function() {
    return this.msg.getSequence();
};

/**
 * Returns the subject the message was published on
 * @returns {string}
 */
Message.prototype.getSubject = function() {
    return this.msg.getSubject();
};

/**
 * Returns a Buffer object with the raw message payload.
 * @returns {Buffer}
 */
Message.prototype.getRawData = function() {
    return Buffer.from(this.msg.getData(), 'binary');
};

/**
 * Convenience API to convert the results of Message#getRawData to
 * a string with the specified 'stanEncoding'. Note that if the encoding
 * is set to binary, this method returns Message#getRawData.
 * @returns {!(string|Uint8Array)|string}
 */
Message.prototype.getData = function() {
    let bytes = this.msg.getData();
    const encoding = this.stanClient.options.stanEncoding;
    if (encoding !== 'binary') {
        bytes = bytes.length > 0 ? Buffer.from(bytes, encoding).toString() : '';
    }
    return bytes;
};

/**
 * Returns the raw timestamp. The NATS streaming server returns a 64bit nanosecond resolution
 * timestamp that is not quite useful in JavaScript. Use Message#getTimestamp to read
 * a timestamp as a Date.
 * @returns {number}
 */
Message.prototype.getTimestampRaw = function() {
    return this.msg.getTimestamp();
};

/**
 * Returns Message#getTimestampRaw as a JavaScript Date.
 * @returns {Date}
 */
Message.prototype.getTimestamp = function() {
    return new Date(this.getTimestampRaw() / 1000000);
};

/**
 * Returns true if this message is being redelivered.
 * @returns {boolean}
 */
Message.prototype.isRedelivered = function() {
    return this.msg.getRedelivered();
};

/**
 * Returns the CRC32 of the message if provided.
 * @returns {number}
 */
Message.prototype.getCrc32 = function() {
    return this.msg.getCrc32();
};

/**
 * Calls Message.ack if the subscription was specified to
 * use manualAcks.
 * @type {Message.ack}
 * @protected
 */
Message.prototype.maybeAutoAck = function() {
    if (!this.subscription.opts.manualAcks) {
        this.ack();
    }
};

/**
 * Acks the message, note this method shouldn't be called unless
 * the manualAcks option was set on the subscription.
 */
Message.prototype.ack = function() {
    if (!this.subscription.isClosed()) {
        const ack = new proto.pb.Ack();
        ack.setSubject(this.getSubject());
        ack.setSequence(this.getSequence());
        this.stanClient.nc.publish(this.subscription.ackInbox, Buffer.from(ack.serializeBinary()));
    }
};

/**
 *
 * @returns {!(string|Uint8Array)}
 */
Message.prototype.getClientID = function() {
    return this.msg.getConnId();
};

Message.prototype.getConnectionID = function() {
    return this.msg.getClientId();
};


/**
 * Returns an object with various constants for StartPosition (NEW_ONLY,
 * LAST_RECEIVED, TIME_DELTA_START, SEQUENCE_START, FIRST)
 * @type {StartPosition}
 */
exports.StartPosition = proto.pb.StartPosition;

function SubscriptionOptions(durableName, maxInFlight, ackWait, startPosition, startSequence, startTime, manualAcks) {
    // DurableName, if set will survive client restarts.
    this.durableName = durableName;
    // Controls the number of messages the cluster will have inflight without an ACK.
    this.maxInFlight = maxInFlight || DEFAULT_MAX_IN_FLIGHT;
    // Controls the time the cluster will wait for an ACK for a given message.
    this.ackWait = ackWait || DEFAULT_ACK_WAIT;
    // StartPosition enum from proto.pb
    this.startPosition = startPosition;
    // Optional start sequence number.
    this.startSequence = startSequence;
    // Optional start time.
    this.startTime = startTime;
    // Option to do Manual Acks
    this.manualAcks = manualAcks;
}

/**
 * Returns a SubscriptionOptions initialized to the defaults
 * @return {SubscriptionOptions}
 */
Stan.prototype.subscriptionOptions = function() {
    return new SubscriptionOptions();
};

/**
 * @param n
 * @returns {SubscriptionOptions}
 */
SubscriptionOptions.prototype.setMaxInFlight = function(n) {
    this.maxInFlight = n;
    return this;
};

SubscriptionOptions.prototype.setAckWait = function(millis) {
    this.ackWait = millis;
    return this;
};

SubscriptionOptions.prototype.setStartAt = function(startPosition) {
    this.startPosition = startPosition;
    return this;
};

SubscriptionOptions.prototype.setStartAtSequence = function(sequence) {
    this.startPosition = proto.pb.StartPosition.SEQUENCE_START;
    this.startSequence = sequence;
    return this;
};


/**
 * @param {Date} date
 * @return {SubscriptionOptions}
 */
SubscriptionOptions.prototype.setStartTime = function(date) {
    this.startPosition = proto.pb.StartPosition.TIME_DELTA_START;
    // server expects values in ns
    this.startTime = (Date.now() - date.valueOf()) * 1000000;
    return this;
};

/**
 * @param {Number} millis
 * @return {SubscriptionOptions}
 */
SubscriptionOptions.prototype.setStartAtTimeDelta = function(millis) {
    this.startPosition = proto.pb.StartPosition.TIME_DELTA_START;
    //noinspection JSUnresolvedFunction
    // server expects values in ns
    this.startTime = millis * 1000000;
    return this;
};


/**
 * @return {SubscriptionOptions}
 */
SubscriptionOptions.prototype.setStartWithLastReceived = function() {
    this.startPosition = proto.pb.StartPosition.LAST_RECEIVED;
    return this;
};


/**
 * @return {SubscriptionOptions}
 */
SubscriptionOptions.prototype.setDeliverAllAvailable = function() {
    this.startPosition = proto.pb.StartPosition.FIRST;
    return this;
};


/**
 * @return {SubscriptionOptions}
 */
SubscriptionOptions.prototype.setManualAckMode = function(tf) {
    this.manualAcks = tf;
    return this;
};


/**
 * @param {String} durableName
 * @return {SubscriptionOptions}
 */
SubscriptionOptions.prototype.setDurableName = function(durableName) {
    this.durableName = durableName;
    return this;
};