feat: Add support for QoS profiles

packages/roslib/src/RosLib.ts

@@ -118,4 +118,6 @@ export {
   type SuccessfulRosbridgeActionResultMessage,
   type FailedRosbridgeServiceResponseMessage,
   type SuccessfulRosbridgeServiceResponseMessage,
+  type RosbridgeQosProfile,
+  type RosbridgeDuration,
 } from "./types/protocol.ts";

packages/roslib/src/core/Topic.ts

@@ -10,6 +10,7 @@ import {
   isRosbridgePublishMessage,
   type RosbridgeAdvertiseMessage,
   type RosbridgeMessage,
+  type RosbridgeQosProfile,
   type RosbridgeSubscribeMessage,
 } from "../types/protocol.ts";
 import type { rosapi } from "../types/rosapi.ts";
@@ -39,6 +40,7 @@ export default class Topic<T = unknown> extends EventEmitter<{
   latch: boolean;
   queue_size: number;
   queue_length: number;
+  qos?: RosbridgeQosProfile;
   reconnect_on_close: boolean;
   callForSubscribeAndAdvertise: (
     message: RosbridgeSubscribeMessage | RosbridgeAdvertiseMessage,
@@ -55,6 +57,7 @@ export default class Topic<T = unknown> extends EventEmitter<{
    * @param [options.queue_size=100] - The queue created at bridge side for re-publishing webtopics.
    * @param [options.latch=false] - Latch the topic when publishing.
    * @param [options.queue_length=0] - The queue length at bridge side used when subscribing.
+   * @param [options.qos] - QoS profile for the topic. When provided, `latch` and `queue_size` are ignored.
    * @param [options.reconnect_on_close=true] - The flag to enable resubscription and readvertisement on close event.
    */
   constructor({
@@ -66,6 +69,7 @@ export default class Topic<T = unknown> extends EventEmitter<{
     latch = false,
     queue_size = 100,
     queue_length = 0,
+    qos,
     reconnect_on_close = true,
   }: {
     ros: Ros;
@@ -76,6 +80,7 @@ export default class Topic<T = unknown> extends EventEmitter<{
     queue_size?: number;
     latch?: boolean;
     queue_length?: number;
+    qos?: RosbridgeQosProfile;
     reconnect_on_close?: boolean;
   }) {
     super();
@@ -88,6 +93,7 @@ export default class Topic<T = unknown> extends EventEmitter<{
     this.latch = latch;
     this.queue_size = queue_size;
     this.queue_length = queue_length;
+    this.qos = qos;
     this.reconnect_on_close = reconnect_on_close;
 
     // Check for valid compression types
@@ -171,6 +177,7 @@ export default class Topic<T = unknown> extends EventEmitter<{
       compression: this.compression,
       throttle_rate: this.throttle_rate,
       queue_length: this.queue_length,
+      qos: this.qos,
     });
   }
   /**
@@ -221,6 +228,7 @@ export default class Topic<T = unknown> extends EventEmitter<{
       topic: this.name,
       latch: this.latch,
       queue_size: this.queue_size,
+      qos: this.qos,
     });
     this.isAdvertised = true;
 

packages/roslib/src/types/protocol.ts

@@ -4,6 +4,20 @@
 
 import type { GoalStatus } from "../core/GoalStatus.ts";
 
+export type RosbridgeDuration =
+  | number
+  | { secs: number; nsecs: number }
+  | "infinite";
+
+export interface RosbridgeQosProfile {
+  history?: "keep_last" | "keep_all";
+  depth?: number;
+  reliability?: "reliable" | "best_effort" | "best_available";
+  durability?: "transient_local" | "volatile" | "best_available";
+  deadline?: RosbridgeDuration | "best_available";
+  lifespan?: RosbridgeDuration;
+}
+
 export interface RosbridgeMessageBase {
   op: string;
 }
@@ -87,8 +101,11 @@ export interface RosbridgeAdvertiseMessage extends RosbridgeMessageBase {
   id?: string;
   type: string;
   topic: string;
+  /** @deprecated Use `qos.durability` instead. */
   latch?: boolean;
+  /** @deprecated Use `qos.depth` instead. */
   queue_size?: number;
+  qos?: RosbridgeQosProfile;
 }
 
 export function isRosbridgeAdvertiseMessage(
@@ -115,6 +132,12 @@ export interface RosbridgePublishMessage<TMessage = unknown>
   id?: string;
   topic: string;
   msg: TMessage;
+  type?: string;
+  /** @deprecated Use `qos.durability` instead. */
+  latch?: boolean;
+  /** @deprecated Use `qos.depth` instead. */
+  queue_size?: number;
+  qos?: RosbridgeQosProfile;
 }
 
 export function isRosbridgePublishMessage<T>(
@@ -132,6 +155,7 @@ export interface RosbridgeSubscribeMessage extends RosbridgeMessageBase {
   queue_length?: number;
   fragment_size?: number;
   compression?: string;
+  qos?: RosbridgeQosProfile;
 }
 
 export function isRosbridgeSubscribeMessage(

packages/roslib/test/examples/pubsub.example.ts

@@ -1,10 +1,8 @@
-import { describe, it, expect, afterAll, vi } from "vitest";
+import { describe, it, expect, afterAll, vi, beforeAll } from "vitest";
 import * as ROSLIB from "../../src/RosLib.ts";
 
 describe("Topics Example", function () {
-  const ros = new ROSLIB.Ros({
-    url: "ws://localhost:9090",
-  });
+  const ros = new ROSLIB.Ros();
 
   const example = ros.Topic<{ data: string }>({
     name: "/example_topic",
@@ -22,6 +20,10 @@ describe("Topics Example", function () {
     messageType: "std_msgs/String",
   });
 
+  beforeAll(async () => {
+    await ros.connect("ws://localhost:9090");
+  });
+
   it("Client-side subscribers receive messages from client-side publishers", async () => {
     // Create copies of the message arrays so we can shift from them
     const messages1Copy = [...messages1];
@@ -128,3 +130,110 @@ describe("Topics Example", function () {
     example2.unsubscribe();
   });
 }, 1000);
+
+// QoS profiles are a ROS 2 rosbridge feature (protocol 2.1.0).
+// Each case uses its own topic, because rosbridge keeps one publisher and one
+// subscription per topic and only the first advertise/subscribe sets its QoS.
+describe.skipIf(process.env["ROS_DISTRO"] === "noetic")(
+  "Topics QoS Example",
+  function () {
+    const ros = new ROSLIB.Ros();
+    const topics: ROSLIB.Topic<{ data: string }>[] = [];
+
+    function makeTopic(name: string, qos: ROSLIB.RosbridgeQosProfile) {
+      const topic = ros.Topic<{ data: string }>({
+        name,
+        messageType: "std_msgs/String",
+        qos,
+      });
+      topics.push(topic);
+      return topic;
+    }
+
+    // Gives rosbridge time to create the ROS publisher/subscription and for
+    // DDS discovery to match them, since advertise/subscribe have no reply
+    const settle = () => new Promise((resolve) => setTimeout(resolve, 1000));
+
+    beforeAll(async () => {
+      await ros.connect("ws://localhost:9090");
+    });
+
+    it("Late subscriber receives a message published with transient local durability", async () => {
+      const name = "/qos_transient_local";
+      const publisher = makeTopic(name, {
+        reliability: "reliable",
+        durability: "transient_local",
+        depth: 1,
+      });
+      publisher.advertise();
+      await settle();
+      publisher.publish({ data: "latched" });
+      await settle();
+
+      const callback = vi.fn();
+      makeTopic(name, {
+        reliability: "reliable",
+        durability: "transient_local",
+      }).subscribe(callback);
+
+      await vi.waitFor(
+        () => {
+          expect(callback).toHaveBeenCalledWith({ data: "latched" });
+        },
+        { timeout: 5000 },
+      );
+    }, 10000);
+
+    it("Late subscriber gets nothing from a volatile publisher", async () => {
+      // rosbridge's default publisher is transient local, so this only passes
+      // when the qos profile actually reaches the server
+      const name = "/qos_volatile";
+      const publisher = makeTopic(name, {
+        reliability: "reliable",
+        durability: "volatile",
+      });
+      publisher.advertise();
+      await settle();
+      publisher.publish({ data: "not latched" });
+      await settle();
+
+      const callback = vi.fn();
+      makeTopic(name, {
+        reliability: "reliable",
+        durability: "transient_local",
+      }).subscribe(callback);
+      await settle();
+
+      expect(callback).not.toHaveBeenCalled();
+    }, 10000);
+
+    it("Reliable subscriber gets nothing from a best effort publisher", async () => {
+      // Incompatible QoS: a reliable reader never matches a best effort writer
+      const name = "/qos_incompatible";
+      const publisher = makeTopic(name, {
+        reliability: "best_effort",
+        durability: "volatile",
+      });
+      const callback = vi.fn();
+      makeTopic(name, {
+        reliability: "reliable",
+        durability: "volatile",
+      }).subscribe(callback);
+      publisher.advertise();
+      await settle();
+
+      publisher.publish({ data: "dropped" });
+      await settle();
+
+      expect(callback).not.toHaveBeenCalled();
+    }, 10000);
+
+    afterAll(function () {
+      for (const topic of topics) {
+        topic.unsubscribe();
+        topic.unadvertise();
+      }
+      ros.close();
+    });
+  },
+);

packages/roslib/test/topic.test.ts

@@ -0,0 +1,127 @@
+import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
+import { EventEmitter } from "eventemitter3";
+import Topic from "../src/core/Topic.ts";
+import type Ros from "../src/core/Ros.ts";
+import type {
+  RosbridgeMessage,
+  RosbridgeQosProfile,
+} from "../src/types/protocol.ts";
+
+const qos: RosbridgeQosProfile = {
+  history: "keep_last",
+  depth: 5,
+  reliability: "best_effort",
+  durability: "transient_local",
+  deadline: { secs: 1, nsecs: 500000000 },
+  lifespan: "infinite",
+};
+
+describe("Topic QoS", () => {
+  let mockRos: Ros & EventEmitter;
+  let callOnConnection: ReturnType<
+    typeof vi.fn<(msg: RosbridgeMessage) => void>
+  >;
+
+  /** All messages sent to the server with the given op, as they go over the wire */
+  function sent(op: string): Record<string, unknown>[] {
+    return callOnConnection.mock.calls
+      .map(
+        ([msg]) => JSON.parse(JSON.stringify(msg)) as Record<string, unknown>,
+      )
+      .filter((msg) => msg["op"] === op);
+  }
+
+  beforeEach(() => {
+    callOnConnection = vi.fn<(msg: RosbridgeMessage) => void>();
+    mockRos = Object.assign(new EventEmitter(), {
+      callOnConnection,
+    }) as unknown as Ros & EventEmitter;
+  });
+
+  afterEach(() => {
+    vi.clearAllMocks();
+  });
+
+  function makeTopic(options: { qos?: RosbridgeQosProfile } = {}) {
+    return new Topic({
+      ros: mockRos,
+      name: "/chatter",
+      messageType: "std_msgs/msg/String",
+      ...options,
+    });
+  }
+
+  it("sends the qos profile when advertising", () => {
+    makeTopic({ qos }).advertise();
+
+    const [advertise] = sent("advertise");
+    expect(advertise).toMatchObject({
+      topic: "/chatter",
+      type: "std_msgs/msg/String",
+      qos,
+    });
+  });
+
+  it("sends the qos profile when subscribing", () => {
+    makeTopic({ qos }).subscribe(() => undefined);
+
+    const [subscribe] = sent("subscribe");
+    expect(subscribe).toMatchObject({
+      topic: "/chatter",
+      type: "std_msgs/msg/String",
+      qos,
+    });
+  });
+
+  it("advertises with qos before the first publish", () => {
+    const topic = makeTopic({ qos });
+    topic.publish({ data: "hello" });
+    topic.publish({ data: "again" });
+
+    expect(sent("advertise")).toHaveLength(1);
+    expect(sent("advertise")[0]).toMatchObject({ qos });
+    expect(sent("publish")).toHaveLength(2);
+    // QoS belongs to the advertisement; the publish messages don't repeat it
+    for (const publish of sent("publish")) {
+      expect(publish).not.toHaveProperty("qos");
+    }
+  });
+
+  it("passes an empty profile through to request full system-default QoS", () => {
+    makeTopic({ qos: {} }).advertise();
+
+    expect(sent("advertise")[0]).toHaveProperty("qos", {});
+  });
+
+  it("omits the qos field when no profile is given", () => {
+    const topic = makeTopic();
+    topic.advertise();
+    topic.subscribe(() => undefined);
+
+    expect(sent("advertise")[0]).not.toHaveProperty("qos");
+    expect(sent("subscribe")[0]).not.toHaveProperty("qos");
+    // Old servers still get the legacy fields
+    expect(sent("advertise")[0]).toMatchObject({
+      latch: false,
+      queue_size: 100,
+    });
+  });
+
+  it("re-sends the qos profile when re-advertising after reconnect", () => {
+    makeTopic({ qos }).advertise();
+
+    mockRos.emit("close");
+
+    expect(sent("advertise")).toHaveLength(2);
+    expect(sent("advertise")[1]).toMatchObject({ qos });
+  });
+
+  it("re-sends the qos profile when re-subscribing after reconnect", () => {
+    makeTopic({ qos }).subscribe(() => undefined);
+
+    mockRos.emit("close");
+
+    expect(sent("subscribe")).toHaveLength(2);
+    expect(sent("subscribe")[1]).toMatchObject({ qos });
+  });
+});