Commit f647068 for jssip.net
commit f6470688954c248dc6e2378aa250c4ff034d1600
Author: Igor Kolosov <10967586+ikq@users.noreply.github.com>
Date: Thu Jan 22 12:23:34 2026 +0200
Subscribe support (#711)
SIP-Specific Event Notification (RFC 6665)
--------
Co-authored-by: Orgad Shaneh <orgad.shaneh@audiocodes.com>
diff --git a/src/Constants.d.ts b/src/Constants.d.ts
index 2d75278..31297cc 100644
--- a/src/Constants.d.ts
+++ b/src/Constants.d.ts
@@ -57,7 +57,7 @@ export declare enum DTMF_TRANSPORT {
}
export const REASON_PHRASE: Record<number, string>
-export const ALLOWED_METHODS = 'INVITE,ACK,CANCEL,BYE,UPDATE,MESSAGE,OPTIONS,REFER,INFO,NOTIFY'
+export const ALLOWED_METHODS = 'INVITE,ACK,CANCEL,BYE,UPDATE,MESSAGE,OPTIONS,REFER,INFO,NOTIFY,SUBSCRIBE'
export const ACCEPTED_BODY_TYPES = 'application/sdp, application/dtmf-relay'
export const MAX_FORWARDS = 69
export const SESSION_EXPIRES = 90
diff --git a/src/Constants.js b/src/Constants.js
index 5cb3a88..bb10550 100644
--- a/src/Constants.js
+++ b/src/Constants.js
@@ -149,7 +149,7 @@ module.exports = {
606 : 'Not Acceptable'
},
- ALLOWED_METHODS : 'INVITE,ACK,CANCEL,BYE,UPDATE,MESSAGE,OPTIONS,REFER,INFO,NOTIFY',
+ ALLOWED_METHODS : 'INVITE,ACK,CANCEL,BYE,UPDATE,MESSAGE,OPTIONS,REFER,INFO,NOTIFY,SUBSCRIBE',
ACCEPTED_BODY_TYPES : 'application/sdp, application/dtmf-relay',
MAX_FORWARDS : 69,
SESSION_EXPIRES : 90,
diff --git a/src/Notifier.d.ts b/src/Notifier.d.ts
new file mode 100644
index 0000000..aa863ea
--- /dev/null
+++ b/src/Notifier.d.ts
@@ -0,0 +1,40 @@
+import {EventEmitter} from 'events'
+import {IncomingRequest} from './SIPMessage'
+import {UA} from './UA'
+
+declare enum NotifierTerminationReason {
+ NOTIFY_RESPONSE_TIMEOUT = 0,
+ NOTIFY_TRANSPORT_ERROR = 1,
+ NOTIFY_NON_OK_RESPONSE = 2,
+ NOTIFY_AUTHENTICATION_FAILED = 3,
+ FINAL_NOTIFY_SENT = 4,
+ UNSUBSCRIBE_RECEIVED = 5,
+ SUBSCRIPTION_EXPIRED = 6
+}
+
+export interface MessageEventMap {
+ terminated: [terminationCode: NotifierTerminationReason];
+ subscribe: [isUnsubscribe: boolean, request: IncomingRequest, body: string | undefined, contentType: string | undefined];
+ expired: [];
+}
+
+interface NotifierOptions {
+ extraHeaders?: Array<string>;
+ allowEvents?: string;
+ pending?: boolean;
+ defaultExpires?: number;
+}
+
+export class Notifier extends EventEmitter<MessageEventMap> {
+ constructor(ua: UA, subscribe: IncomingRequest, contentType: string, options: NotifierOptions)
+ start(): void;
+ setActiveState(): void;
+ notify(body?: string): void;
+ terminate(body?: string, reason?: string, retryAfter?: number): void;
+ get state(): string;
+ get id(): string;
+ set data(_data: any);
+ get data(): any;
+ static get C(): typeof NotifierTerminationReason;
+ get C(): typeof NotifierTerminationReason;
+}
diff --git a/src/Notifier.js b/src/Notifier.js
new file mode 100644
index 0000000..529db4a
--- /dev/null
+++ b/src/Notifier.js
@@ -0,0 +1,469 @@
+const EventEmitter = require('events').EventEmitter;
+const Exceptions = require('./Exceptions');
+const Logger = require('./Logger');
+const JsSIP_C = require('./Constants');
+const Utils = require('./Utils');
+const Dialog = require('./Dialog');
+
+const logger = new Logger('Notifier');
+
+/**
+ * Termination codes.
+ */
+const C = {
+ // Termination codes.
+ NOTIFY_RESPONSE_TIMEOUT : 0,
+ NOTIFY_TRANSPORT_ERROR : 1,
+ NOTIFY_NON_OK_RESPONSE : 2,
+ NOTIFY_AUTHENTICATION_FAILED : 3,
+ FINAL_NOTIFY_SENT : 4,
+ UNSUBSCRIBE_RECEIVED : 5,
+ SUBSCRIPTION_EXPIRED : 6,
+
+ // Notifer states.
+ STATE_PENDING : 0,
+ STATE_ACTIVE : 1,
+ STATE_TERMINATED : 2,
+
+ // RFC 6665 3.1.1, default expires value.
+ DEFAULT_EXPIRES_SEC : 900
+};
+
+/**
+ * RFC 6665 Notifier implementation.
+ */
+module.exports = class Notifier extends EventEmitter
+{
+ /**
+ * Expose C object.
+ */
+ static get C()
+ {
+ return C;
+ }
+
+ static init_incoming(request, callback)
+ {
+ try
+ {
+ Notifier.checkSubscribe(request);
+ }
+ catch (error)
+ {
+ logger.warn('Notifier.init_incoming: invalid request. Error: ', error.message);
+
+ request.reply(405);
+
+ return;
+ }
+
+ callback();
+ }
+
+ static checkSubscribe(subscribe)
+ {
+ if (!subscribe)
+ {
+ throw new TypeError('Not enough arguments. Missing subscribe request');
+ }
+ if (subscribe.method !== JsSIP_C.SUBSCRIBE)
+ {
+ throw new TypeError('Invalid method for Subscribe request');
+ }
+ if (!subscribe.hasHeader('contact'))
+ {
+ throw new TypeError('Missing Contact header in subscribe request');
+ }
+ if (!subscribe.hasHeader('event'))
+ {
+ throw new TypeError('Missing Event header in subscribe request');
+ }
+ const expires = subscribe.getHeader('expires');
+
+ if (expires)
+ {
+ const parsed_expires = parseInt(expires);
+
+ if (!Utils.isDecimal(parsed_expires) || parsed_expires < 0)
+ {
+ throw new TypeError('Invalid Expires header field in subscribe request');
+ }
+ }
+ }
+
+ /**
+ * @param {UA} ua - JsSIP User Agent instance.
+ * @param {IncomingRequest} subscribe - Subscribe request.
+ * @param {string} contentType - Content-Type header value.
+ * @param {NotifierOptions} options - Optional parameters.
+ * @param {Array<string>} extraHeaders - Additional SIP headers.
+ * @param {string} allowEvents - Allow-Events header value.
+ * @param {boolean} pending - Set initial dialog state as "pending".
+ * @param {number} defaultExpires - Default expires value (seconds).
+ */
+ constructor(ua, subscribe, contentType, {
+ extraHeaders, allowEvents, pending, defaultExpires
+ })
+ {
+ logger.debug('new');
+
+ super();
+
+ if (!contentType)
+ {
+ throw new TypeError('Not enough arguments. Missing contentType');
+ }
+
+ Notifier.checkSubscribe(subscribe);
+
+ const eventName = subscribe.getHeader('event');
+
+ this._ua = ua;
+ this._initial_subscribe = subscribe;
+ this._expires_timestamp = null;
+ this._expires_timer = null;
+ this._defaultExpires = defaultExpires || C.DEFAULT_EXPIRES_SEC;
+
+ // Notifier state: pending, active, terminated.
+ this._state = pending ? C.STATE_PENDING : C.STATE_ACTIVE;
+
+ this._content_type = contentType;
+ this._headers = Utils.cloneArray(extraHeaders);
+ this._headers.push(`Event: ${eventName}`);
+
+ // Use contact from extraHeaders or create it.
+ this._contact = this._headers.find((header) => header.startsWith('Contact'));
+ if (!this._contact)
+ {
+ this._contact = `Contact: ${this._ua._contact.toString()}`;
+
+ this._headers.push(this._contact);
+ }
+
+ if (allowEvents)
+ {
+ this._headers.push(`Allow-Events: ${allowEvents}`);
+ }
+
+ this._target = subscribe.from.uri.user;
+ subscribe.to_tag = Utils.newTag();
+
+ // Custom session empty object for high level use.
+ this._data = {};
+ }
+
+ // Expose Notifier constants as a property of the Notifier instance.
+ get C()
+ {
+ return C;
+ }
+
+ /**
+ * Get dialog state.
+ */
+ get state()
+ {
+ return this._state;
+ }
+
+ /**
+ * Get dialog id.
+ */
+ get id()
+ {
+ return this._dialog ? this._dialog.id : null;
+ }
+
+ get data()
+ {
+ return this._data;
+ }
+
+ set data(_data)
+ {
+ this._data = _data;
+ }
+
+ /**
+ * Dialog callback.
+ * Called also for initial subscribe.
+ * Supported RFC 6665 4.4.3: initial fetch subscribe (with expires: 0).
+ */
+ receiveRequest(request)
+ {
+ if (request.method !== JsSIP_C.SUBSCRIBE)
+ {
+ request.reply(405);
+
+ return;
+ }
+
+ this._setExpires(request);
+
+ // Create dialog for normal and fetch-subscribe.
+ if (!this._dialog)
+ {
+ this._dialog = new Dialog(this, request, 'UAS');
+ }
+
+ request.reply(200, null, [ `Expires: ${this._expires}`, `${this._contact}` ]);
+
+ const body = request.body;
+ const content_type = request.getHeader('content-type');
+ const is_unsubscribe = this._expires === 0;
+
+ if (!is_unsubscribe)
+ {
+ this._setExpiresTimer();
+ }
+
+ logger.debug('emit "subscribe"');
+
+ this.emit('subscribe', is_unsubscribe, request, body, content_type);
+
+ if (is_unsubscribe)
+ {
+ this._terminateDialog(C.UNSUBSCRIBE_RECEIVED);
+ }
+ }
+
+ /**
+ * User API
+ */
+
+ /**
+ * Call this method after creating the Notifier instance and setting the event handlers.
+ */
+ start()
+ {
+ logger.debug('start()');
+
+ if (this._state === C.STATE_TERMINATED)
+ {
+ throw new Exceptions.InvalidStateError(this._state);
+ }
+
+ this.receiveRequest(this._initial_subscribe);
+ }
+
+ /**
+ * Switch pending dialog state to active.
+ */
+ setActiveState()
+ {
+ logger.debug('setActiveState()');
+
+ if (this._state === C.STATE_TERMINATED)
+ {
+ throw new Exceptions.InvalidStateError(this._state);
+ }
+
+ if (this._state === C.STATE_PENDING)
+ {
+ this._state = C.STATE_ACTIVE;
+ }
+ }
+
+ /**
+ * Send the initial and subsequent notify request.
+ * @param {string} body - notify request body.
+ */
+ notify(body=null)
+ {
+ logger.debug('notify()');
+
+ if (this._state === C.STATE_TERMINATED)
+ {
+ throw new Exceptions.InvalidStateError(this._state);
+ }
+
+ const expires = Math.floor((this._expires_timestamp - new Date().getTime()) / 1000);
+
+ // expires_timer is about to trigger. Clean up the timer and terminate.
+ if (expires <= 0)
+ {
+ if (!this._expires_timer)
+ {
+ logger.error('expires timer is not set');
+ }
+
+ clearTimeout(this._expires_timer);
+
+ this.terminate(body, 'timeout');
+ }
+ else
+ {
+ this._sendNotify([ `;expires=${expires}` ], body);
+ }
+ }
+
+ /**
+ * Terminate. (Send the final NOTIFY request).
+ *
+ * @param {string} body - Notify message body.
+ * @param {string} reason - Set Subscription-State reason parameter.
+ * @param {number} retryAfter - Set Subscription-State retry-after parameter.
+ */
+ terminate(body = null, reason = null, retryAfter = null)
+ {
+ logger.debug('terminate()');
+
+ if (this._state === C.STATE_TERMINATED)
+ {
+ return;
+ }
+
+ const subsStateParameters = [];
+
+ if (reason)
+ {
+ subsStateParameters.push(`;reason=${reason}`);
+ }
+
+ if (retryAfter !== null)
+ {
+ subsStateParameters.push(`;retry-after=${retryAfter}`);
+ }
+
+ this._sendNotify(subsStateParameters, body, null, 'terminated');
+
+ this._terminateDialog(reason === 'timeout' ? C.SUBSCRIPTION_EXPIRED : C.FINAL_NOTIFY_SENT);
+ }
+
+ /**
+ * Private API
+ */
+
+ _terminateDialog(termination_code)
+ {
+ if (this._state === C.STATE_TERMINATED)
+ {
+ return;
+ }
+ this._state = C.STATE_TERMINATED;
+
+ clearTimeout(this._expires_timer);
+ if (this._dialog)
+ {
+ this._dialog.terminate();
+ this._dialog = null;
+ }
+ logger.debug(`emit "terminated" code=${termination_code}`);
+
+ this.emit('terminated', termination_code);
+ }
+
+ _setExpires(request)
+ {
+ if (request.hasHeader('expires'))
+ {
+ this._expires = parseInt(request.getHeader('expires'));
+ }
+ else
+ {
+ this._expires = this._defaultExpires;
+ logger.debug(`missing Expires header field, default value set: ${this._expires}`);
+ }
+ }
+
+ /**
+ * @param {Array<string>} subsStateParams subscription state parameters.
+ * @param {String} body Notify body
+ * @param {Array<string>} extraHeaders
+ */
+ _sendNotify(subsStateParameters, body=null, extraHeaders=null, state=null)
+ {
+ // Prevent send notify after final notify.
+ if (this._state === C.STATE_TERMINATED)
+ {
+ logger.warn('final notify already sent');
+
+ return;
+ }
+
+ // Build Subscription-State header with parameters.
+ let subsState = `Subscription-State: ${state || this._parseState()}`;
+
+ for (const param of subsStateParameters)
+ {
+ subsState += param;
+ }
+
+ let headers = Utils.cloneArray(this._headers);
+
+ headers.push(subsState);
+
+ if (extraHeaders)
+ {
+ headers = headers.concat(extraHeaders);
+ }
+
+ if (body)
+ {
+ headers.push(`Content-Type: ${this._content_type}`);
+ }
+
+ this._dialog.sendRequest(JsSIP_C.NOTIFY, {
+ body,
+ extraHeaders : headers,
+ eventHandlers : {
+ onRequestTimeout : () =>
+ {
+ this._terminateDialog(C.NOTIFY_RESPONSE_TIMEOUT);
+ },
+ onTransportError : () =>
+ {
+ this._terminateDialog(C.NOTIFY_TRANSPORT_ERROR);
+ },
+ onErrorResponse : (response) =>
+ {
+ if (response.status_code === 401 || response.status_code === 407)
+ {
+ this._terminateDialog(C.NOTIFY_AUTHENTICATION_FAILED);
+ }
+ else
+ {
+ this._terminateDialog(C.NOTIFY_NON_OK_RESPONSE);
+ }
+ },
+ onDialogError : () =>
+ {
+ this._terminateDialog(C.NOTIFY_NON_OK_RESPONSE);
+ }
+ }
+ });
+ }
+
+ _setExpiresTimer()
+ {
+ this._expires_timestamp = new Date().getTime() + (this._expires * 1000);
+
+ clearTimeout(this._expires_timer);
+ this._expires_timer = setTimeout(() =>
+ {
+ if (this._state === C.STATE_TERMINATED)
+ {
+ return;
+ }
+
+ logger.debug('emit "expired"');
+
+ // Client can hook into the 'expired' event and call terminate to send a custom notify.
+ this.emit('expired');
+
+ // This will be no-op if the client already called `terminate()`.
+ this.terminate(null, 'timeout');
+ }, this._expires * 1000);
+ }
+
+ _parseState()
+ {
+ switch (this._state)
+ {
+ case C.STATE_PENDING: return 'pending';
+ case C.STATE_ACTIVE: return 'active';
+ case C.STATE_TERMINATED: return 'terminated';
+ default: throw new TypeError('wrong state value');
+ }
+ }
+};
diff --git a/src/Subscriber.d.ts b/src/Subscriber.d.ts
new file mode 100644
index 0000000..060f0ea
--- /dev/null
+++ b/src/Subscriber.d.ts
@@ -0,0 +1,42 @@
+import {EventEmitter} from 'events'
+import {IncomingRequest} from './SIPMessage'
+import {UA} from './UA'
+
+declare enum SubscriberTerminatedCode {
+ SUBSCRIBE_RESPONSE_TIMEOUT = 0,
+ SUBSCRIBE_TRANSPORT_ERROR = 1,
+ SUBSCRIBE_NON_OK_RESPONSE = 2,
+ SUBSCRIBE_WRONG_OK_RESPONSE = 3,
+ SUBSCRIBE_AUTHENTICATION_FAILED = 4,
+ UNSUBSCRIBE_TIMEOUT = 5,
+ FINAL_NOTIFY_RECEIVED = 6,
+ WRONG_NOTIFY_RECEIVED = 7
+}
+
+export interface MessageEventMap {
+ pending: [];
+ accepted: [];
+ active: [];
+ terminated: [terminationCode: SubscriberTerminatedCode, reason: string | undefined, retryAfter: number | undefined];
+ notify: [isFinal: boolean, request: IncomingRequest, body: string | undefined, contentType: string | undefined];
+}
+
+interface SubscriberOptions {
+ expires?: number;
+ contentType: string;
+ allowEvents?: string;
+ params?: Record<string, any>;
+ extraHeaders?: Array<string>;
+}
+
+export class Subscriber extends EventEmitter<MessageEventMap> {
+ constructor(ua: UA, target: string, eventName: string, accept: string, options: SubscriberOptions)
+ subscribe(body?: string): void;
+ terminate(body?: string): void;
+ get state(): string;
+ get id(): string;
+ set data(_data: any);
+ get data(): any;
+ static get C(): typeof SubscriberTerminatedCode;
+ get C(): typeof SubscriberTerminatedCode;
+}
diff --git a/src/Subscriber.js b/src/Subscriber.js
new file mode 100644
index 0000000..f8d5fe9
--- /dev/null
+++ b/src/Subscriber.js
@@ -0,0 +1,622 @@
+const EventEmitter = require('events').EventEmitter;
+const Exceptions = require('./Exceptions');
+const Logger = require('./Logger');
+const JsSIP_C = require('./Constants');
+const Utils = require('./Utils');
+const Grammar = require('./Grammar');
+const SIPMessage = require('./SIPMessage');
+const RequestSender = require('./RequestSender');
+const Dialog = require('./Dialog');
+
+const logger = new Logger('Subscriber');
+
+/**
+ * Termination codes.
+ */
+const C = {
+ // Termination codes.
+ SUBSCRIBE_RESPONSE_TIMEOUT : 0,
+ SUBSCRIBE_TRANSPORT_ERROR : 1,
+ SUBSCRIBE_NON_OK_RESPONSE : 2,
+ SUBSCRIBE_WRONG_OK_RESPONSE : 3,
+ SUBSCRIBE_AUTHENTICATION_FAILED : 4,
+ UNSUBSCRIBE_TIMEOUT : 5,
+ FINAL_NOTIFY_RECEIVED : 6,
+ WRONG_NOTIFY_RECEIVED : 7,
+
+ // Subscriber states.
+ STATE_PENDING : 0,
+ STATE_ACTIVE : 1,
+ STATE_TERMINATED : 2,
+ STATE_INIT : 3,
+ STATE_WAITING_NOTIFY : 4,
+
+ // RFC 6665 3.1.1, default expires value.
+ DEFAULT_EXPIRES_SEC : 900
+};
+
+/**
+ * RFC 6665 Subscriber implementation.
+ */
+module.exports = class Subscriber extends EventEmitter
+{
+ /**
+ * Expose C object.
+ */
+ static get C()
+ {
+ return C;
+ }
+
+ /**
+ * @param {UA} ua - reference to JsSIP.UA
+ * @param {string} target
+ * @param {string} eventName - Event header value. May end with optional ;id=xxx
+ * @param {string} accept - Accept header value.
+ *
+ * @param {SubscriberOption} options - optional parameters.
+ * @param {number} expires - Expires header value. Default is 900.
+ * @param {string} contentType - Content-Type header value. Used for SUBSCRIBE with body
+ * @param {string} allowEvents - Allow-Events header value.
+ * @param {RequestParams} params - Will have priority over ua.configuration.
+ * If set please define: to_uri, to_display_name, from_uri, from_display_name
+ * @param {Array<string>} extraHeaders - Additional SIP headers.
+ */
+ constructor(ua, target, eventName, accept, { expires, contentType,
+ allowEvents, params, extraHeaders })
+ {
+ logger.debug('new');
+
+ super();
+
+ // Check that arguments are defined.
+ if (!target)
+ {
+ throw new TypeError('Not enough arguments: Missing target');
+ }
+
+ if (!eventName)
+ {
+ throw new TypeError('Not enough arguments: Missing eventName');
+ }
+
+ if (!accept)
+ {
+ throw new TypeError('Not enough arguments: Missing accept');
+ }
+
+ const event_header = Grammar.parse(eventName, 'Event');
+
+ if (event_header === -1)
+ {
+ throw new TypeError('Missing Event header field');
+ }
+
+ this._ua = ua;
+ this._target = target;
+
+ if (!Utils.isDecimal(expires) || expires <= 0)
+ {
+ expires = C.DEFAULT_EXPIRES_SEC;
+ }
+
+ this._expires = expires;
+
+ // Used to subscribe with body.
+ this._content_type = contentType;
+
+ // Set initial subscribe parameters.
+ this._params = Utils.cloneObject(params);
+
+ if (!this._params.from_uri)
+ {
+ this._params.from_uri = this._ua.configuration.uri;
+ }
+
+ this._params.from_tag = Utils.newTag();
+ this._params.to_tag = null;
+ this._params.call_id = Utils.createRandomToken(20);
+
+ // Create subscribe cseq if not defined custom cseq.
+ if (this._params.cseq === undefined)
+ {
+ this._params.cseq = Math.floor((Math.random() * 10000) + 1);
+ }
+
+ // Subscriber state.
+ this._state = C.STATE_INIT;
+
+ // Dialog.
+ this._dialog = null;
+
+ // To refresh subscription.
+ this._expires_timer = null;
+ this._expires_timestamp = null;
+
+ // To prevent duplicate terminated call.
+ this._terminated = false;
+
+ this._event_name = event_header.event;
+ this._event_id = event_header.params && event_header.params.id;
+
+ let eventValue = this._event_name;
+
+ if (this._event_id)
+ {
+ eventValue += `;id=${this._event_id}`;
+ }
+
+ this._headers = Utils.cloneArray(extraHeaders);
+ this._headers = this._headers.concat([
+ `Event: ${eventValue}`,
+ `Expires: ${this._expires}`,
+ `Accept: ${accept}`
+ ]);
+
+ if (!this._headers.find((header) => header.startsWith('Contact')))
+ {
+ const contact = `Contact: ${this._ua._contact.toString()}`;
+
+ this._headers.push(contact);
+ }
+
+ if (allowEvents)
+ {
+ this._headers.push(`Allow-Events: ${allowEvents}`);
+ }
+
+ // To enqueue SUBSCRIBE requests created before the reception of the initial subscribe OK response.
+ this._queue = [];
+
+ // Custom session empty object for high level use.
+ this._data = {};
+ }
+
+ // Expose Subscriber constants as a property of the Subscriber instance.
+ get C()
+ {
+ return C;
+ }
+
+ /**
+ * Get dialog state.
+ */
+ get state()
+ {
+ return this._state;
+ }
+
+ /**
+ * Get dialog id.
+ */
+ get id()
+ {
+ return this._dialog ? this._dialog.id : null;
+ }
+
+ get data()
+ {
+ return this._data;
+ }
+
+ set data(_data)
+ {
+ this._data = _data;
+ }
+
+ onRequestTimeout()
+ {
+ this._terminateDialog(C.SUBSCRIBE_RESPONSE_TIMEOUT);
+ }
+
+ onTransportError()
+ {
+ this._terminateDialog(C.SUBSCRIBE_TRANSPORT_ERROR);
+ }
+
+ /**
+ * Dialog callback.
+ */
+ receiveRequest(request)
+ {
+ if (request.method !== JsSIP_C.NOTIFY)
+ {
+ logger.warn('received non-NOTIFY request');
+ request.reply(405);
+
+ return;
+ }
+
+ // RFC 6665 8.2.1. Check if event header matches.
+ const event_header = request.parseHeader('Event');
+
+ if (!event_header)
+ {
+ logger.warn('missing Event header');
+ request.reply(400);
+ this._terminateDialog(C.WRONG_NOTIFY_RECEIVED);
+
+ return;
+ }
+
+ const event_name = event_header.event;
+ const event_id = event_header.params && event_header.params.id;
+
+ if (event_name !== this._event_name || event_id !== this._event_id)
+ {
+ logger.warn('Event header does not match the one in SUBSCRIBE request');
+ request.reply(489);
+ this._terminateDialog(C.WRONG_NOTIFY_RECEIVED);
+
+ return;
+ }
+
+ // Process Subscription-State header.
+ const subs_state = request.parseHeader('subscription-state');
+
+ if (!subs_state)
+ {
+ logger.warn('missing Subscription-State header');
+ request.reply(400);
+ this._terminateDialog(C.WRONG_NOTIFY_RECEIVED);
+
+ return;
+ }
+
+ const new_state = this._parseSubscriptionState(subs_state.state);
+
+ if (new_state === undefined)
+ {
+ logger.warn(`Invalid Subscription-State header value: ${subs_state.state}`);
+ request.reply(400);
+ this._terminateDialog(C.WRONG_NOTIFY_RECEIVED);
+
+ return;
+ }
+ request.reply(200);
+
+ const prev_state = this._state;
+
+ if (prev_state !== C.STATE_TERMINATED && new_state !== C.STATE_TERMINATED)
+ {
+ this._state = new_state;
+
+ if (subs_state.expires !== undefined)
+ {
+ const expires = subs_state.expires;
+ const expires_timestamp = new Date().getTime() + (expires * 1000);
+ const max_time_deviation = 2000;
+
+ // Expiration time is shorter and the difference is not too small.
+ if (this._expires_timestamp - expires_timestamp > max_time_deviation)
+ {
+ logger.debug('update sending re-SUBSCRIBE time');
+
+ this._scheduleSubscribe(expires);
+ }
+ }
+ }
+
+ if (prev_state !== C.STATE_PENDING && new_state === C.STATE_PENDING)
+ {
+ logger.debug('emit "pending"');
+
+ this.emit('pending');
+ }
+ else if (prev_state !== C.STATE_ACTIVE && new_state === C.STATE_ACTIVE)
+ {
+ logger.debug('emit "active"');
+
+ this.emit('active');
+ }
+
+ const body = request.body;
+
+ // Check if the notify is final.
+ const is_final = new_state === C.STATE_TERMINATED;
+
+ // Notify event fired only for notify with body.
+ if (body)
+ {
+ const content_type = request.getHeader('content-type');
+
+ logger.debug('emit "notify"');
+
+ this.emit('notify', is_final, request, body, content_type);
+ }
+
+ if (is_final)
+ {
+ const reason = subs_state.reason;
+ let retry_after = undefined;
+
+ if (subs_state.params && subs_state.params['retry-after'] !== undefined)
+ {
+ retry_after = parseInt(subs_state.params['retry-after']);
+ }
+
+ this._terminateDialog(C.FINAL_NOTIFY_RECEIVED, reason, retry_after);
+ }
+ }
+
+ /**
+ * User API
+ */
+
+ /**
+ * Send the initial (non-fetch) and subsequent subscribe.
+ * @param {string} body - subscribe request body.
+ */
+ subscribe(body = null)
+ {
+ logger.debug('subscribe()');
+
+ if (this._state === C.STATE_INIT)
+ {
+ this._sendInitialSubscribe(body, this._headers);
+ }
+ else
+ {
+ this._sendSubsequentSubscribe(body, this._headers);
+ }
+ }
+
+ /**
+ * terminate.
+ * Send un-subscribe or fetch-subscribe (with Expires: 0).
+ * @param {string} body - un-subscribe request body
+ */
+ terminate(body = null)
+ {
+ logger.debug('terminate()');
+
+ if (this._state === C.STATE_INIT)
+ {
+ throw new Exceptions.InvalidStateError(this._state);
+ }
+
+ // Prevent duplication un-subscribe sending.
+ if (this._terminated)
+ {
+ return;
+ }
+ this._terminated = true;
+
+ // Set header Expires: 0.
+ const headers = this._headers.map((header) =>
+ {
+ return header.startsWith('Expires') ? 'Expires: 0' : header;
+ });
+
+ this._sendSubsequentSubscribe(body, headers);
+ }
+
+ /**
+ * Private API.
+ */
+ _terminateDialog(terminationCode, reason = undefined, retryAfter = undefined)
+ {
+ // To prevent duplicate emit terminated event.
+ if (this._state === C.STATE_TERMINATED)
+ {
+ return;
+ }
+
+ this._state = C.STATE_TERMINATED;
+
+ // Clear timers.
+ clearTimeout(this._expires_timer);
+
+ if (this._dialog)
+ {
+ this._dialog.terminate();
+ this._dialog = null;
+ }
+
+ logger.debug(`emit "terminated" code=${terminationCode}`);
+
+ this.emit('terminated', terminationCode, reason, retryAfter);
+ }
+
+ _sendInitialSubscribe(body, headers)
+ {
+ if (body)
+ {
+ if (!this._content_type)
+ {
+ throw new TypeError('content_type is undefined');
+ }
+
+ headers = Utils.cloneArray(headers);
+ headers.push(`Content-Type: ${this._content_type}`);
+ }
+
+ this._state = C.STATE_WAITING_NOTIFY;
+
+ const request = new SIPMessage.OutgoingRequest(JsSIP_C.SUBSCRIBE,
+ this._ua.normalizeTarget(this._target), this._ua, this._params, headers, body);
+
+ const request_sender = new RequestSender(this._ua, request, {
+ onRequestTimeout : () =>
+ {
+ this.onRequestTimeout();
+ },
+ onTransportError : () =>
+ {
+ this.onTransportError();
+ },
+ onReceiveResponse : (response) =>
+ {
+ this._receiveSubscribeResponse(response);
+ }
+ });
+
+ request_sender.send();
+ }
+
+ _sendSubsequentSubscribe(body, headers)
+ {
+ if (this._state === C.STATE_TERMINATED)
+ {
+ return;
+ }
+
+ if (!this._dialog)
+ {
+ logger.debug('enqueue subscribe');
+
+ this._queue.push({ body, headers: Utils.cloneArray(headers) });
+
+ return;
+ }
+
+ if (body)
+ {
+ if (!this._content_type)
+ {
+ throw new TypeError('content_type is undefined');
+ }
+
+ headers = Utils.cloneArray(headers);
+ headers.push(`Content-Type: ${this._content_type}`);
+ }
+
+ this._dialog.sendRequest(JsSIP_C.SUBSCRIBE, {
+ body,
+ extraHeaders : headers,
+ eventHandlers : {
+ onRequestTimeout : () =>
+ {
+ this.onRequestTimeout();
+ },
+ onTransportError : () =>
+ {
+ this.onTransportError();
+ },
+ onSuccessResponse : (response) =>
+ {
+ this._receiveSubscribeResponse(response);
+ },
+ onErrorResponse : (response) =>
+ {
+ this._receiveSubscribeResponse(response);
+ },
+ onDialogError : (response) =>
+ {
+ this._receiveSubscribeResponse(response);
+ }
+ }
+ });
+ }
+
+ _receiveSubscribeResponse(response)
+ {
+ if (this._state === C.STATE_TERMINATED)
+ {
+ return;
+ }
+
+ if (response.status_code >= 200 && response.status_code < 300)
+ {
+ // Create dialog.
+ if (this._dialog === null)
+ {
+ const dialog = new Dialog(this, response, 'UAC');
+
+ if (dialog.error)
+ {
+ // OK response without Contact.
+ logger.warn(dialog.error);
+ this._terminateDialog(C.SUBSCRIBE_WRONG_OK_RESPONSE);
+
+ return;
+ }
+
+ this._dialog = dialog;
+
+ logger.debug('emit "accepted"');
+
+ this.emit('accepted');
+
+ // Subsequent subscribes saved in the queue until dialog created.
+ for (const subscribe of this._queue)
+ {
+ logger.debug('dequeue subscribe');
+
+ this._sendSubsequentSubscribe(subscribe.body, subscribe.headers);
+ }
+ }
+
+ // Check expires value.
+ const expires_value = response.getHeader('expires');
+
+ let expires = parseInt(expires_value);
+
+ if (!Utils.isDecimal(expires) || expires <= 0)
+ {
+ logger.warn(`response without Expires header, setting a default value of ${C.DEFAULT_EXPIRES_SEC}`);
+
+ // RFC 6665 3.1.1 subscribe OK response must contain Expires header.
+ // Use workaround expires value.
+ expires = C.DEFAULT_EXPIRES_SEC;
+ }
+
+ if (expires > 0)
+ {
+ this._scheduleSubscribe(expires);
+ }
+ }
+ else if (response.status_code === 401 || response.status_code === 407)
+ {
+ this._terminateDialog(C.SUBSCRIBE_AUTHENTICATION_FAILED);
+ }
+ else if (response.status_code >= 300)
+ {
+ this._terminateDialog(C.SUBSCRIBE_NON_OK_RESPONSE);
+ }
+ }
+
+ _scheduleSubscribe(expires)
+ {
+ /*
+ If the expires time is less than 140 seconds we do not support Chrome intensive timer throttling mode.
+ In this case, the re-subcribe is sent 5 seconds before the subscription expiration.
+
+ When Chrome is in intensive timer throttling mode, in the worst case,
+ the timer will be 60 seconds late.
+ We give the server 10 seconds to make sure it will execute the command even if it is heavily loaded.
+ As a result, we order the time no later than 70 seconds before the subscription expiration.
+ Resulting time calculated as half time interval + (half interval - 70) * random.
+
+ E.g. expires is 140, re-subscribe will be ordered to send in 70 seconds.
+ expires is 600, re-subscribe will be ordered to send in 300 + (0 .. 230) seconds.
+ */
+
+ const timeout = expires >= 140 ? (expires * 1000 / 2)
+ + Math.floor(((expires / 2) - 70) * 1000 * Math.random()) : (expires * 1000) - 5000;
+
+ this._expires_timestamp = new Date().getTime() + (expires * 1000);
+
+ logger.debug(`next SUBSCRIBE will be sent in ${Math.floor(timeout / 1000)} sec`);
+
+ clearTimeout(this._expires_timer);
+ this._expires_timer = setTimeout(() =>
+ {
+ this._expires_timer = null;
+ this._sendSubsequentSubscribe(null, this._headers);
+ }, timeout);
+ }
+
+ _parseSubscriptionState(strState)
+ {
+ switch (strState)
+ {
+ case 'pending': return C.STATE_PENDING;
+ case 'active': return C.STATE_ACTIVE;
+ case 'terminated': return C.STATE_TERMINATED;
+ case 'init': return C.STATE_INIT;
+ case 'notify_wait': return C.STATE_WAITING_NOTIFY;
+ default: return undefined;
+ }
+ }
+};
diff --git a/src/UA.d.ts b/src/UA.d.ts
index 6ee4a78..0fff768 100644
--- a/src/UA.d.ts
+++ b/src/UA.d.ts
@@ -5,6 +5,8 @@ import {AnswerOptions, Originator, RTCSession, RTCSessionEventMap, TerminateOpti
import {IncomingRequest, IncomingResponse, OutgoingRequest} from './SIPMessage'
import {Message, SendMessageOptions} from './Message'
import {Registrator} from './Registrator'
+import {Notifier} from './Notifier'
+import {Subscriber} from './Subscriber'
import {URI} from './URI'
import {causes} from './Constants'
@@ -125,6 +127,7 @@ export type IncomingOptionsListener = (event: IncomingOptionsEvent) => void;
export type OutgoingOptionsListener = (event: OutgoingOptionsEvent) => void;
export type OptionsListener = IncomingOptionsListener | OutgoingOptionsListener;
export type SipEventListener = <T = any>(event: { event: T; request: IncomingRequest; }) => void
+export type SipSubscribeListener = <T = any>(event: { event: T; request: IncomingRequest; }) => void
export interface UAEventMap {
connecting: ConnectingListener;
@@ -137,6 +140,7 @@ export interface UAEventMap {
newRTCSession: RTCSessionListener;
newMessage: MessageListener;
sipEvent: SipEventListener;
+ newSubscribe: SipSubscribeListener;
newOptions: OptionsListener;
}
@@ -153,6 +157,38 @@ export interface UAContact {
toString(options?: UAContactOptions): string
}
+export interface RequestParams {
+ from_uri: URI;
+ from_display_name?: string;
+ from_tag: string;
+ to_uri: URI;
+ to_display_name?: string;
+ to_tag?: string;
+ call_id: string;
+ cseq: number;
+}
+
+export interface SubscriberParams {
+ from_uri: URI;
+ from_display_name?: string;
+ to_uri: URI;
+ to_display_name?: string;
+}
+
+export interface SubscriberOptions {
+ expires?: number;
+ contentType?: string;
+ allowEvents?: string;
+ params?: SubscriberParams;
+ extraHeaders?: string[];
+}
+
+export interface NotifierOptions {
+ allowEvents?: string;
+ extraHeaders?: string[];
+ pending?: boolean;
+}
+
declare enum UAStatus {
// UA status codes.
STATUS_INIT = 0,
@@ -189,6 +225,10 @@ export class UA extends EventEmitter {
sendMessage(target: string | URI, body: string, options?: SendMessageOptions): Message;
+ subscribe(target: string, eventName: string, accept: string, options?: SubscriberOptions): Subscriber;
+
+ notify( subscribe: IncomingRequest, contentType: string, options?: NotifierOptions): Notifier;
+
terminateSessions(options?: TerminateOptions): void;
isRegistered(): boolean;
diff --git a/src/UA.js b/src/UA.js
index 4d66fdc..066df6c 100644
--- a/src/UA.js
+++ b/src/UA.js
@@ -3,6 +3,8 @@ const Logger = require('./Logger');
const JsSIP_C = require('./Constants');
const Registrator = require('./Registrator');
const RTCSession = require('./RTCSession');
+const Subscriber = require('./Subscriber');
+const Notifier = require('./Notifier');
const Message = require('./Message');
const Options = require('./Options');
const Transactions = require('./Transactions');
@@ -262,6 +264,26 @@ module.exports = class UA extends EventEmitter
return message;
}
+ /**
+ * Create subscriber instance
+ */
+ subscribe(target, eventName, accept, options)
+ {
+ logger.debug('subscribe()');
+
+ return new Subscriber(this, target, eventName, accept, options);
+ }
+
+ /**
+ * Create notifier instance
+ */
+ notify(subscribe, contentType, options)
+ {
+ logger.debug('notify()');
+
+ return new Notifier(this, subscribe, contentType, options);
+ }
+
/**
* Send a SIP OPTIONS.
*
@@ -647,6 +669,15 @@ module.exports = class UA extends EventEmitter
message.init_incoming(request);
}
+ else if (method === JsSIP_C.SUBSCRIBE)
+ {
+ if (this.listeners('newSubscribe').length === 0)
+ {
+ request.reply(405);
+
+ return;
+ }
+ }
else if (method === JsSIP_C.INVITE)
{
// Initial INVITE.
@@ -733,6 +764,15 @@ module.exports = class UA extends EventEmitter
});
request.reply(200);
break;
+ case JsSIP_C.SUBSCRIBE:
+ Notifier.init_incoming(request, () =>
+ {
+ this.emit('newSubscribe', {
+ event : request.event,
+ request
+ });
+ });
+ break;
default:
request.reply(405);
break;
diff --git a/test/include/loopSocket.js b/test/include/loopSocket.js
new file mode 100644
index 0000000..9a92d9a
--- /dev/null
+++ b/test/include/loopSocket.js
@@ -0,0 +1,48 @@
+// LoopSocket send message itself.
+// Used P2P logic: message call-id is modified in each leg.
+module.exports = class LoopSocket
+{
+ constructor()
+ {
+ this.url = 'ws://localhost:12345';
+ this.via_transport = 'WS';
+ this.sip_uri = 'sip:localhost:12345;transport=ws';
+ }
+
+ connect()
+ {
+ setTimeout(() => { this.onconnect(); }, 0);
+ }
+
+ disconnect()
+ {
+ }
+
+ send(message)
+ {
+ const message2 = this._modifyCallId(message);
+
+ setTimeout(() => { this.ondata(message2); }, 0);
+
+ return true;
+ }
+
+ // Call-ID: add or drop word '_second'.
+ _modifyCallId(message)
+ {
+ const ixBegin = message.indexOf('Call-ID');
+ const ixEnd = message.indexOf('\r', ixBegin);
+ let callId = message.substring(ixBegin+9, ixEnd);
+
+ if (callId.endsWith('_second'))
+ {
+ callId = callId.substring(0, callId.length - 7);
+ }
+ else
+ {
+ callId += '_second';
+ }
+
+ return `${message.substring(0, ixBegin)}Call-ID: ${callId}${message.substring(ixEnd)}`;
+ }
+};
diff --git a/test/test-UA-subscriber-notifier.js b/test/test-UA-subscriber-notifier.js
new file mode 100644
index 0000000..62b5979
--- /dev/null
+++ b/test/test-UA-subscriber-notifier.js
@@ -0,0 +1,186 @@
+require('./include/common');
+const JsSIP = require('../');
+const LoopSocket = require('./include/loopSocket');
+
+describe('subscriber/notifier communication', () =>
+{
+ test('should handle subscriber/notifier communication', () => new Promise((resolve) =>
+ {
+ let eventSequence = 0;
+
+ const TARGET = 'ikq';
+ const REQUEST_URI = 'sip:ikq@example.com';
+ const CONTACT_URI = 'sip:ikq@abcdefabcdef.invalid;transport=ws';
+ const SUBSCRIBE_ACCEPT = 'application/text, text/plain';
+ const EVENT_NAME = 'weather';
+ const CONTENT_TYPE = 'text/plain';
+ const WEATHER_REQUEST = 'Please report the weather condition';
+ const WEATHER_REPORT = '+20..+24°C, no precipitation, light wind';
+
+ /**
+ * @param {JsSIP.UA} ua
+ */
+ function createSubscriber(ua)
+ {
+ const options = {
+ expires : 3600,
+ contentType : CONTENT_TYPE,
+ params : null
+ };
+
+ const subscriber = ua.subscribe(TARGET, EVENT_NAME, SUBSCRIBE_ACCEPT, options);
+
+ subscriber.on('active', () =>
+ {
+ // 'receive notify with subscription-state: active'
+ expect(++eventSequence).toBe(6);
+ });
+
+ subscriber.on('notify', (isFinal, notify, body, contType) =>
+ {
+ eventSequence++;
+ // 'receive notify'
+ expect(eventSequence === 7 || eventSequence === 11).toBe(true);
+
+ expect(notify.method).toBe('NOTIFY');
+ expect(notify.getHeader('contact')).toBe(`<${CONTACT_URI}>`); // 'notify contact'
+ expect(body).toBe(WEATHER_REPORT); // 'notify body'
+ expect(contType).toBe(CONTENT_TYPE); // 'notify content-type'
+
+ const subsState = notify.parseHeader('subscription-state').state;
+
+ expect(subsState === 'pending' || subsState === 'active' || subsState === 'terminated').toBe(true); // 'notify subscription-state'
+
+ // After receiving the first notify, send un-subscribe.
+ if (eventSequence === 7)
+ {
+ ++eventSequence; // 'send un-subscribe'
+
+ subscriber.terminate(WEATHER_REQUEST);
+ }
+ });
+
+ subscriber.on('terminated', (terminationCode, reason, retryAfter) =>
+ {
+ expect(++eventSequence).toBe(12); // 'subscriber terminated'
+ expect(terminationCode).toBe(subscriber.C.FINAL_NOTIFY_RECEIVED);
+ expect(reason).toBeUndefined();
+ expect(retryAfter).toBeUndefined();
+
+ ua.stop();
+ });
+
+ subscriber.on('accepted', () =>
+ {
+ expect(++eventSequence).toBe(5); // 'initial subscribe accepted'
+ });
+
+ expect(++eventSequence).toBe(2); // 'send subscribe'
+
+ subscriber.subscribe(WEATHER_REQUEST);
+ }
+
+ /**
+ * @param {JsSIP.UA} ua
+ */
+ function createNotifier(ua, subscribe)
+ {
+ const notifier = ua.notify(subscribe, CONTENT_TYPE, { pending: false });
+
+ // Receive subscribe (includes initial)
+ notifier.on('subscribe', (isUnsubscribe, subs, body, contType) =>
+ {
+ expect(subscribe.method).toBe('SUBSCRIBE');
+ expect(subscribe.getHeader('contact')).toBe(`<${CONTACT_URI}>`); // 'subscribe contact'
+ expect(subscribe.getHeader('accept')).toBe(SUBSCRIBE_ACCEPT); // 'subscribe accept'
+ expect(body).toBe(WEATHER_REQUEST); // 'subscribe body'
+ expect(contType).toBe(CONTENT_TYPE); // 'subscribe content-type'
+
+ expect(++eventSequence).toBe(isUnsubscribe ? 9 : 4);
+ if (isUnsubscribe)
+ {
+ // 'send final notify'
+ notifier.terminate(WEATHER_REPORT);
+ }
+ else
+ {
+ // 'send notify'
+ notifier.notify(WEATHER_REPORT);
+ }
+ });
+
+ // Example only. Never reached.
+ notifier.on('expired', () =>
+ {
+ notifier.terminate(WEATHER_REPORT, 'timeout');
+ });
+
+ notifier.on('terminated', () =>
+ {
+ expect(++eventSequence).toBe(10); // 'notifier terminated'
+ });
+
+ notifier.start();
+ }
+
+ // Start JsSIP UA with loop socket.
+ const config =
+ {
+ sockets : new LoopSocket(), // message sending itself, with modified Call-ID
+ uri : REQUEST_URI,
+ contact_uri : CONTACT_URI,
+ register : false
+ };
+
+ const ua = new JsSIP.UA(config);
+
+ // Uncomment to see SIP communication
+ // JsSIP.debug.enable('JsSIP:*');
+
+ ua.on('newSubscribe', (e) =>
+ {
+ expect(++eventSequence).toBe(3); // 'receive initial subscribe'
+
+ const subs = e.request;
+ const ev = subs.parseHeader('event');
+
+ expect(subs.ruri.toString()).toBe(REQUEST_URI); // 'initial subscribe uri'
+ expect(ev.event).toBe(EVENT_NAME); // 'subscribe event'
+
+ if (ev.event !== EVENT_NAME)
+ {
+ subs.reply(489); // "Bad Event"
+
+ return;
+ }
+
+ const accepts = subs.getHeaders('accept');
+ const canUse = accepts && accepts.some((v) => v.includes(CONTENT_TYPE));
+
+ expect(canUse).toBe(true); // 'notifier can use subscribe accept header'
+
+ if (!canUse)
+ {
+ subs.reply(406); // "Not Acceptable"
+
+ return;
+ }
+
+ createNotifier(ua, subs);
+ });
+
+ ua.on('connected', () =>
+ {
+ expect(++eventSequence).toBe(1); // 'socket connected'
+
+ createSubscriber(ua);
+ });
+
+ ua.on('disconnected', () =>
+ {
+ resolve();
+ });
+
+ ua.start();
+ }));
+});