"use strict";
/**
* @module opcua.client
*/
const util = require("util");
const assert = require("node-opcua-assert").assert;
const _ = require("underscore");
const EventEmitter = require("events").EventEmitter;
const utils = require("node-opcua-utils");
const subscription_service = require("node-opcua-service-subscription");
const ClientSession = require("../src/client_session").ClientSession;
const ClientMonitoredItem = require("../src/client_monitored_item").ClientMonitoredItem;
const StatusCodes = require("node-opcua-status-code").StatusCodes;
const resolveNodeId = require("node-opcua-nodeid").resolveNodeId;
const AttributeIds = require("node-opcua-data-model").AttributeIds;
const TimestampsToReturn = require("node-opcua-service-read").TimestampsToReturn;
const debugLog = require("node-opcua-debug").make_debugLog(__filename);
const doDebug = require("node-opcua-debug").checkDebugFlag(__filename);
/**
* a object to manage a subscription on the client side.
*
* @class ClientSubscription
* @extends EventEmitter
* events:
* "started", callback(subscriptionId) : the subscription has been initiated
* "terminated" : the subscription has been deleted
* "error", : the subscription has received an error
* "keepalive", : the subscription has received a keep alive message from the server
* "received_notifications", : the subscription has received one or more notification
*
* @param session
* @param options {Object}
* @param options.requestedPublishingInterval {Number}
* @param options.requestedLifetimeCount {Number}
* @param options.requestedMaxKeepAliveCount {Number}
* @param options.maxNotificationsPerPublish {Number}
* @param options.publishingEnabled {Boolean}
* @param options.priority {Number}
*
*
* @constructor
*/
function ClientSubscription(session, options) {
assert(session instanceof ClientSession);
const self = this;
self.publish_engine = session.getPublishEngine();
self.lastSequenceNumber = -1;
//// options should have
//var allowedProperties = [
// 'requestedPublishingInterval',
// 'requestedLifetimeCount',
// 'requestedMaxKeepAliveCount',
// 'maxNotificationsPerPublish',
// 'publishingEnabled',
// 'priority'
//];
options = options || {};
options.requestedPublishingInterval = options.requestedPublishingInterval || 100;
options.requestedLifetimeCount = options.requestedLifetimeCount || 60;
options.requestedMaxKeepAliveCount = options.requestedMaxKeepAliveCount || 10;
options.maxNotificationsPerPublish = utils.isNullOrUndefined(options.maxNotificationsPerPublish) ? 0 : options.maxNotificationsPerPublish;
options.publishingEnabled = options.publishingEnabled ? true : false;
options.priority = options.priority || 1;
self.publishingInterval = options.requestedPublishingInterval;
self.lifetimeCount = options.requestedLifetimeCount;
self.maxKeepAliveCount = options.requestedMaxKeepAliveCount;
self.maxNotificationsPerPublish = options.maxNotificationsPerPublish;
self.publishingEnabled = options.publishingEnabled;
self.priority = options.priority;
self.subscriptionId = "pending";
self._next_client_handle = 0;
self.monitoredItems = {};
/**
* set to True when the server has notified us that this sbuscription has timed out
* ( maxLifeCounter x published interval without being able to process a PublishRequest
* @property hasTimedOut
* @type {boolean}
*/
self.hasTimedOut = false;
setImmediate(function () {
self.__create_subscription(function (err) {
if (!err) {
setImmediate(function () {
/**
* notify the observers that the subscription has now started
* @event started
*/
self.emit("started", self.subscriptionId);
});
}
});
});
}
util.inherits(ClientSubscription, EventEmitter);
ClientSubscription.prototype.__create_subscription = function (callback) {
assert(_.isFunction(callback));
const self = this;
const session = self.publish_engine.session;
debugLog("ClientSubscription created ".yellow.bold);
const request = new subscription_service.CreateSubscriptionRequest({
requestedPublishingInterval: self.publishingInterval,
requestedLifetimeCount: self.lifetimeCount,
requestedMaxKeepAliveCount: self.maxKeepAliveCount,
maxNotificationsPerPublish: self.maxNotificationsPerPublish,
publishingEnabled: self.publishingEnabled,
priority: self.priority
});
session.createSubscription(request, function (err, response) {
if (err) {
/* istanbul ignore next */
self.emit("internal_error", err);
if (callback) {
return callback(err);
}
} else {
self.subscriptionId = response.subscriptionId;
self.publishingInterval = response.revisedPublishingInterval;
self.lifetimeCount = response.revisedLifetimeCount;
self.maxKeepAliveCount = response.revisedMaxKeepAliveCount;
self.timeoutHint = (self.maxKeepAliveCount + 10) * self.publishingInterval;
if (doDebug) {
debugLog("registering callback".yellow.bold);
debugLog("publishingInterval ".yellow.bold, self.publishingInterval);
debugLog("lifetimeCount ".yellow.bold, self.lifetimeCount);
debugLog("maxKeepAliveCount ".yellow.bold, self.maxKeepAliveCount);
debugLog("publish request timeout hint = ".yellow.bold, self.timeoutHint);
debugLog("hasTimedOut ".yellow.bold, self.hasTimedOut);
}
self.publish_engine.registerSubscription(self);
if (callback) {
callback(err);
}
}
});
};
ClientSubscription.prototype.__on_publish_response_DataChangeNotification = function (notification) {
assert(notification._schema.name === "DataChangeNotification");
const self = this;
const monitoredItems = notification.monitoredItems;
monitoredItems.forEach(function (monitoredItem) {
const monitorItemObj = self.monitoredItems[monitoredItem.clientHandle];
if (monitorItemObj) {
if (monitorItemObj.itemToMonitor.attributeId === AttributeIds.EventNotifier) {
console.log("Warning".yellow, " Server send a DataChangeNotification for an EventNotifier. EventNotificationList was expected".cyan);
console.log(" the Server may not be fully OPCUA compliant".cyan, ". This notification will be ignored.".yellow);
} else {
monitorItemObj._notify_value_change(monitoredItem.value);
}
}
});
};
ClientSubscription.prototype.__on_publish_response_StatusChangeNotification = function (notification) {
const self = this;
assert(notification._schema.name === "StatusChangeNotification");
debugLog("Client has received a Status Change Notification ", notification.statusCode.toString());
self.publish_engine.cleanup_acknowledgment_for_subscription(notification.subscriptionId);
if (notification.statusCode === StatusCodes.GoodSubscriptionTransferred) {
// 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.
debugLog("ClientSubscription#__on_publish_response_StatusChangeNotification : GoodSubscriptionTransferred");
self.hasTimedOut = true;
self.terminate(function () {
});
}
if (notification.statusCode === StatusCodes.BadTimeout) {
// the server tells use that the subscription has timed out ..
// this mean that this subscription has been closed on the server side and cannot process any
// new PublishRequest.
//
// from Spec OPCUA Version 1.03 Part 4 - 5.13.1.1 Description : Page 69:
//
// h. Subscriptions have a lifetime counter that counts the number of consecutive publishing cycles in
// which there have been no Publish requests available to send a Publish response for the
// Subscription. Any Service call that uses the SubscriptionId or the processing of a Publish
// response resets the lifetime counter of this Subscription. When this counter reaches the value
// calculated for the lifetime of a Subscription based on the MaxKeepAliveCount parameter in the
// CreateSubscription Service (5.13.2), the Subscription is closed. Closing the Subscription causes
// its MonitoredItems to be deleted. In addition the Server shall issue a StatusChangeNotification
// notificationMessage with the status code Bad_Timeout.
//
self.hasTimedOut = true;
self.terminate(function () {
});
}
/**
* notify the observers that the server has send a status changed notification (such as BadTimeout )
* @event status_changed
*/
self.emit("status_changed", notification.statusCode, notification.diagnosticInfo);
};
ClientSubscription.prototype.__on_publish_response_EventNotificationList = function (notification) {
assert(notification._schema.name === "EventNotificationList");
const self = this;
notification.events.forEach(function (event) {
const monitorItemObj = self.monitoredItems[event.clientHandle];
assert(monitorItemObj, "Expecting a monitored item");
monitorItemObj._notify_value_change(event.eventFields);
});
};
ClientSubscription.prototype.onNotificationMessage = function (notificationMessage) {
const self = this;
self.lastRequestSentTime = Date.now();
assert(notificationMessage.hasOwnProperty("sequenceNumber"));
self.lastSequenceNumber = notificationMessage.sequenceNumber;
self.emit("raw_notification", notificationMessage);
const notificationData = notificationMessage.notificationData;
if (notificationData.length === 0) {
// this is a keep alive message
debugLog("Client : received a keep alive notification from client".yellow);
/**
* notify the observers that a keep alive Publish Response has been received from the server.
* @event keepalive
*/
self.emit("keepalive");
} else {
/**
* notify the observers that some notifications has been received from the server in a PublishResponse
* each modified monitored Item
* @event received_notifications
*/
self.emit("received_notifications", notificationMessage);
// let publish a global event
// now process all notifications
notificationData.forEach(function (notification) {
// DataChangeNotification / StatusChangeNotification / EventNotification
switch (notification._schema.name) {
case "DataChangeNotification":
// now inform each individual monitored item
self.__on_publish_response_DataChangeNotification(notification);
break;
case "StatusChangeNotification":
self.__on_publish_response_StatusChangeNotification(notification);
break;
case "EventNotificationList":
self.__on_publish_response_EventNotificationList(notification);
break;
default:
console.log(" Invalid notification :", notification.toString());
}
});
}
};
/**
* the associated session
* @property session
* @type {ClientSession}
*/
ClientSubscription.prototype.__defineGetter__("session", function () {
return this.publish_engine.session;
});
ClientSubscription.prototype._terminate_step2 = function (callback) {
const self = this;
setImmediate(function () {
/**
* notify the observers tha the client subscription has terminated
* @event terminated
*/
self.subscriptionId = "terminated";
self.emit("terminated");
callback();
});
};
/**
* @method terminate
* @param callback
*
*/
ClientSubscription.prototype.terminate = function (callback) {
const self = this;
assert(_.isFunction(callback), "expecting a callback function");
if (self.subscriptionId === "terminated" || self.subscriptionId == "terminating") {
// already terminated... just ignore
return callback(new Error("Already Terminated"));
}
if (_.isFinite(self.subscriptionId)) {
const subscriptionId = self.subscriptionId;
self.subscriptionId = "terminating";
self.publish_engine.unregisterSubscription(subscriptionId);
if (!self.session) {
return self._terminate_step2(callback);
}
self.session.deleteSubscriptions({
subscriptionIds: [subscriptionId]
}, function (err) {
if (err) {
/**
* notify the observers that an error has occurred
* @event internal_error
* @param {Error} err the error
*/
self.emit("internal_error", err);
}
self._terminate_step2(callback);
});
} else {
assert(self.subscriptionId === "pending");
self._terminate_step2(callback);
}
};
/**
* @method nextClientHandle
*/
ClientSubscription.prototype.nextClientHandle = function () {
this._next_client_handle += 1;
return this._next_client_handle;
};
ClientSubscription.prototype._add_monitored_item = function (clientHandle, monitoredItem) {
const self = this;
assert(self.isActive(), "subscription must be active and not terminated");
assert(monitoredItem.monitoringParameters.clientHandle === clientHandle);
self.monitoredItems[clientHandle] = monitoredItem;
/**
* notify the observers that a new monitored item has been added to the subscription.
* @event item_added
* @param {MonitoredItem} the monitored item.
*/
self.emit("item_added", monitoredItem);
};
ClientSubscription.prototype._wait_for_subscription_to_be_ready = function (done) {
const self = this;
let _watch_dog = 0;
function wait_for_subscription_and_monitor() {
_watch_dog++;
if (self.subscriptionId === "pending") {
// the subscriptionID is not yet known because the server hasn't replied yet
// let postpone this call, a little bit, to let things happen
setImmediate(wait_for_subscription_and_monitor);
} else if (self.subscriptionId === "terminated") {
// the subscription has been terminated in the meantime
// this indicates a potential issue in the code using this api.
if (_.isFunction(done)) {
done(new Error("subscription has been deleted"));
}
} else {
done();
}
}
setImmediate(wait_for_subscription_and_monitor);
};
/**
* add a monitor item to the subscription
*
* @method monitor
* @async
* @param itemToMonitor {ReadValueId}
* @param itemToMonitor.nodeId {NodeId}
* @param itemToMonitor.attributeId {AttributeId}
* @param itemToMonitor.indexRange {null|NumericRange}
* @param itemToMonitor.dataEncoding
* @param requestedParameters {MonitoringParameters}
* @param requestedParameters.clientHandle {IntegerId}
* @param requestedParameters.samplingInterval {Duration}
* @param requestedParameters.filter {ExtensionObject|null} EventFilter/DataChangeFilter
* @param requestedParameters.queueSize {Counter}
* @param requestedParameters.discardOldest {Boolean}
* @param timestampsToReturn {TimestampsToReturn} //{TimestampsToReturnId}
* @param [done] {Function} optional done callback
* @return {ClientMonitoredItem}
*
*
* Monitoring a simple Value Change
* ---------------------------------
*
* @example:
*
* clientSubscription.monitor(
* // itemToMonitor:
* {
* nodeId: "ns=0;i=2258",
* attributeId: AttributeIds.Value,
* indexRange: null,
* dataEncoding: { namespaceIndex: 0, name: null }
* },
* // requestedParameters:
* {
* samplingInterval: 3000,
* filter: null,
* queueSize: 1,
* discardOldest: true
* },
* TimestampsToReturn.Neither
* );
*
* Monitoring a Value Change With a DataChange Filter
* ---------------------------------------------------
*
* options.trigger {DataChangeTrigger} {Status|StatusValue|StatusValueTimestamp}
* options.deadbandType {DeadbandType} {None|Absolute|Percent}
* options.deadbandValue {Double}
* @example:
*
* clientSubscription.monitor(
* // itemToMonitor:
* {
* nodeId: "ns=0;i=2258",
* attributeId: AttributeIds.Value,
* },
* // requestedParameters:
* {
* samplingInterval: 3000,
* filter: new DataChangeFilter({
* trigger: DataChangeTrigger.StatusValue,
* deadbandType: DeadBandType.Absolute,
* deadbandValue: 0.1
* }),
* queueSize: 1,
* discardOldest: true
* },
* TimestampsToReturn.Neither
* );
*
*
* Monitoring an Event
* -------------------
*
* If the monitor attributeId is EventNotifier then the filter must be specified
*
* @example:
*
* var filter = new subscription_service.EventFilter({
* selectClauses: [
* { browsePath: [ {name: 'ActiveState' }, {name: 'id'} ]},
* { browsePath: [ {name: 'ConditionName'} ]}
* ],
* whereClause: []
* });
*
* clientSubscription.monitor(
* // itemToMonitor:
* {
* nodeId: "ns=0;i=2258",
* attributeId: AttributeIds.EventNotifier,
* indexRange: null,
* dataEncoding: { namespaceIndex: 0, name: null }
* },
* // requestedParameters:
* {
* samplingInterval: 3000,
*
* filter: filter,
*
* queueSize: 1,
* discardOldest: true
* },
* TimestampsToReturn.Neither
* );
*
*
*
*
*
*
*/
ClientSubscription.prototype.monitor = function (itemToMonitor, requestedParameters, timestampsToReturn, done) {
const self = this;
assert(done === undefined || _.isFunction(done));
itemToMonitor.nodeId = resolveNodeId(itemToMonitor.nodeId);
const monitoredItem = new ClientMonitoredItem(this, itemToMonitor, requestedParameters, timestampsToReturn);
self._wait_for_subscription_to_be_ready(function (err) {
if (err) {
return done && done(err);
}
monitoredItem._monitor(function (err) {
if (err) {
return done && done(err);
}
done && done(err, monitoredItem);
});
});
return monitoredItem;
};
const ClientMonitoredItemGroup = require("./client_monitored_item_group").ClientMonitoredItemGroup;
/**
* @method monitorItems
* @param itemsToMonitor
* @param requestedParameters
* @param timestampsToReturn
* @param done
*/
ClientSubscription.prototype.monitorItems = function (itemsToMonitor, requestedParameters, timestampsToReturn, done) {
const self = this;
// Try to resolve the nodeId and fail fast if we can't.
itemsToMonitor.forEach(function (itemToMonitor) {
itemToMonitor.nodeId = resolveNodeId(itemToMonitor.nodeId);
});
const monitoredItemGroup = new ClientMonitoredItemGroup(this, itemsToMonitor, requestedParameters, timestampsToReturn);
self._wait_for_subscription_to_be_ready(function (err) {
if (err) {
return done && done(err);
}
monitoredItemGroup._monitor(function (err) {
if (err) {
return done && done(err);
}
done && done(err, monitoredItemGroup);
});
});
return monitoredItemGroup;
};
ClientSubscription.prototype.isActive = function () {
const self = this;
return typeof self.subscriptionId !== "string";
};
ClientSubscription.prototype._remove = function (monitoredItem) {
const self = this;
const clientHandle = monitoredItem.monitoringParameters.clientHandle;
assert(clientHandle > 0);
if (!self.monitoredItems.hasOwnProperty(clientHandle)) {
return; // may be monitoredItem failed to be created ....
}
assert(self.monitoredItems.hasOwnProperty(clientHandle));
monitoredItem.removeAllListeners();
delete self.monitoredItems[clientHandle];
};
ClientSubscription.prototype._delete_monitored_items = function (monitoredItems, callback) {
assert(_.isFunction(callback));
assert(_.isArray(monitoredItems));
const self = this;
assert(self.isActive());
monitoredItems.forEach(function (monitoredItem) {
self._remove(monitoredItem);
});
self.session.deleteMonitoredItems({
subscriptionId: self.subscriptionId,
monitoredItemIds: monitoredItems.map(function (monitoredItem) {
return monitoredItem.monitoredItemId;
})
}, function (err) {
callback(err);
});
};
ClientSubscription.prototype._delete_monitored_item = function (monitoredItem, callback) {
const self = this;
self._delete_monitored_items([monitoredItem], callback);
};
ClientSubscription.prototype.setPublishingMode = function (publishingEnabled, callback) {
assert(_.isFunction(callback));
const self = this;
self.session.setPublishingMode(publishingEnabled, self.subscriptionId, function (err, results) {
if (err) {
return callback(err);
}
if (results[0] !== StatusCodes.Good) {
return callback(new Error("Cannot setPublishingMode " + results[0].toString()));
}
callback();
});
};
const async = require("async");
/**
* utility function to recreate new subscription
* @method recreateSubscriptionAndMonitoredItem
*/
ClientSubscription.prototype.recreateSubscriptionAndMonitoredItem = function (callback) {
debugLog("ClientSubscription#recreateSubscriptionAndMonitoredItem");
const subscription = this;
const monitoredItems_old = subscription.monitoredItems;
subscription.publish_engine.unregisterSubscription(subscription.subscriptionId);
async.series([
subscription.__create_subscription.bind(subscription),
function (callback) {
const test = subscription.publish_engine.getSubscription(subscription.subscriptionId);
assert(test === subscription);
// re-create monitored items
const itemsToCreate = [];
_.forEach(monitoredItems_old, function (monitoredItem /*, clientHandle*/) {
assert(monitoredItem.monitoringParameters.clientHandle > 0);
itemsToCreate.push({
itemToMonitor: monitoredItem.itemToMonitor,
monitoringMode: monitoredItem.monitoringMode,
requestedParameters: monitoredItem.monitoringParameters
});
});
const createMonitorItemsRequest = new subscription_service.CreateMonitoredItemsRequest({
subscriptionId: subscription.subscriptionId,
timestampsToReturn: TimestampsToReturn.Both, // self.timestampsToReturn,
itemsToCreate: itemsToCreate
});
subscription.session.createMonitoredItems(createMonitorItemsRequest, function (err, response) {
if (!err) {
assert(response instanceof subscription_service.CreateMonitoredItemsResponse);
const monitoredItemResults = response.results;
monitoredItemResults.forEach(function (monitoredItemResult, index) {
const clientHandle = itemsToCreate[index].requestedParameters.clientHandle;
const monitoredItem = subscription.monitoredItems[clientHandle];
if (monitoredItemResult.statusCode === StatusCodes.Good) {
monitoredItem.result = monitoredItemResult;
monitoredItem.monitoredItemId = monitoredItemResult.monitoredItemId;
monitoredItem.monitoringParameters.samplingInterval = monitoredItemResult.revisedSamplingInterval;
monitoredItem.monitoringParameters.queueSize = monitoredItemResult.revisedQueueSize;
monitoredItem.filterResult = monitoredItemResult.filterResult;
// istanbul ignore next
if (doDebug) {
debugLog("monitoredItemResult.statusCode = ", monitoredItemResult.toString());
}
} else {
// TODO: what should we do ?
debugLog("monitoredItemResult.statusCode = ", monitoredItemResult.statusCode.toString());
}
});
}
callback(err);
});
}
], callback);
};
ClientSubscription.prototype.toString = function () {
const subscription = this;
let str = "";
str += "subscriptionId :" + subscription.subscriptionId + "\n";
str += "publishingInterval :" + subscription.publishingInterval + "\n";
str += "lifetimeCsount :" + subscription.lifetimeCount + "\n";
str += "maxKeepAliveCount :" + subscription.maxKeepAliveCount + "\n";
return str;
};
ClientSubscription.prototype.evaluateRemainingLifetime = function() {
const subscription = this;
const now = Date.now();
const timeout = subscription.publishingInterval * subscription.lifetimeCount;
const expiryTime = subscription.lastRequestSentTime + timeout;
return Math.max(0,(expiryTime - now));
};
exports.ClientSubscription = ClientSubscription;
const thenify = require("thenify");
ClientSubscription.prototype.setPublishingMode = thenify.withCallback(ClientSubscription.prototype.setPublishingMode);
//xx ClientSubscription.prototype.monitor = thenify.withCallback(ClientSubscription.prototype.monitor);
//xx ClientSubscription.prototype.monitorItems = thenify.withCallback(ClientSubscription.prototype.monitorItems);
ClientSubscription.prototype.recreateSubscriptionAndMonitoredItem = thenify.withCallback(ClientSubscription.prototype.recreateSubscriptionAndMonitoredItem);
ClientSubscription.prototype.terminate = thenify.withCallback(ClientSubscription.prototype.terminate);
The NodeOPCUA API