Support waitForAllAcked() for publisher (#1138)

This pull request adds support for the waitForAllAcked() method in the publisher API to allow clients to block until all published messages are acknowledged (or until a timeout occurs).  
- Adds type definitions for waitForAllAcked() in the TypeScript tests.  
- Implements and exposes waitForAllAcked() in the C++ bindings and in the JavaScript publisher class.  
- Introduces a new test case validating the waitForAllAcked() behavior.

Fix:  #1137

Author: minggangw

lib/publisher.js

@@ -80,6 +80,25 @@ class Publisher extends Entity {
   get subscriptionCount() {
     return rclnodejs.getSubscriptionCount(this._handle);
   }
+
+  /**
+   * Wait until all published message data is acknowledged or until the specified timeout elapses
+   *
+   * If the timeout is negative then this function will block indefinitely until all published
+   * message data is acknowledged.
+   * If the timeout is 0 then it will check if all published message has been acknowledged without
+   * waiting.
+   * If the timeout is greater than 0 then it will return after that period of time has elapsed or
+   * all published message data is acknowledged.
+   *
+   * Raises an error if failed, such as the middleware not supporting this feature.
+   *
+   * @param {timeout} timeout - The duration to wait for all published message data to be acknowledged in nanoseconds.
+   * @return {boolean} `true` if all published message data is acknowledged before the timeout, otherwise `false`.
+   */
+  waitForAllAcked(timeout) {
+    return rclnodejs.waitForAllAcked(this._handle, timeout);
+  }
 }
 
 module.exports = Publisher;

src/rcl_publisher_bindings.cpp

@@ -128,13 +128,32 @@ Napi::Value GetSubscriptionCount(const Napi::CallbackInfo& info) {
   return Napi::Number::New(env, count);
 }
 
+Napi::Value WaitForAllAcked(const Napi::CallbackInfo& info) {
+  Napi::Env env = info.Env();
+  rcl_publisher_t* publisher = reinterpret_cast<rcl_publisher_t*>(
+      RclHandle::Unwrap(info[0].As<Napi::Object>())->ptr());
+  bool lossless;
+  int64_t nanoseconds = info[1].As<Napi::BigInt>().Int64Value(&lossless);
+
+  rcl_ret_t ret = rcl_publisher_wait_for_all_acked(publisher, nanoseconds);
+  if (RCL_RET_OK == ret) {
+    return Napi::Boolean::New(env, true);
+  } else if (RCL_RET_TIMEOUT == ret) {
+    return Napi::Boolean::New(env, false);
+  }
+  Napi::Error::New(env, "Failed to wait for all acknowledgements")
+      .ThrowAsJavaScriptException();
+  return env.Undefined();
+}
+
 Napi::Object InitPublisherBindings(Napi::Env env, Napi::Object exports) {
   exports.Set("createPublisher", Napi::Function::New(env, CreatePublisher));
   exports.Set("publish", Napi::Function::New(env, Publish));
   exports.Set("getPublisherTopic", Napi::Function::New(env, GetPublisherTopic));
   exports.Set("publishRawMessage", Napi::Function::New(env, PublishRawMessage));
   exports.Set("getSubscriptionCount",
               Napi::Function::New(env, GetSubscriptionCount));
+  exports.Set("waitForAllAcked", Napi::Function::New(env, WaitForAllAcked));
   return exports;
 }
 

test/test-publisher.js

@@ -51,4 +51,15 @@ describe('rclnodejs publisher test suite', function () {
     node.createSubscription(String, 'topic', (msg) => {});
     assert.strictEqual(publisher.subscriptionCount, 1);
   });
+
+  it('Wait for all acked', function () {
+    const node = rclnodejs.createNode('publisher_node');
+    const String = 'std_msgs/msg/String';
+    const publisher = node.createPublisher(String, 'topic');
+    node.createSubscription(String, 'topic', (msg) => {});
+    assert.strictEqual(publisher.subscriptionCount, 1);
+
+    publisher.publish('Hello World');
+    assert.strictEqual(publisher.waitForAllAcked(BigInt(1000000000)), true);
+  });
 });

test/types/index.test-d.ts

@@ -134,6 +134,7 @@ expectType<void>(publisher.publish(MSG));
 expectType<void>(publisher.publish(Buffer.from('Hello ROS World')));
 expectType<void>(node.destroyPublisher(publisher));
 expectType<boolean>(publisher.isDestroyed());
+expectType<boolean>(publisher.waitForAllAcked(BigInt(1000)));
 
 // ---- LifecyclePublisher ----
 const lifecyclePublisher = lifecycleNode.createLifecyclePublisher(

types/publisher.d.ts

@@ -20,5 +20,22 @@ declare module 'rclnodejs' {
      * @returns The number of subscriptions
      */
     subscriptionCount(): number;
+
+    /**
+     * Wait until all published message data is acknowledged or until the specified timeout elapses
+     *
+     * If the timeout is negative then this function will block indefinitely until all published
+     * message data is acknowledged.
+     * If the timeout is 0 then it will check if all published message has been acknowledged without
+     * waiting.
+     * If the timeout is greater than 0 then it will return after that period of time has elapsed or
+     * all published message data is acknowledged.
+     *
+     * Raises an error if failed, such as the middleware not supporting this feature.
+     *
+     * @param {timeout} timeout - The duration to wait for all published message data to be acknowledged in nanoseconds.
+     * @return {boolean} `true` if all published message data is acknowledged before the timeout, otherwise `false`.
+     */
+    waitForAllAcked(timeout: bigint): boolean;
   }
 }