address comments

Author: mahmoud-ghalayini

example/topics/subscriber/subscription-observable-example.js

@@ -39,6 +39,9 @@ async function main() {
     });
 
   // Example 2: Filtered messages (only containing "ROS")
+  // Note: RxJS filter() operates at the application level after messages are received.
+  // For more efficient filtering at the DDS middleware level (reducing network traffic),
+  // use the contentFilter option. See: tutorials/content-filtering-subscription.md
   obsSub.observable
     .pipe(
       map((msg) => msg.data),

lib/node.js

@@ -839,44 +839,9 @@ class Node extends rclnodejs.ShadowNode {
    * @return {ObservableSubscription} - An ObservableSubscription with an RxJS Observable.
    */
   createObservableSubscription(typeClass, topic, options, eventCallbacks) {
-    if (typeof typeClass === 'string' || typeof typeClass === 'object') {
-      typeClass = loader.loadInterface(typeClass);
-    }
-
-    options = this._validateOptions(options);
-
-    if (typeof typeClass !== 'function') {
-      throw new TypeValidationError('typeClass', typeClass, 'function', {
-        nodeName: this.name(),
-        entityType: 'subscription',
-      });
-    }
-    if (typeof topic !== 'string') {
-      throw new TypeValidationError('topic', topic, 'string', {
-        nodeName: this.name(),
-        entityType: 'subscription',
-      });
-    }
-    if (
-      eventCallbacks &&
-      !(eventCallbacks instanceof SubscriptionEventCallbacks)
-    ) {
-      throw new TypeValidationError(
-        'eventCallbacks',
-        eventCallbacks,
-        'SubscriptionEventCallbacks',
-        {
-          nodeName: this.name(),
-          entityType: 'subscription',
-          entityName: topic,
-        }
-      );
-    }
-
     let observableSubscription = null;
 
-    const subscription = Subscription.createSubscription(
-      this,
+    const subscription = this.createSubscription(
       typeClass,
       topic,
       options,
@@ -888,13 +853,7 @@ class Node extends rclnodejs.ShadowNode {
       eventCallbacks
     );
 
-    observableSubscription = new ObservableSubscription(subscription);
-
-    debug('Finish creating observable subscription, topic = %s.', topic);
-    this._subscriptions.push(subscription);
-    this.syncHandles();
-
-    return observableSubscription;
+    return new ObservableSubscription(subscription);
   }
 
   /**

tutorials/observable-subscriptions.md

@@ -25,6 +25,8 @@ rclnodejs provides RxJS Observable support for subscriptions, enabling reactive
 
 Observable subscriptions are ideal for complex message processing pipelines where you need to combine, filter, or transform data from multiple sources.
 
+> **Note:** RxJS `filter()` operates at the application level after messages are received. For simple content-based filtering that reduces network traffic, consider using [DDS Content Filtering](content-filtering-subscription.md) instead, which filters at the middleware level before messages reach your application. See [Example 6](#example-6-combining-dds-content-filtering-with-rxjs) for combined usage.
+
 ## Basic Usage
 
 ```javascript
@@ -179,6 +181,47 @@ tempSub.observable
   });
 ```
 
+### Example 6: Combining DDS Content Filtering with RxJS
+
+For optimal performance, use DDS content filtering to reduce network traffic at the middleware level, then apply RxJS operators for additional processing:
+
+```javascript
+const { throttleTime, map } = require('rxjs');
+
+// DDS filters at middleware level - only temperatures > 30°C are delivered
+const tempSub = node.createObservableSubscription(
+  'sensor_msgs/msg/Temperature',
+  '/temperature',
+  {
+    contentFilter: {
+      expression: 'temperature > %0',
+      parameters: ['30.0'],
+    },
+  }
+);
+
+// RxJS processes the pre-filtered stream
+tempSub.observable
+  .pipe(
+    throttleTime(1000), // Rate limit to 1 msg/sec
+    map((msg) => ({
+      celsius: msg.temperature,
+      fahrenheit: msg.temperature * 1.8 + 32,
+    }))
+  )
+  .subscribe((temp) => {
+    console.log(`High temp alert: ${temp.celsius}°C (${temp.fahrenheit}°F)`);
+  });
+```
+
+This approach provides:
+
+- **Network efficiency**: DDS drops messages below 30°C before transmission
+- **CPU efficiency**: RxJS only processes relevant messages
+- **Flexibility**: RxJS handles rate limiting and transformation
+
+> See [Content Filtering Subscription](content-filtering-subscription.md) for more details on DDS content filtering.
+
 ## Cleanup
 
 Always clean up subscriptions when done: