server_subscription.js

"use strict";
/**
 * @module opcua.server
 */


const Dequeue = require("dequeue");

const subscription_service = require("node-opcua-service-subscription");
const DataChangeNotification = subscription_service.DataChangeNotification;
const EventNotificationList = subscription_service.EventNotificationList;
const NotificationMessage = subscription_service.NotificationMessage;
const StatusChangeNotification = subscription_service.StatusChangeNotification;
const MonitoringMode = subscription_service.MonitoringMode;
const NodeId = require("node-opcua-nodeid").NodeId;
const StatusCodes = require("node-opcua-status-code").StatusCodes;
const Enum = require("node-opcua-enum").Enum;
const assert = require("node-opcua-assert").assert;
const NodeClass = require("node-opcua-data-model").NodeClass;
const _ = require("underscore");

const AttributeIds = require("node-opcua-data-model").AttributeIds;

const SequenceNumberGenerator = require("node-opcua-secure-channel").SequenceNumberGenerator;

const EventEmitter = require("events").EventEmitter;
const util = require("util");

const SessionContext = require("node-opcua-address-space").SessionContext;

const EventFilter = require("node-opcua-service-filter").EventFilter;
const DataChangeFilter = require("node-opcua-service-subscription").DataChangeFilter;
const AggregateFilter = require("node-opcua-service-subscription").AggregateFilter;

const UAVariable = require("node-opcua-address-space").UAVariable;
const validateFilter = require("./validate_filter").validateFilter;
const isValidDataEncoding = require("node-opcua-data-model").isValidDataEncoding;


const debugLog = require("node-opcua-debug").make_debugLog(__filename);
const doDebug = require("node-opcua-debug").checkDebugFlag(__filename);


const SubscriptionState = new Enum([
    "CLOSED",   // The Subscription has not yet been created or has terminated.
    "CREATING", // The Subscription is being created
    "NORMAL",   // The Subscription is cyclically checking for Notifications from its MonitoredItems.
                // The keep-alive counter is not used in this state.
    "LATE",     // The publishing timer has expired and there are Notifications available or a keep-alive Message is
                // ready to be sent, but there are no Publish requests queued. When in this state, the next Publish
                // request is processed when it is received. The keep-alive counter is not used in this state.
    "KEEPALIVE",// The Subscription is cyclically checking for Notification
                // alive counter to count down to 0 from its maximum.
    "TERMINATED"
]);
exports.SubscriptionState = SubscriptionState;


const SessionDiagnosticsDataType = require("node-opcua-common").SessionDiagnosticsDataType;


function _adjust_publishing_interval(publishingInterval) {
    publishingInterval = publishingInterval || Subscription.defaultPublishingInterval;
    publishingInterval = Math.max(publishingInterval, Subscription.minimumPublishingInterval);
    publishingInterval = Math.min(publishingInterval, Subscription.maximumPublishingInterval);
    return publishingInterval;
}

const minimumMaxKeepAliveCount = 2;
const maximumMaxKeepAliveCount = 12000;

function _adjust_maxKeepAliveCount(maxKeepAliveCount/*,publishingInterval*/) {
    maxKeepAliveCount = maxKeepAliveCount || minimumMaxKeepAliveCount;
    maxKeepAliveCount = Math.max(maxKeepAliveCount, minimumMaxKeepAliveCount);
    maxKeepAliveCount = Math.min(maxKeepAliveCount, maximumMaxKeepAliveCount);
    return maxKeepAliveCount;
}

function _adjust_lifeTimeCount(lifeTimeCount, maxKeepAliveCount, publishingInterval) {
    lifeTimeCount = lifeTimeCount || 1;

    // let's make sure that lifeTimeCount is at least three time maxKeepAliveCount
    // Note : the specs say ( part 3  - CreateSubscriptionParameter )
    //        "The lifetime count shall be a minimum of three times the keep keep-alive count."
    lifeTimeCount = Math.max(lifeTimeCount, maxKeepAliveCount * 3);

    const minTicks = Math.ceil(5 * 1000 / (publishingInterval)); // we want 5 seconds min

    lifeTimeCount = Math.max(minTicks, lifeTimeCount);
    return lifeTimeCount;
}

function _adjust_publishinEnable(publishingEnabled) {
    return (publishingEnabled === null || publishingEnabled === undefined) ? true : !!publishingEnabled;
}

function _adjust_maxNotificationsPerPublish(maxNotificationsPerPublish) {
    maxNotificationsPerPublish += 0;
    assert(_.isNumber(maxNotificationsPerPublish));
    return (maxNotificationsPerPublish >= 0) ? maxNotificationsPerPublish : 0;
}

// verify that the injected publishEngine provides the expected services
// regarding the Subscription requirements...
function _assert_valid_publish_engine(publishEngine) {
    assert(_.isObject(publishEngine));
    assert(_.isNumber(publishEngine.pendingPublishRequestCount));
    assert(_.isFunction(publishEngine.send_notification_message));
    assert(_.isFunction(publishEngine.send_keep_alive_response));
    assert(_.isFunction(publishEngine.on_close_subscription));
}

const SubscriptionDiagnosticsDataType = require("node-opcua-common").SubscriptionDiagnosticsDataType;

function createSubscriptionDiagnostics(self) {

    assert(self instanceof Subscription);

    self.subscriptionDiagnostics = new SubscriptionDiagnosticsDataType({});

    self.subscriptionDiagnostics.$subscription = self;
    // "sessionId"
    self.subscriptionDiagnostics.__defineGetter__("sessionId", function () {
      return this.$subscription.getSessionId();
    });
    self.subscriptionDiagnostics.__defineGetter__("subscriptionId", function () {
        return this.$subscription.id;
    });
    self.subscriptionDiagnostics.__defineGetter__("priority", function () {
        return this.$subscription.priority;
    });
    self.subscriptionDiagnostics.__defineGetter__("publishingInterval", function () {
        return this.$subscription.publishingInterval;
    });
    self.subscriptionDiagnostics.__defineGetter__("maxLifetimeCount", function () {
        return this.$subscription.lifeTimeCount;
    });
    self.subscriptionDiagnostics.__defineGetter__("maxKeepAliveCount", function () {
        return this.$subscription.maxKeepAliveCount;
    });
    self.subscriptionDiagnostics.__defineGetter__("maxNotificationsPerPublish", function () {
        return this.$subscription.maxNotificationsPerPublish;
    });
    self.subscriptionDiagnostics.__defineGetter__("publishingEnabled", function () {
        return this.$subscription.publishingEnabled;
    });
    self.subscriptionDiagnostics.__defineGetter__("monitoredItemCount", function () {
        return this.$subscription.monitoredItemCount;
    });
    self.subscriptionDiagnostics.__defineGetter__("nextSequenceNumber", function () {
        return this.$subscription._get_future_sequence_number();
    });
    self.subscriptionDiagnostics.__defineGetter__("disabledMonitoredItemCount", function () {
        return this.$subscription.disabledMonitoredItemCount;
    });

    /* those member of self.subscriptionDiagnostics are handled directly

     modifyCount
     enableCount,
     disableCount,
     republishRequestCount,
     notificationsCount,
     publishRequestCount,
     dataChangeNotificationsCount,
     eventNotificationsCount,
    */

    /*
     those members are not updated yet in the code :
     "republishMessageRequestCount",
     "republishMessageCount",
     "transferRequestCount",
     "transferredToAltClientCount",
     "transferredToSameClientCount",
     "latePublishRequestCount",
     "currentKeepAliveCount",
     "currentLifetimeCount",
     "unacknowledgedMessageCount",
     "discardedMessageCount",
     "monitoringQueueOverflowCount",
     "eventQueueOverFlowCount"
     */
    // add object in Variable SubscriptionDiagnosticArray (i=2290) ( Array of SubscriptionDiagnostics)
    // add properties in Variable to reflect
}

/**
 * The Subscription class used in the OPCUA server side.
 * @class Subscription
 * @param {Object} options
 * @param options.id {Integer} - a unique identifier
 * @param options.publishingInterval {Integer} - [optional](default:1000) the publishing interval.
 * @param options.maxKeepAliveCount  {Integer} - [optional](default:10) the max KeepAlive Count.
 * @param options.lifeTimeCount      {Integer} - [optional](default:10) the max Life Time Count
 * @param options.publishingEnabled  {Boolean} - [optional](default:true)
 * @param options.sessionId          {NodeId}  - [optional]
 * @param options.maxNotificationsPerPublish {Integer} - [optional](default:0)
 * @param options.priority {Byte}
 * @constructor
 */
function Subscription(options) {

    options = options || {};

    EventEmitter.apply(this, arguments);


    const subscription = this;

    Subscription.registry.register(subscription);

    subscription.sessionId = options.sessionId|| NodeId.nullNodeId;
    assert(subscription.sessionId instanceof NodeId,"expecting a sessionId NodeId");

    subscription.publishEngine = options.publishEngine;
    _assert_valid_publish_engine(subscription.publishEngine);

    subscription.id = options.id || "<invalid_id>";

    subscription.priority = options.priority || 0;


    /**
     * the Subscription publishing interval
     * @property  publishingInterval
     * @type {number}
     * @default 1000
     */
    subscription.publishingInterval = _adjust_publishing_interval(options.publishingInterval);

    /**
     * The keep alive count defines how many times the publish interval need to
     * expires without having notifications available before the server send an
     * empty message.
     * OPCUA Spec says: a value of 0 is invalid.
     * @property  maxKeepAliveCount
     * @type {number}
     * @default 10
     *
     */
    subscription.maxKeepAliveCount = _adjust_maxKeepAliveCount(options.maxKeepAliveCount, subscription.publishingInterval);

    subscription.resetKeepAliveCounter();

    /**
     * The life time count defines how many times the publish interval expires without
     * having a connection to the client to deliver data.
     * If the life time count reaches maxKeepAliveCount, the subscription will
     * automatically terminate.
     * OPCUA Spec: The life-time count shall be a minimum of three times the keep keep-alive count.
     *
     * Note: this has to be interpreted as without having a PublishRequest available
     * @property  lifeTimeCount
     * @type {Number}
     * @default 1
     */
    subscription.lifeTimeCount = _adjust_lifeTimeCount(options.lifeTimeCount, subscription.maxKeepAliveCount, subscription.publishingInterval);


    /**
     * The maximum number of notifications that the Client wishes to receive in a
     * single Publish response. A value of zero indicates that there is no limit.
     * The number of notifications per Publish is the sum of monitoredItems in the
     * DataChangeNotification and events in the EventNotificationList.
     *
     * @property maxNotificationsPerPublish
     * @type {Number}
     * #default 0
     */
    subscription.maxNotificationsPerPublish = _adjust_maxNotificationsPerPublish(options.maxNotificationsPerPublish);

    subscription._life_time_counter = 0;
    subscription.resetLifeTimeCounter();

    // notification message that are ready to be sent to the client
    subscription._pending_notifications = new Dequeue();

    subscription._sent_notifications = [];

    subscription._sequence_number_generator = new SequenceNumberGenerator();

    // initial state of the subscription
    subscription.state = SubscriptionState.CREATING;

    subscription.publishIntervalCount = 0;

    subscription.monitoredItems = {}; // monitored item map

    /**
     *  number of monitored Item
     *  @property monitoredItemIdCounter
     *  @type {Number}
     */
    subscription.monitoredItemIdCounter = 0;

    subscription.publishingEnabled = _adjust_publishinEnable(options.publishingEnabled);

    subscription.subscriptionDiagnostics = null;
    createSubscriptionDiagnostics(subscription);

    // A boolean value that is set to TRUE to mean that either a NotificationMessage or a keep-alive Message has been
    // sent on the Subscription. It is a flag that is used to ensure that either a NotificationMessage or a keep-alive
    // Message is sent out the first time the publishing timer expires.
    subscription.messageSent = false;

    subscription._unacknowledgedMessageCount = 0;

    subscription.timerId = null;
    subscription._start_timer();

}

util.inherits(Subscription, EventEmitter);


Subscription.minimumPublishingInterval = 50;  // fastest possible
Subscription.defaultPublishingInterval = 1000; // one second
Subscription.maximumPublishingInterval = 1000 * 60 * 60 * 24 * 15; // 15 days
assert(Subscription.maximumPublishingInterval < 2147483647, "maximumPublishingInterval cannot exceed (2**31-1) ms ");


const ObjectRegistry = require("node-opcua-object-registry").ObjectRegistry;
Subscription.registry = new ObjectRegistry();

Subscription.prototype.getSessionId = function () {
    const subscription = this;
    return subscription.sessionId;
};

Subscription.prototype.toString = function () {

    const subscription = this;
    let str = "Subscription:\n";
    str += "  subscriptionId          " + subscription.id + "\n";
    str += "  sessionId          " + subscription.getSessionId().toString() + "\n";

    str += "  publishingEnabled  " + subscription.publishingEnabled + "\n";
    str += "  maxKeepAliveCount  " + subscription.maxKeepAliveCount + "\n";
    str += "  publishingInterval " + subscription.publishingInterval + "\n";
    str += "  lifeTimeCount      " + subscription.lifeTimeCount + "\n";
    str += "  maxKeepAliveCount  " + subscription.maxKeepAliveCount + "\n";
    return str;
};

/**
 * @method modify
 * @param param {Object}
 * @param param.requestedPublishingInterval  {Duration}  requestedPublishingInterval =0 means fastest possible
 * @param param.requestedLifetimeCount       {Counter}   requestedLifetimeCount      ===0 means no change
 * @param param.requestedMaxKeepAliveCount   {Counter}   requestedMaxKeepAliveCount  ===0 means no change
 * @param param.maxNotificationsPerPublish   {Counter}
 * @param param.priority                     {Byte}
 *
 */
Subscription.prototype.modify = function (param) {
    const subscription = this;

    // update diagnostic counter
    subscription.subscriptionDiagnostics.modifyCount += 1;

    const publishingInterval_old = subscription.publishingInterval;

    param.requestedPublishingInterval = param.requestedPublishingInterval || 0;
    param.requestedMaxKeepAliveCount = param.requestedMaxKeepAliveCount || subscription.maxKeepAliveCount;
    param.requestedLifetimeCount = param.requestedLifetimeCount || subscription.lifeTimeCount;

    subscription.publishingInterval = _adjust_publishing_interval(param.requestedPublishingInterval);
    subscription.maxKeepAliveCount = _adjust_maxKeepAliveCount(param.requestedMaxKeepAliveCount, subscription.publishingInterval);
    subscription.lifeTimeCount = _adjust_lifeTimeCount(param.requestedLifetimeCount, subscription.maxKeepAliveCount, subscription.publishingInterval);

    subscription.maxNotificationsPerPublish = param.maxNotificationsPerPublish;
    subscription.priority = param.priority;

    subscription.resetLifeTimeAndKeepAliveCounters();

    if (publishingInterval_old !== subscription.publishingInterval) {
        // todo
    }
    subscription._stop_timer();
    subscription._start_timer();

};

Subscription.prototype._stop_timer = function () {
    const subscription = this;
    if (subscription.timerId) {
        debugLog("Subscription#_stop_timer subscriptionId=".bgWhite.blue, subscription.id);
        clearInterval(subscription.timerId);
        subscription.timerId = null;
    }
};


Subscription.prototype._start_timer = function () {

    const subscription = this;
    debugLog("Subscription#_start_timer  subscriptionId=".bgWhite.blue, subscription.id, " publishingInterval = ", subscription.publishingInterval);

    assert(subscription.timerId === null);
    // from the spec:
    // When a Subscription is created, the first Message is sent at the end of the first publishing cycle to
    // inform the Client that the Subscription is operational. A NotificationMessage is sent if there are
    // Notifications ready to be reported. If there are none, a keep-alive Message is sent instead that
    // contains a sequence number of 1, indicating that the first NotificationMessage has not yet been sent.
    // This is the only time a keep-alive Message is sent without waiting for the maximum keep-alive count
    // to be reached, as specified in (f) above.


    // make sure that a keep-alive Message will be send at the end of the first publishing cycle
    // if there are no Notifications ready.
    subscription._keep_alive_counter = subscription.maxKeepAliveCount;

    assert(subscription.publishingInterval >= Subscription.minimumPublishingInterval);
    subscription.timerId = setInterval(subscription._tick.bind(subscription), subscription.publishingInterval);
};

// counter
Subscription.prototype._get_next_sequence_number = function () {
    return this._sequence_number_generator  ? this._sequence_number_generator.next() : 0;
};

// counter
Subscription.prototype._get_future_sequence_number = function () {
    return this._sequence_number_generator ? this._sequence_number_generator.future() : 0;
};

Subscription.prototype.setPublishingMode = function (publishingEnabled) {

    this.publishingEnabled = !!publishingEnabled;
    // update diagnostics
    if (this.publishingEnabled) {
        this.subscriptionDiagnostics.enableCount += 1;
    } else {
        this.subscriptionDiagnostics.disableCount += 1;
    }

    this.resetLifeTimeCounter();

    if (!publishingEnabled && this.state !== SubscriptionState.CLOSED) {
        this.state = SubscriptionState.NORMAL;
    }
    return StatusCodes.Good;
};


/**
 *  _publish_pending_notifications send a "notification" event:
 *
 * @method _publish_pending_notifications *
 * @private
 *
 */
Subscription.prototype._publish_pending_notifications = function () {

    const subscription = this;
    const publishEngine = subscription.publishEngine;
    const subscriptionId = subscription.id;

    // preconditions
    assert(publishEngine.pendingPublishRequestCount > 0);
    assert(subscription.hasPendingNotifications);

    function _count_notification_message(notifData) {

        if (notifData instanceof DataChangeNotification) {
            subscription.subscriptionDiagnostics.dataChangeNotificationsCount += 1;
        } else if (notifData instanceof EventNotificationList) {
            subscription.subscriptionDiagnostics.eventNotificationsCount += 1;
        } else {
            // TODO
        }
    }

    // todo : get rid of this....
    subscription.emit("notification");

    const notificationMessage = subscription._popNotificationToSend().notification;

    subscription.emit("notificationMessage", notificationMessage);


    assert(_.isArray(notificationMessage.notificationData));

    notificationMessage.notificationData.forEach(_count_notification_message);


    assert(notificationMessage.hasOwnProperty("sequenceNumber"));
    assert(notificationMessage.hasOwnProperty("notificationData"));

    const moreNotifications = (subscription.hasPendingNotifications);

    // update diagnostics
    if (subscription.subscriptionDiagnostics) {
        subscription.subscriptionDiagnostics.notificationsCount += 1;
        subscription.subscriptionDiagnostics.publishRequestCount += 1;
    }

    publishEngine.send_notification_message({
        subscriptionId: subscriptionId,
        sequenceNumber: notificationMessage.sequenceNumber,
        notificationData: notificationMessage.notificationData,
        moreNotifications: moreNotifications
    }, false);
    subscription.messageSent = true;
    subscription._unacknowledgedMessageCount++;

    subscription.resetLifeTimeAndKeepAliveCounters();

    if (doDebug) {
        debugLog("Subscription sending a notificationMessage subscriptionId=", subscriptionId,
            "sequenceNumber = ", notificationMessage.sequenceNumber.toString());
        // debugLog(notificationMessage.toString());
    }

    if (subscription.state !== SubscriptionState.CLOSED) {
        assert(notificationMessage.notificationData.length > 0, "We are not expecting a keep-alive message here");
        subscription.state = SubscriptionState.NORMAL;
        debugLog("subscription " + subscription.id + " set to NORMAL".bgYellow);
    }

};

Subscription.prototype._process_keepAlive = function () {
    const subscription = this;

    //xx assert(!self.publishingEnabled || (!self.hasPendingNotifications && !self.hasMonitoredItemNotifications));

    subscription.increaseKeepAliveCounter();

    if (subscription.keepAliveCounterHasExpired) {

        if (subscription._sendKeepAliveResponse()) {

            subscription.resetLifeTimeAndKeepAliveCounters();

        } else {
            debugLog("     -> subscription.state === LATE , because keepAlive Response cannot be send due to lack of PublishRequest");
            subscription.state = SubscriptionState.LATE;
        }
    }
};


Subscription.prototype.process_subscription = function () {

    const subscription = this;

    assert(subscription.publishEngine.pendingPublishRequestCount > 0);

    if (!subscription.publishingEnabled) {
        // no publish to do, except keep alive
        subscription._process_keepAlive();
        return;
    }

    if (!subscription.hasPendingNotifications && subscription.hasMonitoredItemNotifications) {
        // collect notification from monitored items
        subscription._harvestMonitoredItems();
    }

    // let process them first
    if (subscription.hasPendingNotifications) {

        subscription._publish_pending_notifications();

        if (subscription.state === SubscriptionState.NORMAL && subscription.hasPendingNotifications) {

            // istanbul ignore next
            if (doDebug) {
                debugLog("    -> pendingPublishRequestCount > 0 && normal state => re-trigger tick event immediately ");
            }

            // let process an new publish request
            setImmediate(subscription._tick.bind(subscription));
        }

    } else {
        subscription._process_keepAlive();
    }
};

function w(s, w) {
    return ("000" + s).substr(-w);
}

function t(d) {
    return w(d.getHours(), 2) + ":" + w(d.getMinutes(), 2) + ":" + w(d.getSeconds(), 2) + ":" + w(d.getMilliseconds(), 3);
}

/**
 * @method _tick
 * @private
 */
Subscription.prototype._tick = function () {

    const subscription = this;

    debugLog("Subscription#_tick  aborted=", subscription.aborted, "state=", subscription.state.toString());

    if (subscription.aborted) {
        //xx  console.log(" Log aborteds")
        //xx  // underlying channel has been aborted ...
        //xx self.publishEngine.cancelPendingPublishRequestBeforeChannelChange();
        //xx // let's still increase lifetime counter to detect timeout
    }


    if (subscription.state === SubscriptionState.CLOSED) {
        console.log("Warning: Subscription#_tick called while subscription is CLOSED");
        return;
    }

    subscription.discardOldSentNotifications();

    // istanbul ignore next
    if (doDebug) {
        debugLog((t(new Date()) + "  " + subscription._life_time_counter + "/" + subscription.lifeTimeCount + "   Subscription#_tick").cyan, "  processing subscriptionId=", subscription.id, "hasMonitoredItemNotifications = ", subscription.hasMonitoredItemNotifications, " publishingIntervalCount =", subscription.publishIntervalCount);
    }
    if (subscription.publishEngine._on_tick) {
        subscription.publishEngine._on_tick();
    }

    subscription.publishIntervalCount += 1;

    subscription.increaseLifeTimeCounter();

    if (subscription.lifeTimeHasExpired) {

        /* istanbul ignore next */
        if (doDebug) {
            debugLog("Subscription " + subscription.id + " has expired !!!!! => Terminating".red.bold);
        }
        /**
         * notify the subscription owner that the subscription has expired by exceeding its life time.
         * @event expired
         *
         */
        subscription.emit("expired");

        // notify new terminated status only when subscription has timeout.
        debugLog("adding StatusChangeNotification notification message for BadTimeout subscription = ", subscription.id);
        subscription._addNotificationMessage([new StatusChangeNotification({ status: StatusCodes.BadTimeout})]);

        // kill timer and delete monitored items and transfer pending notification messages
        subscription.terminate();

        return;

    }

    const publishEngine = subscription.publishEngine;

    // istanbul ignore next
    if (doDebug) {
        debugLog("Subscription#_tick  self._pending_notifications= ", subscription._pending_notifications.length);
    }

    if (publishEngine.pendingPublishRequestCount === 0 && (subscription.hasPendingNotifications || subscription.hasMonitoredItemNotifications)) {

        // istanbul ignore next
        if (doDebug) {
            debugLog("subscription set to LATE  hasPendingNotifications = ", subscription.hasPendingNotifications, " hasMonitoredItemNotifications =", subscription.hasMonitoredItemNotifications);
        }
        subscription.state = SubscriptionState.LATE;
        return;
    }

    if (publishEngine.pendingPublishRequestCount > 0) {

        if (subscription.hasPendingNotifications) {
            // simply pop pending notification and send it
            subscription.process_subscription();

        } else if (subscription.hasMonitoredItemNotifications) {
            subscription.process_subscription();

        } else {
            subscription._process_keepAlive();
        }
    } else {
        subscription._process_keepAlive();
    }
};


/**
 * @method _sendKeepAliveResponse
 * @private
 */
Subscription.prototype._sendKeepAliveResponse = function () {

    const subscription = this;
    const future_sequence_number = subscription._get_future_sequence_number();

    debugLog("     -> Subscription#_sendKeepAliveResponse subscriptionId", subscription.id);

    if (subscription.publishEngine.send_keep_alive_response(subscription.id, future_sequence_number)) {

        subscription.messageSent = true;

        /**
         * notify the subscription owner that a keepalive message has to be sent.
         * @event keepalive
         *
         */
        subscription.emit("keepalive", future_sequence_number);
        subscription.state = SubscriptionState.KEEPALIVE;

        return true;
    }
    return false;
};


/**
 * @method resetKeepAliveCounter
 * @private
 * Reset the Lifetime Counter Variable to the value specified for the lifetime of a Subscription in
 * the CreateSubscription Service( 5.13.2).
 */
Subscription.prototype.resetKeepAliveCounter = function () {
    const subscription = this;
    subscription._keep_alive_counter = 0;

    // istanbul ignore next
    if (doDebug) {
        debugLog("     -> subscriptionId", subscription.id, " Resetting keepAliveCounter = ", subscription._keep_alive_counter, subscription.maxKeepAliveCount);
    }
};

/**
 * @method increaseKeepAliveCounter
 * @private
 */
Subscription.prototype.increaseKeepAliveCounter = function () {
    const subscription = this;
    subscription._keep_alive_counter += 1;

    // istanbul ignore next
    if (doDebug) {
        debugLog("     -> subscriptionId", subscription.id, " Increasing keepAliveCounter = ", subscription._keep_alive_counter, subscription.maxKeepAliveCount);
    }
};

/**
 * @property keepAliveCounterHasExpired
 * @private
 * @type {Boolean} true if the keep alive counter has reach its limit.
 */
Subscription.prototype.__defineGetter__("keepAliveCounterHasExpired", function () {
    const subscription = this;
    return subscription._keep_alive_counter >= subscription.maxKeepAliveCount;
});


/**
 * Reset the Lifetime Counter Variable to the value specified for the lifetime of a Subscription in
 * the CreateSubscription Service( 5.13.2).
 * @method resetLifeTimeCounter
 * @private
 */
Subscription.prototype.resetLifeTimeCounter = function () {
    const subscription = this;
    subscription._life_time_counter = 0;
};
/**
 * @method increaseLifeTimeCounter
 * @private
 */
Subscription.prototype.increaseLifeTimeCounter = function () {
    const subscription = this;
    subscription._life_time_counter += 1;
};

/**
 *  True if the subscription life time has expired.
 *
 * @property lifeTimeHasExpired
 * @type {boolean} - true if the subscription life time has expired.
 */
Subscription.prototype.__defineGetter__("lifeTimeHasExpired", function () {
    const subscription = this;
    assert(subscription.lifeTimeCount > 0);
    return subscription._life_time_counter >= subscription.lifeTimeCount;
});

/**
 * number of milliseconds before this subscription times out (lifeTimeHasExpired === true);
 * @property timeToExpiration
 * @type {Number}
 */
Subscription.prototype.__defineGetter__("timeToExpiration", function () {
    const subscription = this;
    return (subscription.lifeTimeCount - subscription._life_time_counter) * subscription.publishingInterval;
});

Subscription.prototype.__defineGetter__("timeToKeepAlive", function () {
    const subscription = this;
    return (subscription.maxKeepAliveCount - subscription._keep_alive_counter) * subscription.publishingInterval;
});

/**
 *
 *  the server invokes the resetLifeTimeAndKeepAliveCounters method of the subscription
 *  when the server  has send a Publish Response, so that the subscription
 *  can reset its life time counter.
 *
 * @method resetLifeTimeAndKeepAliveCounters
 *
 */
Subscription.prototype.resetLifeTimeAndKeepAliveCounters = function () {
    const subscription = this;
    subscription.resetLifeTimeCounter();
    subscription.resetKeepAliveCounter();
};

/**
 * Terminates the subscription.
 * @method terminate
 *
 * Calling this method will also remove any monitored items.
 *
 */
Subscription.prototype.terminate = function () {

    assert(arguments.length === 0);

    const subscription = this;
    debugLog("Subscription#terminate status", subscription.state);

    if (subscription.state === SubscriptionState.CLOSED) {
        // todo verify if asserting is required here
        return;
    }
    assert(subscription.state !== SubscriptionState.CLOSED, "terminate already called ?");

    // stop timer
    subscription._stop_timer();

    debugLog("terminating Subscription  ", subscription.id, " with ", subscription.monitoredItemCount, " monitored items");

    // dispose all monitoredItem
    const keys = Object.keys(subscription.monitoredItems);

    for (const key of keys) {
        const status = subscription.removeMonitoredItem(key);
        assert(status === StatusCodes.Good);
    }
    assert(subscription.monitoredItemCount === 0);


    if (subscription.$session) {
        subscription.$session._unexposeSubscriptionDiagnostics(subscription);
    }
    subscription.state = SubscriptionState.CLOSED;

    /**
     * notify the subscription owner that the subscription has been terminated.
     * @event "terminated"
     */
    subscription.emit("terminated");

    subscription.publishEngine.on_close_subscription(subscription);

};

Subscription.prototype.dispose = function () {

    const subscription = this;


    if (doDebug) {
        debugLog("Subscription#dispose" ,subscription.id, subscription.monitoredItemCount);
    }

    assert(subscription.monitoredItemCount === 0, "MonitoredItems haven't been  deleted first !!!");
    assert(subscription.timerId === null, "Subscription timer haven't been terminated");

    if (subscription.subscriptionDiagnostics) {
        subscription.subscriptionDiagnostics.$subscription = null;
        subscription.subscriptionDiagnostics = null;
    }

    subscription.publishEngine = null;
    subscription._pending_notifications = null;
    subscription._sent_notifications = null;

    subscription.sessionId = null;
    subscription._sequence_number_generator = null;

    subscription.$session = null;
    subscription.removeAllListeners();

    Subscription.registry.unregister(subscription);

};

Subscription.prototype.__defineGetter__("aborted", function () {
    const subscription = this;
    const session = subscription.$session;
    if (!session) {
        return true;
    }
    return session.aborted;
});

function assert_validNotificationData(n) {
    assert(
        n instanceof DataChangeNotification ||
        n instanceof EventNotificationList ||
        n instanceof StatusChangeNotification
    );
}

/**
 * @method _addNotificationMessage
 * @param notificationData {Array<DataChangeNotification|EventNotificationList|StatusChangeNotification>}
 */
Subscription.prototype._addNotificationMessage = function (notificationData) {

    assert(_.isArray(notificationData));
    assert(notificationData.length === 1 || notificationData.length === 2); // as per spec part 3.

    // istanbul ignore next
    if (doDebug) {
        debugLog("Subscription#_addNotificationMessage".yellow, notificationData.toString());
    }
    const subscription = this;
    assert(_.isObject(notificationData[0]));

    assert_validNotificationData(notificationData[0]);
    if (notificationData.length === 2) {
        assert_validNotificationData(notificationData[1]);
    }

    const notification_message = new NotificationMessage({
        sequenceNumber: subscription._get_next_sequence_number(),
        publishTime: new Date(),
        notificationData: notificationData
    });

    subscription._pending_notifications.push({
        notification: notification_message,
        start_tick: subscription.publishIntervalCount,
        publishTime: new Date(),
        sequenceNumber: notification_message.sequenceNumber
    });
    debugLog("pending notification to send ", subscription._pending_notifications.length);

};


Subscription.prototype.getMessageForSequenceNumber = function (sequenceNumber) {

    const subscription = this;

    function filter_func(e) {
        return e.sequenceNumber === sequenceNumber;
    }

    const notification_message = _.find(subscription._sent_notifications, filter_func);

    if (!notification_message) {
        return null;
    }
    return notification_message;

};

/**
 * Extract the next Notification that is ready to be sent to the client.
 * @method _popNotificationToSend
 * @return {NotificationMessage}  the Notification to send._pending_notifications
 */
Subscription.prototype._popNotificationToSend = function () {
    const subscription = this;
    assert(subscription._pending_notifications.length > 0);
    const notification_message = subscription._pending_notifications.shift();
    subscription._sent_notifications.push(notification_message);
    return notification_message;
};

/**
 * returns true if the notification has expired
 * @method notificationHasExpired
 * @param notification
 * @return {boolean}
 */
Subscription.prototype.notificationHasExpired = function (notification) {
    const subscription = this;
    assert(notification.hasOwnProperty("start_tick"));
    assert(_.isFinite(notification.start_tick + subscription.maxKeepAliveCount));
    return (notification.start_tick + subscription.maxKeepAliveCount) < subscription.publishIntervalCount;
};

const maxNotificationMessagesInQueue = 100;
/**
 * discardOldSentNotification find all sent notification message that have expired keep-alive
 * and destroy them.
 * @method discardOldSentNotifications
 * @private
 *
 * Subscriptions maintain a retransmission queue of sent  NotificationMessages.
 * NotificationMessages are retained in this queue until they are acknowledged or until they have
 * been in the queue for a minimum of one keep-alive interval.
 *
 */
Subscription.prototype.discardOldSentNotifications = function () {

    const subscription = this;

    // Sessions maintain a retransmission queue of sent NotificationMessages. NotificationMessages
    // are retained in this queue until they are acknowledged. The Session shall maintain a
    // retransmission queue size of at least two times the number of Publish requests per Session the
    // Server supports.  Clients are required to acknowledge NotificationMessages as they are received. In the
    // case of a retransmission queue overflow, the oldest sent NotificationMessage gets deleted. If a
    // Subscription is transferred to another Session, the queued NotificationMessages for this
    // Subscription are moved from the old to the new Session.
    if (maxNotificationMessagesInQueue <= subscription._sent_notifications.length) {
        debugLog("discardOldSentNotifications = ", subscription._sent_notifications.length);
        subscription._sent_notifications.splice(subscription._sent_notifications.length - maxNotificationMessagesInQueue);
    }
    //
    //var arr = _.filter(self._sent_notifications,function(notification){
    //   return self.notificationHasExpired(notification);
    //});
    //var results = arr.map(function(notification){
    //    return self.acknowledgeNotification(notification.sequenceNumber);
    //});
    //xx return results;
};

function getSequenceNumbers(arr) {
    return arr.map(function (e) {
        return e.notification.sequenceNumber;
    });
}

/**
 *  returns in an array the sequence numbers of the notifications that haven't been
 *  acknowledged yet.
 *
 *  @method getAvailableSequenceNumbers
 *  @return {Integer[]}
 *
 */
Subscription.prototype.getAvailableSequenceNumbers = function () {
    const subscription = this;
    const availableSequenceNumbers = getSequenceNumbers(subscription._sent_notifications);
    return availableSequenceNumbers;
};


/**
 * @method acknowledgeNotification
 * @param sequenceNumber {Number}
 * @return {StatusCode}
 */
Subscription.prototype.acknowledgeNotification = function (sequenceNumber) {
    const subscription = this;

    let foundIndex = -1;
    _.find(subscription._sent_notifications, function (e, index) {
        if (e.sequenceNumber === sequenceNumber) {
            foundIndex = index;
        }
    });
    if (foundIndex === -1) {
        if (doDebug) {
            debugLog("acknowledging sequence FAILED !!! ".red, sequenceNumber.toString().cyan);
        }
        return StatusCodes.BadSequenceNumberUnknown;
    } else {
        if (doDebug) {
            debugLog("acknowledging sequence ".yellow, sequenceNumber.toString().cyan);
        }
        subscription._sent_notifications.splice(foundIndex, 1);
        subscription._unacknowledgedMessageCount--;
        return StatusCodes.Good;
    }
};


/**
 *
 * @property pendingNotificationsCount  - number of pending notifications
 * @type {Number}
 */
Subscription.prototype.__defineGetter__("pendingNotificationsCount", function () {
    return this._pending_notifications ? this._pending_notifications.length : 0;
});

/**
 * return True is there are pending notifications for this subscription. (i.e moreNotifications)
 *
 * @property hasPendingNotifications
 * @type {Boolean}
 */
Subscription.prototype.__defineGetter__("hasPendingNotifications", function () {
    const subscription = this;
    return subscription.pendingNotificationsCount > 0;
});

/**
 * number of sent notifications
 * @property sentNotificationsCount
 * @type {Number}
 */
Subscription.prototype.__defineGetter__("sentNotificationsCount", function () {
    return this._sent_notifications.length;
});

/**
 * number of monitored items.
 * @property monitoredItemCount
 * @type {Number}
 */
Subscription.prototype.__defineGetter__("monitoredItemCount", function () {
    return Object.keys(this.monitoredItems).length;
});

/**
 * number of disabled monitored items.
 * @property disabledMonitoredItemCount
 * @type {Number}
 */
Subscription.prototype.__defineGetter__("disabledMonitoredItemCount", function () {

    return _.reduce(_.values(this.monitoredItems), function (cumul, monitoredItem) {
        return cumul + ((monitoredItem.monitoringMode === MonitoringMode.Disabled) ? 1 : 0);
    }, 0);

});

/**
 * The number of unacknowledged messages saved in the republish queue.
 * @property unacknowledgedMessageCount
 * @type {Number}
 */
Subscription.prototype.__defineGetter__("unacknowledgedMessageCount", function () {
    const subscription = this;
    return subscription._unacknowledgedMessageCount;
});


const MonitoredItem = require("./monitored_item").MonitoredItem;


const MonitoredItemCreateRequest = require("node-opcua-service-subscription").MonitoredItemCreateRequest;


/**
 * adjust monitored item sampling interval
 *  - an samplingInterval ===0 means that we use a event-base model ( no sampling)
 *  - otherwise the sampling is adjusted
 *
 * @method adjustSamplingInterval
 * @param samplingInterval
 * @param node
 * @return {number|*}
 * @private
 */
Subscription.prototype.adjustSamplingInterval = function (samplingInterval, node) {

    const subscription = this;

    if (samplingInterval < 0) {
        // - The value -1 indicates that the default sampling interval defined by the publishing
        //   interval of the Subscription is requested.
        // - Any negative number is interpreted as -1.
        samplingInterval = subscription.publishingInterval;

    } else if (samplingInterval === 0) {

        // OPCUA 1.0.3 Part 4 - 5.12.1.2
        // The value 0 indicates that the Server should use the fastest practical rate.

        // The fastest supported sampling interval may be equal to 0, which indicates
        // that the data item is exception-based rather than being sampled at some period.
        // An exception-based model means that the underlying system does not require sampling and reports data changes.

        const dataValueSamplingInterval = node.readAttribute(SessionContext.defaultContext, AttributeIds.MinimumSamplingInterval);

        // TODO if attributeId === AttributeIds.Value : sampling interval required here
        if (dataValueSamplingInterval.statusCode === StatusCodes.Good) {
            // node provides a Minimum sampling interval ...
            samplingInterval = dataValueSamplingInterval.value.value;
            assert(samplingInterval >= 0 && samplingInterval <= MonitoredItem.maximumSamplingInterval);

            // note : at this stage, a samplingInterval===0 means that the data item is really exception-based

        }

    } else if (samplingInterval < MonitoredItem.minimumSamplingInterval) {

        samplingInterval = MonitoredItem.minimumSamplingInterval;

    } else if (samplingInterval > MonitoredItem.maximumSamplingInterval) {

        // If the requested samplingInterval is higher than the
        // maximum sampling interval supported by the Server, the maximum sampling
        // interval is returned.
        samplingInterval = MonitoredItem.maximumSamplingInterval;
    }

    const node_minimumSamplingInterval = (node && node.minimumSamplingInterval) ? node.minimumSamplingInterval : 0;
    samplingInterval = Math.max(samplingInterval, node_minimumSamplingInterval);

    return samplingInterval;
};


const checkSelectClauses = require("node-opcua-address-space").checkSelectClauses;

function analyseEventFilterResult(node, eventFilter) {
    assert(eventFilter instanceof EventFilter);
    const selectClauseResults = checkSelectClauses(node, eventFilter.selectClauses);

    const whereClauseResult = new subscription_service.ContentFilterResult();

    return new subscription_service.EventFilterResult({
        selectClauseResults: selectClauseResults,
        selectClauseDiagnosticInfos: [],
        whereClauseResult: whereClauseResult
    });
}

function analyseDataChangeFilterResult(node, dataChangeFilter) {
    assert(dataChangeFilter instanceof subscription_service.DataChangeFilter);
    // the opcua specification doesn't provide dataChangeFilterResult
    return null;
}

function analyseAggregateFilterResult(node, aggregateFilter) {
    assert(aggregateFilter instanceof subscription_service.AggregateFilter);
    return new subscription_service.AggregateFilterResult({});
}


function _process_filter(node, filter) {

    if (!filter) {
        return null;
    }

    if (filter instanceof EventFilter) {
        return analyseEventFilterResult(node, filter);
    } else if (filter instanceof DataChangeFilter) {
        return analyseDataChangeFilterResult(node, filter);
    } else if (filter instanceof AggregateFilter) {
        return analyseAggregateFilterResult(node, filter);
    }
    // istanbul ignore next
    throw new Error("invalid filter");
}


/**
 *
 * @param addressSpace
 * @param timestampsToReturn
 * @param monitoredItemCreateRequest
 * @returns {monitoredItemCreateResult}
 */
Subscription.prototype.createMonitoredItem = function (addressSpace, timestampsToReturn, monitoredItemCreateRequest) {

    const subscription = this;
    assert(addressSpace.constructor.name === "AddressSpace");
    assert(monitoredItemCreateRequest instanceof MonitoredItemCreateRequest);


    function handle_error(statusCode) {
        return new subscription_service.MonitoredItemCreateResult({statusCode: statusCode});
    }

    const itemToMonitor = monitoredItemCreateRequest.itemToMonitor;

    const node = addressSpace.findNode(itemToMonitor.nodeId);
    if (!node) {
        return handle_error(StatusCodes.BadNodeIdUnknown);
    }


    if (itemToMonitor.attributeId === AttributeIds.Value && !(node.nodeClass === NodeClass.Variable)) {
        // AttributeIds.Value is only valid for monitoring value of UAVariables.
        return handle_error(StatusCodes.BadAttributeIdInvalid);
    }


    if (itemToMonitor.attributeId === AttributeIds.INVALID) {
        return handle_error(StatusCodes.BadAttributeIdInvalid);
    }

    if (!itemToMonitor.indexRange.isValid()) {
        return handle_error(StatusCodes.BadIndexRangeInvalid);
    }

    // check dataEncoding applies only on Values
    if (itemToMonitor.dataEncoding.name && itemToMonitor.attributeId !== AttributeIds.Value) {
        return handle_error(StatusCodes.BadDataEncodingInvalid);
    }

    // check dataEncoding
    if (!isValidDataEncoding(itemToMonitor.dataEncoding)) {
        return handle_error(StatusCodes.BadDataEncodingUnsupported);
    }

    // check that item can be read by current user session

    // filter
    const requestedParameters = monitoredItemCreateRequest.requestedParameters;
    const filter = requestedParameters.filter;
    const statusCodeFilter = validateFilter(filter, itemToMonitor, node);
    if (statusCodeFilter !== StatusCodes.Good) {
        return handle_error(statusCodeFilter);
    }
    //xx var monitoringMode      = monitoredItemCreateRequest.monitoringMode; // Disabled, Sampling, Reporting
    //xx var requestedParameters = monitoredItemCreateRequest.requestedParameters;

    const monitoredItemCreateResult = subscription._createMonitoredItemStep2(timestampsToReturn, monitoredItemCreateRequest, node);

    assert(monitoredItemCreateResult.statusCode === StatusCodes.Good);

    const monitoredItem = subscription.getMonitoredItem(monitoredItemCreateResult.monitoredItemId);
    assert(monitoredItem);

    // TODO: fix old way to set node. !!!!
    monitoredItem.setNode(node);

    subscription.emit("monitoredItem", monitoredItem, itemToMonitor);

    subscription._createMonitoredItemStep3(monitoredItem, monitoredItemCreateRequest);

    return monitoredItemCreateResult;

};

let g_monitoredItemId = 1;

function getNextMonitoredItemId() {
    return g_monitoredItemId++;
}

/**
 *
 * @method _createMonitoredItemStep2
 * @param timestampsToReturn
 * @param {MonitoredItemCreateRequest} monitoredItemCreateRequest - the parameters describing the monitored Item to create
 * @param node {BaseNode}
 * @return {subscription_service.MonitoredItemCreateResult}
 * @private
 */
Subscription.prototype._createMonitoredItemStep2 = function (timestampsToReturn, monitoredItemCreateRequest, node) {

    const subscription = this;

    // note : most of the parameter inconsistencies shall have been handled by the caller
    // any error here will raise an assert here

    assert(monitoredItemCreateRequest instanceof MonitoredItemCreateRequest);
    const itemToMonitor = monitoredItemCreateRequest.itemToMonitor;

    //xx check if attribute Id invalid (we only support Value or EventNotifier )
    //xx assert(itemToMonitor.attributeId !== AttributeIds.INVALID);

    subscription.monitoredItemIdCounter += 1;


    const monitoredItemId = getNextMonitoredItemId();

    const requestedParameters = monitoredItemCreateRequest.requestedParameters;

    // adjust requestedParameters.samplingInterval
    requestedParameters.samplingInterval = subscription.adjustSamplingInterval(requestedParameters.samplingInterval, node);

    // reincorporate monitoredItemId and itemToMonitor into the requestedParameters
    requestedParameters.monitoredItemId = monitoredItemId;
    requestedParameters.itemToMonitor = itemToMonitor;


    const monitoredItem = new MonitoredItem(requestedParameters);
    monitoredItem.timestampsToReturn = timestampsToReturn;
    monitoredItem.$subscription = subscription;

    assert(monitoredItem.monitoredItemId === monitoredItemId);
    subscription.monitoredItems[monitoredItemId] = monitoredItem;

    const filterResult = _process_filter(node, requestedParameters.filter);


    const monitoredItemCreateResult = new subscription_service.MonitoredItemCreateResult({
        statusCode: StatusCodes.Good,
        monitoredItemId: monitoredItemId,
        revisedSamplingInterval: monitoredItem.samplingInterval,
        revisedQueueSize: monitoredItem.queueSize,
        filterResult: filterResult
    });
    return monitoredItemCreateResult;
};


Subscription.prototype._createMonitoredItemStep3 = function (monitoredItem, monitoredItemCreateRequest) {

    assert(monitoredItem.monitoringMode === MonitoringMode.Invalid);
    assert(_.isFunction(monitoredItem.samplingFunc));
    const monitoringMode = monitoredItemCreateRequest.monitoringMode; // Disabled, Sampling, Reporting
    monitoredItem.setMonitoringMode(monitoringMode);

};

/**
 * get a monitoredItem by Id.
 * @method getMonitoredItem
 * @param monitoredItemId  {Number} the id of the monitored item to get.
 * @return {MonitoredItem}
 */
Subscription.prototype.getMonitoredItem = function (monitoredItemId) {
    assert(_.isFinite(monitoredItemId));
    const subscription = this;
    return subscription.monitoredItems[monitoredItemId];
};

/**
 * getMonitoredItems is used to get information about monitored items of a subscription.Its intended
 * use is defined in Part 4. This method is the implementation of the Standard OPCUA GetMonitoredItems Method.
 * @method getMonitoredItems
 * @param  result.serverHandles {Int32[]} Array of serverHandles for all MonitoredItems of the subscription identified by subscriptionId.
 *         result.clientHandles {Int32[]} Array of clientHandles for all MonitoredItems of the subscription identified by subscriptionId.
 *         result.statusCode    {StatusCode}
 * from spec:
 * This method can be used to get the  list of monitored items in a subscription if CreateMonitoredItems failed due to
 * a network interruption and the client does not know if the creation succeeded in the server.
 *
 */
Subscription.prototype.getMonitoredItems = function (/*out*/ result) {

    result = result || {};
    const subscription = this;
    result.serverHandles = [];
    result.clientHandles = [];
    result.statusCode = StatusCodes.Good;

    Object.keys(subscription.monitoredItems).forEach(function (monitoredItemId) {

        const monitoredItem = subscription.getMonitoredItem(monitoredItemId);

        result.clientHandles.push(monitoredItem.clientHandle);
        // TODO:  serverHandle is defined anywhere in the OPCUA Specification 1.02
        //        I am not sure what shall be reported for serverHandle...
        //        using monitoredItem.monitoredItemId instead...
        //        May be a clarification in the OPCUA Spec is required.
        result.serverHandles.push(monitoredItemId);

    });
    return result;
};

MonitoredItem.prototype.resendInitialValues = function () {
    // tte first Publish response(s) after the TransferSubscriptions call shall contain the current values of all
    // Monitored Items in the Subscription where the Monitoring Mode is set to Reporting.
    // the first Publish response after the TransferSubscriptions call shall contain only the value changes since
    // the last Publish response was sent.
    // This parameter only applies to MonitoredItems used for monitoring Attribute changes.
    const subscription = this;
    subscription._stop_sampling();
    subscription._start_sampling(true);
};

Subscription.prototype.resendInitialValues = function () {
    const subscription = this;
    _.forEach(subscription.monitoredItems, function (monitoredItem/*,monitoredItemId*/) {
        monitoredItem.resendInitialValues();
    });
};


/**
 * remove a monitored Item from the subscription.
 * @method removeMonitoredItem
 * @param monitoredItemId  {Number} the id of the monitored item to get.
 */
Subscription.prototype.removeMonitoredItem = function (monitoredItemId) {

    debugLog("Removing monitoredIem ", monitoredItemId);

    assert(_.isFinite(monitoredItemId));
    const subscription = this;
    if (!subscription.monitoredItems.hasOwnProperty(monitoredItemId)) {
        return StatusCodes.BadMonitoredItemIdInvalid;
    }

    const monitoredItem = subscription.monitoredItems[monitoredItemId];

    monitoredItem.terminate();

    monitoredItem.dispose();

    /**
     *
     * notify that a monitored item has been removed from the subscription
     * @event removeMonitoredItem
     * @param monitoredItem {MonitoredItem}
     */
    subscription.emit("removeMonitoredItem", monitoredItem);

    delete subscription.monitoredItems[monitoredItemId];

    return StatusCodes.Good;

};


/**
 * @property hasMonitoredItemNotifications true if monitored Item have uncollected Notifications
 * @type {Boolean}
 */
Subscription.prototype.__defineGetter__("hasMonitoredItemNotifications", function () {
    const subscription = this;
    if (subscription._hasMonitoredItemNotifications) {
        return true;
    }
    const keys = Object.keys(subscription.monitoredItems);
    let i, key;
    const n = keys.length;
    for (i = 0; i < n; i++) {
        key = keys[i];
        const monitoredItem = subscription.monitoredItems[key];
        if (monitoredItem.hasMonitoredItemNotifications) {
            subscription._hasMonitoredItemNotifications = true;
            return true;
        }
    }
    return false;
});

/**
 * @method extract_notifications_chunk
 * extract up to maxNotificationsPerPublish notifications
 * @param monitoredItems {Array<MonitoredItem>}
 * @param maxNotificationsPerPublish {Number} the maximum number of notification to extract
 * @return {Array<MonitoredItem>}
 * @static
 * @private
 */
function extract_notifications_chunk(monitoredItems, maxNotificationsPerPublish) {

    let n = maxNotificationsPerPublish === 0 ?
        monitoredItems.length :
        Math.min(monitoredItems.length, maxNotificationsPerPublish);

    const chunk_monitoredItems = [];
    while (n) {
        chunk_monitoredItems.push(monitoredItems.shift());
        n--;
    }
    return chunk_monitoredItems;
}

function add_all_in(notifications, all_notifications) {
    for (let i = 0; i < notifications.length; i++) {
        const n = notifications[i];
        all_notifications.push(n);
    }
}

function filter_instanceof(Class, e) {
    return (e instanceof Class);
}

// collect DataChangeNotification
Subscription.prototype._collectNotificationData = function () {

    const subscription = this;
    let notifications = [];

    // reset cache ...
    subscription._hasMonitoredItemNotifications = false;

    const all_notifications = new Dequeue();

    // visit all monitored items
    const keys = Object.keys(subscription.monitoredItems);
    let i, key;
    const n = keys.length;
    for (i = 0; i < n; i++) {
        key = keys[i];
        const monitoredItem = subscription.monitoredItems[key];
        notifications = monitoredItem.extractMonitoredItemNotifications();
        add_all_in(notifications, all_notifications);
    }

    const notificationsMessage = [];

    while (all_notifications.length > 0) {

        // split into one or multiple dataChangeNotification with no more than
        //  self.maxNotificationsPerPublish monitoredItems
        const notifications_chunk = extract_notifications_chunk(all_notifications, subscription.maxNotificationsPerPublish);

        // separate data for DataChangeNotification (MonitoredItemNotification) from data for EventNotificationList(EventFieldList)
        const dataChangedNotificationData = notifications_chunk.filter(filter_instanceof.bind(null, subscription_service.MonitoredItemNotification));
        const eventNotificationListData = notifications_chunk.filter(filter_instanceof.bind(null, subscription_service.EventFieldList));

        assert(notifications_chunk.length === dataChangedNotificationData.length + eventNotificationListData.length);

        notifications = [];

        // add dataChangeNotification
        if (dataChangedNotificationData.length) {
            const dataChangeNotification = new DataChangeNotification({
                monitoredItems: dataChangedNotificationData,
                diagnosticInfos: []
            });
            notifications.push(dataChangeNotification);
        }

        // add dataChangeNotification
        if (eventNotificationListData.length) {
            const eventNotificationList = new EventNotificationList({
                events: eventNotificationListData
            });

            notifications.push(eventNotificationList);
        }

        assert(notifications.length === 1 || notifications.length === 2);
        notificationsMessage.push(notifications);
    }

    assert(notificationsMessage instanceof Array);
    return notificationsMessage;
};

Subscription.prototype._harvestMonitoredItems = function () {

    const subscription = this;

    // Only collect data change notification for the time being
    const notificationData = subscription._collectNotificationData();
    assert(notificationData instanceof Array);

    // istanbul ignore next
    if (doDebug) {
        debugLog("Subscription#_harvestMonitoredItems =>", notificationData.length);
    }
    notificationData.forEach(function (notificationMessage) {
        subscription._addNotificationMessage(notificationMessage);
    });
    subscription._hasMonitoredItemNotifications = false;

};
Subscription.prototype.__defineGetter__("subscriptionId", function () {
    return this.id;
});


Subscription.prototype.notifyTransfer = function () {

    // OPCUA UA Spec 1.0.3 : part 3 - page 82 - 5.13.7 TransferSubscriptions:
    // If the Server transfers the Subscription to the new Session, the Server shall issue a StatusChangeNotification
    // notificationMessage with the status code Good_SubscriptionTransferred to the old Session.
    const subscription = this;

    debugLog(" Subscription => Notifying Transfer                                  ".red);

    const notificationData = [new StatusChangeNotification({status: StatusCodes.GoodSubscriptionTransferred})];

    subscription.publishEngine.send_notification_message({
        subscriptionId: subscription.id,
        sequenceNumber: subscription._get_next_sequence_number(),
        notificationData: notificationData,
        moreNotifications: false
    }, true);

};


exports.Subscription = Subscription;