19565048871

Committed 21 Nov 2025 08:48AM UTC coverage: 82.843% (+0.4%) from 82.465%

Build # 19565048871

Build Type

push

github

Committed by

web-flow

Commit Message

feat: add ParameterWatcher for real time parameter monitoring (#1326)

Implement ParameterWatcher class to enable real-time monitoring of parameter changes on remote nodes. Subscribes to /parameter_events topic and emits 'change' events when watched parameters are modified.

Features:
- Watch specific parameters on remote nodes
- Real-time change notifications via EventEmitter
- Dynamic parameter list management (add/remove)
- Async/await support with timeout and AbortSignal
- Proper lifecycle management and cleanup

Coverage Stats

1074 of 1420 branches covered (75.63%)

Branch coverage included in aggregate %.

92 of 98 new or added lines in 4 files covered. (93.88%)

2446 of 2829 relevant lines covered (86.46%)

488.13 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.55

/lib/parameter_watcher.js

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

'use strict';

const EventEmitter = require('events');
const { TypeValidationError, OperationError } = require('./errors');
const { normalizeNodeName } = require('./utils');
const debug = require('debug')('rclnodejs:parameter_watcher');

/**
 * @class ParameterWatcher - Watches parameter changes on a remote node
 *
 * Subscribes to /parameter_events and emits 'change' events when
 * watched parameters on the target node are modified.
 *
 * @extends EventEmitter
 */
class ParameterWatcher extends EventEmitter {
  #node;
  #paramClient;
  #subscription;
  #watchedParams;
  #remoteNodeName;
  #destroyed;

  /**
   * Create a ParameterWatcher instance.
   * Note: Use node.createParameterWatcher() instead of calling this directly.
   *
   * @param {object} node - The local rclnodejs Node instance
   * @param {string} remoteNodeName - Name of the remote node to watch
   * @param {string[]} parameterNames - Array of parameter names to watch
   * @param {object} [options] - Options for the parameter client
   * @param {number} [options.timeout=5000] - Default timeout for parameter operations
   * @hideconstructor
   */
  constructor(node, remoteNodeName, parameterNames, options = {}) {
    super();

    if (!node || typeof node.createParameterClient !== 'function') {
      throw new TypeValidationError('node', node, 'Node instance', {
        entityType: 'parameter watcher',
      });
    }

    if (typeof remoteNodeName !== 'string' || remoteNodeName.trim() === '') {
      throw new TypeValidationError(
        'remoteNodeName',
        remoteNodeName,
        'non-empty string',
        {
          entityType: 'parameter watcher',
        }
      );
    }

    if (!Array.isArray(parameterNames) || parameterNames.length === 0) {
      throw new TypeValidationError(
        'parameterNames',
        parameterNames,
        'non-empty array',
        {
          entityType: 'parameter watcher',
        }
      );
    }

    this.#node = node;
    this.#watchedParams = new Set(parameterNames);
    this.#paramClient = node.createParameterClient(remoteNodeName, options);
    // Cache the remote node name for error messages (in case paramClient is destroyed)
    this.#remoteNodeName = this.#paramClient.remoteNodeName;
    this.#subscription = null;
    this.#destroyed = false;

    debug(
      'Created ParameterWatcher for node=%s, params=%o',
      remoteNodeName,
      parameterNames
    );
  }

  /**
   * Get the remote node name being watched.
   * @type {string}
   * @readonly
   */
  get remoteNodeName() {
    return this.#remoteNodeName;
  }

  /**
   * Get the list of watched parameter names.
   * @type {string[]}
   * @readonly
   */
  get watchedParameters() {
    return Array.from(this.#watchedParams);
  }

  /**
   * Start watching for parameter changes.
   * Waits for the remote node's parameter services and subscribes to parameter events.
   *
   * @param {number} [timeout=5000] - Timeout in milliseconds to wait for services
   * @returns {Promise<boolean>} Resolves to true when watching has started
   * @throws {Error} If the watcher has been destroyed
   */
  async start(timeout = 5000) {
    this.#checkNotDestroyed();

    debug('Starting ParameterWatcher for node=%s', this.remoteNodeName);

    const available = await this.#paramClient.waitForService(timeout);

    if (!available) {
      debug(
        'Parameter services not available for node=%s',
        this.remoteNodeName
      );
      return false;
    }

    if (!this.#subscription) {
      this.#subscription = this.#node.createSubscription(
        'rcl_interfaces/msg/ParameterEvent',
        '/parameter_events',
        (event) => this.#handleParameterEvent(event)
      );

      debug('Subscribed to /parameter_events');
    }

    return true;
  }

  /**
   * Get current values of all watched parameters.
   *
   * @param {object} [options] - Options for the parameter client
   * @param {number} [options.timeout] - Timeout in milliseconds
   * @param {AbortSignal} [options.signal] - AbortSignal for cancellation
   * @returns {Promise<Parameter[]>} Array of Parameter objects
   * @throws {Error} If the watcher has been destroyed
   */
  async getCurrentValues(options) {
    this.#checkNotDestroyed();
    return await this.#paramClient.getParameters(
      Array.from(this.#watchedParams),
      options
    );
  }

  /**
   * Add a parameter name to the watch list.
   *
   * @param {string} name - Parameter name to watch
   * @throws {TypeError} If name is not a string
   * @throws {Error} If the watcher has been destroyed
   */
  addParameter(name) {
    this.#checkNotDestroyed();

    if (typeof name !== 'string' || name.trim() === '') {
      throw new TypeValidationError('name', name, 'non-empty string', {
        entityType: 'parameter watcher',
        entityName: this.remoteNodeName,
      });
    }

    const wasAdded = !this.#watchedParams.has(name);
    this.#watchedParams.add(name);

    if (wasAdded) {
      debug('Added parameter to watch list: %s', name);
    }
  }

  /**
   * Remove a parameter name from the watch list.
   *
   * @param {string} name - Parameter name to stop watching
   * @returns {boolean} True if the parameter was in the watch list
   * @throws {Error} If the watcher has been destroyed
   */
  removeParameter(name) {
    this.#checkNotDestroyed();

    const wasRemoved = this.#watchedParams.delete(name);

    if (wasRemoved) {
      debug('Removed parameter from watch list: %s', name);
    }

    return wasRemoved;
  }

  /**
   * Check if the watcher has been destroyed.
   *
   * @returns {boolean} True if destroyed
   */
  isDestroyed() {
    return this.#destroyed;
  }

  /**
   * Destroy the watcher and clean up resources.
   * Unsubscribes from parameter events and destroys the parameter client.
   */
  destroy() {
    if (this.#destroyed) {
      return;
    }

    debug('Destroying ParameterWatcher for node=%s', this.remoteNodeName);

    if (this.#subscription) {
      try {
        this.#node.destroySubscription(this.#subscription);
      } catch (error) {
        debug('Error destroying subscription: %s', error.message);
      }
      this.#subscription = null;
    }

    if (this.#paramClient) {
      try {
        this.#node.destroyParameterClient(this.#paramClient);
      } catch (error) {
        debug('Error destroying parameter client: %s', error.message);
      }
      this.#paramClient = null;
    }

    this.removeAllListeners();

    this.#destroyed = true;
  }

  /**
   * Handle parameter event from /parameter_events topic.
   * @private
   */
  #handleParameterEvent(event) {
    if (normalizeNodeName(event.node) !== this.remoteNodeName) {
      return;
    }

    const relevantChanges = [];

    if (event.new_parameters) {
      const newParams = event.new_parameters.filter((p) =>
        this.#watchedParams.has(p.name)
      );
      relevantChanges.push(...newParams);
    }

    if (event.changed_parameters) {
      const changedParams = event.changed_parameters.filter((p) =>
        this.#watchedParams.has(p.name)
      );
      relevantChanges.push(...changedParams);
    }

    if (event.deleted_parameters) {
      const deletedParams = event.deleted_parameters.filter((p) =>
        this.#watchedParams.has(p.name)
      );
      relevantChanges.push(...deletedParams);
    }

    if (relevantChanges.length > 0) {
      debug(
        'Parameter change detected: %o',
        relevantChanges.map((p) => p.name)
      );
      this.emit('change', relevantChanges);
    }
  }

  /**
   * Check if the watcher has been destroyed and throw if so.
   * @private
   */
  #checkNotDestroyed() {
    if (this.#destroyed) {
      throw new OperationError('ParameterWatcher has been destroyed', {
        code: 'WATCHER_DESTROYED',
        entityType: 'parameter watcher',
        entityName: this.remoteNodeName,
      });
    }
  }
}

module.exports = ParameterWatcher;

1	// Copyright (c) 2025 Mahmoud Alghalayini. All rights reserved.
2	//
3	// Licensed under the Apache License, Version 2.0 (the "License");
4	// you may not use this file except in compliance with the License.
5	// You may obtain a copy of the License at
6	//
7	// http://www.apache.org/licenses/LICENSE-2.0
8	//
9	// Unless required by applicable law or agreed to in writing, software
10	// distributed under the License is distributed on an "AS IS" BASIS,
11	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	// See the License for the specific language governing permissions and
13	// limitations under the License.
14
15	'use strict';
16
17	const EventEmitter = require('events');	26✔
18	const { TypeValidationError, OperationError } = require('./errors');	26✔
19	const { normalizeNodeName } = require('./utils');	26✔
20	const debug = require('debug')('rclnodejs:parameter_watcher');	26✔
21
22	/**
23	* @class ParameterWatcher - Watches parameter changes on a remote node
24	*
25	* Subscribes to /parameter_events and emits 'change' events when
26	* watched parameters on the target node are modified.
27	*
28	* @extends EventEmitter
29	*/
30	class ParameterWatcher extends EventEmitter {
31	#node;
32	#paramClient;
33	#subscription;
34	#watchedParams;
35	#remoteNodeName;
36	#destroyed;
37
38	/**
39	* Create a ParameterWatcher instance.
40	* Note: Use node.createParameterWatcher() instead of calling this directly.
41	*
42	* @param {object} node - The local rclnodejs Node instance
43	* @param {string} remoteNodeName - Name of the remote node to watch
44	* @param {string[]} parameterNames - Array of parameter names to watch
45	* @param {object} [options] - Options for the parameter client
46	* @param {number} [options.timeout=5000] - Default timeout for parameter operations
47	* @hideconstructor
48	*/
49	constructor(node, remoteNodeName, parameterNames, options = {}) {	×
50	super();	57✔
51
52	if (!node \|\| typeof node.createParameterClient !== 'function') {	57!
NEW 53	throw new TypeValidationError('node', node, 'Node instance', {	×
54	entityType: 'parameter watcher',
55	});
56	}
57
58	if (typeof remoteNodeName !== 'string' \|\| remoteNodeName.trim() === '') {	57✔
59	throw new TypeValidationError(	2✔
60	'remoteNodeName',
61	remoteNodeName,
62	'non-empty string',
63	{
64	entityType: 'parameter watcher',
65	}
66	);
67	}
68
69	if (!Array.isArray(parameterNames) \|\| parameterNames.length === 0) {	55✔
70	throw new TypeValidationError(	2✔
71	'parameterNames',
72	parameterNames,
73	'non-empty array',
74	{
75	entityType: 'parameter watcher',
76	}
77	);
78	}
79
80	this.#node = node;	53✔
81	this.#watchedParams = new Set(parameterNames);	53✔
82	this.#paramClient = node.createParameterClient(remoteNodeName, options);	53✔
83	// Cache the remote node name for error messages (in case paramClient is destroyed)
84	this.#remoteNodeName = this.#paramClient.remoteNodeName;	53✔
85	this.#subscription = null;	53✔
86	this.#destroyed = false;	53✔
87
88	debug(	53✔
89	'Created ParameterWatcher for node=%s, params=%o',
90	remoteNodeName,
91	parameterNames
92	);
93	}
94
95	/**
96	* Get the remote node name being watched.
97	* @type {string}
98	* @readonly
99	*/
100	get remoteNodeName() {
101	return this.#remoteNodeName;	128✔
102	}
103
104	/**
105	* Get the list of watched parameter names.
106	* @type {string[]}
107	* @readonly
108	*/
109	get watchedParameters() {
110	return Array.from(this.#watchedParams);	14✔
111	}
112
113	/**
114	* Start watching for parameter changes.
115	* Waits for the remote node's parameter services and subscribes to parameter events.
116	*
117	* @param {number} [timeout=5000] - Timeout in milliseconds to wait for services
118	* @returns {Promise<boolean>} Resolves to true when watching has started
119	* @throws {Error} If the watcher has been destroyed
120	*/
121	async start(timeout = 5000) {	1✔
122	this.#checkNotDestroyed();	46✔
123
124	debug('Starting ParameterWatcher for node=%s', this.remoteNodeName);	45✔
125
126	const available = await this.#paramClient.waitForService(timeout);	45✔
127
128	if (!available) {	45✔
129	debug(	4✔
130	'Parameter services not available for node=%s',
131	this.remoteNodeName
132	);
133	return false;	4✔
134	}
135
136	if (!this.#subscription) {	41!
137	this.#subscription = this.#node.createSubscription(	41✔
138	'rcl_interfaces/msg/ParameterEvent',
139	'/parameter_events',
140	(event) => this.#handleParameterEvent(event)	15✔
141	);
142
143	debug('Subscribed to /parameter_events');	41✔
144	}
145
146	return true;	41✔
147	}
148
149	/**
150	* Get current values of all watched parameters.
151	*
152	* @param {object} [options] - Options for the parameter client
153	* @param {number} [options.timeout] - Timeout in milliseconds
154	* @param {AbortSignal} [options.signal] - AbortSignal for cancellation
155	* @returns {Promise<Parameter[]>} Array of Parameter objects
156	* @throws {Error} If the watcher has been destroyed
157	*/
158	async getCurrentValues(options) {
159	this.#checkNotDestroyed();	6✔
160	return await this.#paramClient.getParameters(	5✔
161	Array.from(this.#watchedParams),
162	options
163	);
164	}
165
166	/**
167	* Add a parameter name to the watch list.
168	*
169	* @param {string} name - Parameter name to watch
170	* @throws {TypeError} If name is not a string
171	* @throws {Error} If the watcher has been destroyed
172	*/
173	addParameter(name) {
174	this.#checkNotDestroyed();	9✔
175
176	if (typeof name !== 'string' \|\| name.trim() === '') {	7✔
177	throw new TypeValidationError('name', name, 'non-empty string', {	2✔
178	entityType: 'parameter watcher',
179	entityName: this.remoteNodeName,
180	});
181	}
182
183	const wasAdded = !this.#watchedParams.has(name);	5✔
184	this.#watchedParams.add(name);	5✔
185
186	if (wasAdded) {	5✔
187	debug('Added parameter to watch list: %s', name);	3✔
188	}
189	}
190
191	/**
192	* Remove a parameter name from the watch list.
193	*
194	* @param {string} name - Parameter name to stop watching
195	* @returns {boolean} True if the parameter was in the watch list
196	* @throws {Error} If the watcher has been destroyed
197	*/
198	removeParameter(name) {
199	this.#checkNotDestroyed();	6✔
200
201	const wasRemoved = this.#watchedParams.delete(name);	4✔
202
203	if (wasRemoved) {	4✔
204	debug('Removed parameter from watch list: %s', name);	3✔
205	}
206
207	return wasRemoved;	4✔
208	}
209
210	/**
211	* Check if the watcher has been destroyed.
212	*
213	* @returns {boolean} True if destroyed
214	*/
215	isDestroyed() {
216	return this.#destroyed;	46✔
217	}
218
219	/**
220	* Destroy the watcher and clean up resources.
221	* Unsubscribes from parameter events and destroys the parameter client.
222	*/
223	destroy() {
224	if (this.#destroyed) {	105✔
225	return;	52✔
226	}
227
228	debug('Destroying ParameterWatcher for node=%s', this.remoteNodeName);	53✔
229
230	if (this.#subscription) {	53✔
231	try {	41✔
232	this.#node.destroySubscription(this.#subscription);	41✔
233	} catch (error) {
NEW 234	debug('Error destroying subscription: %s', error.message);	×
235	}
236	this.#subscription = null;	41✔
237	}
238
239	if (this.#paramClient) {	53!
240	try {	53✔
241	this.#node.destroyParameterClient(this.#paramClient);	53✔
242	} catch (error) {
NEW 243	debug('Error destroying parameter client: %s', error.message);	×
244	}
245	this.#paramClient = null;	53✔
246	}
247
248	this.removeAllListeners();	53✔
249
250	this.#destroyed = true;	53✔
251	}
252
253	/**
254	* Handle parameter event from /parameter_events topic.
255	* @private
256	*/
257	#handleParameterEvent(event) {
258	if (normalizeNodeName(event.node) !== this.remoteNodeName) {	15✔
259	return;	8✔
260	}
261
262	const relevantChanges = [];	7✔
263
264	if (event.new_parameters) {	7!
265	const newParams = event.new_parameters.filter((p) =>	7✔
NEW 266	this.#watchedParams.has(p.name)	×
267	);
268	relevantChanges.push(...newParams);	7✔
269	}
270
271	if (event.changed_parameters) {	7!
272	const changedParams = event.changed_parameters.filter((p) =>	7✔
273	this.#watchedParams.has(p.name)	7✔
274	);
275	relevantChanges.push(...changedParams);	7✔
276	}
277
278	if (event.deleted_parameters) {	7!
279	const deletedParams = event.deleted_parameters.filter((p) =>	7✔
NEW 280	this.#watchedParams.has(p.name)	×
281	);
282	relevantChanges.push(...deletedParams);	7✔
283	}
284
285	if (relevantChanges.length > 0) {	7✔
286	debug(	5✔
287	'Parameter change detected: %o',
288	relevantChanges.map((p) => p.name)	5✔
289	);
290	this.emit('change', relevantChanges);	5✔
291	}
292	}
293
294	/**
295	* Check if the watcher has been destroyed and throw if so.
296	* @private
297	*/
298	#checkNotDestroyed() {
299	if (this.#destroyed) {	67✔
300	throw new OperationError('ParameterWatcher has been destroyed', {	6✔
301	code: 'WATCHER_DESTROYED',
302	entityType: 'parameter watcher',
303	entityName: this.remoteNodeName,
304	});
305	}
306	}
307	}
308
309	module.exports = ParameterWatcher;	26✔

RobotWebTools / rclnodejs / 19565048871

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous