Six foxglove_bridge-parity features: whitelist, QoS depths, topic push, backpressure, latched replay, TLS

CHANGELOG.rst

@@ -2,6 +2,27 @@
 Changelog for package pj_ros_bridge
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
+0.7.0 (2026-07-06)
+------------------
+* Foxglove-parity feature set, closing the gap with foxglove_bridge's
+  topic-subscription features:
+  - Topic whitelist: full-match regex filtering of visible/subscribable
+    topics (``topic_whitelist`` / ``--topic-whitelist``)
+  - QoS depth heuristics (ROS2): KEEP_LAST subscription depth derived from
+    discovered publisher depths, clamped to ``min_qos_depth``/``max_qos_depth``
+  - Opt-in pushed topic advertisement: ``subscribe_topic_updates`` /
+    ``unsubscribe_topic_updates`` commands and a ``topics_changed``
+    notification, polled at ``topic_poll_interval``
+  - Slow-client backpressure: bounded, drop-oldest per-client send queue
+    (``client_backlog_size``) instead of blocking the publish loop or
+    disconnecting the client
+  - Latched topic replay (ROS2): new subscribers to a ``TRANSIENT_LOCAL``
+    topic (e.g. ``/tf_static``) immediately receive the retained last message
+  - TLS (``wss://``): optional server-certificate TLS via OpenSSL
+    (``tls``/``certfile``/``keyfile``, CMake option ``PJ_BRIDGE_TLS``)
+* All changes are additive; ``protocol_version`` remains ``1``
+* 231 unit tests passing (TSAN/ASAN clean)
+
 0.1.0 (2026-02-11)
 ------------------
 * **License changed to AGPL-3.0** (was Apache-2.0 in development)

CLAUDE.md

@@ -197,7 +197,7 @@ Then, for each message in the (compressed) payload:
 
 ## Testing
 
-### Test Count: 176 unit tests across 11 test suites
+### Test Count: 231 unit tests across 11 test suites
 
 ### Commands
 ```bash
@@ -229,22 +229,38 @@ pre-commit run -a
 
 ### ROS2 (via `--ros-args -p`):
 ```yaml
-port: 9090                  # WebSocket port
-publish_rate: 50.0          # Hz
-session_timeout: 10.0       # seconds
-strip_large_messages: false # Opt-in: strip Image/PointCloud2/etc data fields
+port: 9090                     # WebSocket port
+publish_rate: 50.0             # Hz
+session_timeout: 10.0          # seconds
+strip_large_messages: false    # Opt-in: strip Image/PointCloud2/etc data fields
+topic_whitelist: [".*"]        # Full-match regex patterns restricting visible/subscribable topics
+min_qos_depth: 1               # Minimum KEEP_LAST subscription depth after aggregating publisher depths
+max_qos_depth: 100             # Maximum KEEP_LAST subscription depth after aggregating publisher depths
+topic_poll_interval: 1.0       # Seconds between topics_changed notification polls; 0 disables polling
+client_backlog_size: 100       # Max frames queued per slow client before dropping the oldest (must be > 0)
+tls: false                     # Enable TLS (wss://); requires certfile and keyfile
+certfile: ""                   # TLS server certificate file
+keyfile: ""                    # TLS private key file
 ```
 
 ### RTI (via CLI flags):
 ```bash
-pj_bridge_rti --domains 0 1 --port 9090 --publish-rate 50 --session-timeout 10
+pj_bridge_rti --domains 0 1 --port 9090 --publish-rate 50 --session-timeout 10 \
+  --topic-whitelist ".*" --topic-poll-interval 1.0 --client-backlog-size 100 \
+  --certfile cert.pem --keyfile key.pem
 ```
 
 ### FastDDS (via CLI flags):
 ```bash
-pj_bridge_fastdds --domains 0 1 --port 9090 --publish-rate 50 --session-timeout 10
+pj_bridge_fastdds --domains 0 1 --port 9090 --publish-rate 50 --session-timeout 10 \
+  --topic-whitelist ".*" --topic-poll-interval 1.0 --client-backlog-size 100 \
+  --certfile cert.pem --keyfile key.pem
 ```
 
+See `docs/API.md` for full semantics of each option (topic whitelist matching rules,
+QoS depth heuristics, `topics_changed` notification, backpressure policy, latched
+replay, and TLS setup).
+
 ## Important Design Decisions
 
 1. **Backend-agnostic core via interfaces**: `TopicSourceInterface` and `SubscriptionManagerInterface` allow the same `BridgeServer` to work with ROS2 or RTI DDS.
@@ -261,7 +277,7 @@ pj_bridge_fastdds --domains 0 1 --port 9090 --publish-rate 50 --session-timeout
 
 ---
 
-**Last Updated**: 2026-07-02
+**Last Updated**: 2026-07-06
 **Project Phase**: Unified multi-backend architecture
-**Test Status**: 176 unit tests passing (all sanitizers clean)
+**Test Status**: 231 unit tests passing (all sanitizers clean)
 **Executables**: `pj_bridge_ros2` (ROS2), `pj_bridge_rti` (RTI DDS, disabled), `pj_bridge_fastdds` (FastDDS)

CMakeLists.txt

@@ -53,6 +53,8 @@ find_package(ZSTD REQUIRED)
 include(FetchContent)
 
 # IXWebSocket — try system/conda first, fall back to FetchContent
+option(PJ_BRIDGE_TLS "Build IXWebSocket with TLS (OpenSSL) support" ON)
+
 find_package(ixwebsocket QUIET)
 if(NOT ixwebsocket_FOUND)
   message(STATUS "IXWebSocket not found — fetching via FetchContent")
@@ -61,7 +63,10 @@ if(NOT ixwebsocket_FOUND)
     URL https://github.com/machinezone/IXWebSocket/archive/refs/tags/v11.4.6.tar.gz
     URL_HASH SHA256=c024334f8e45980836c67008979a884d6dcc5ef067dd2eb1fa7241f4c17ddc32
   )
-  set(USE_TLS OFF CACHE BOOL "" FORCE)
+  set(USE_TLS ${PJ_BRIDGE_TLS} CACHE BOOL "" FORCE)
+  if(PJ_BRIDGE_TLS)
+    set(USE_OPEN_SSL ON CACHE BOOL "" FORCE)
+  endif()
   set(USE_ZLIB ON CACHE BOOL "" FORCE)
   set(IXWEBSOCKET_INSTALL OFF CACHE BOOL "" FORCE)
   FetchContent_MakeAvailable(ixwebsocket)
@@ -93,6 +98,7 @@ add_library(pj_bridge_app STATIC
   app/src/message_serializer.cpp
   app/src/bridge_server.cpp
   app/src/standalone_event_loop.cpp
+  app/src/whitelist_filter.cpp
 )
 
 target_include_directories(pj_bridge_app PUBLIC
@@ -241,11 +247,13 @@ if(BUILD_TESTING AND ament_cmake_FOUND)
 
   ament_add_gtest(${PROJECT_NAME}_tests
     tests/unit/test_websocket_middleware.cpp
+    tests/unit/test_bounded_frame_queue.cpp
     tests/unit/test_message_buffer.cpp
     tests/unit/test_session_manager.cpp
     tests/unit/test_message_serializer.cpp
     tests/unit/test_bridge_server.cpp
     tests/unit/test_protocol_constants.cpp
+    tests/unit/test_whitelist_filter.cpp
     tests/unit/test_topic_discovery.cpp
     tests/unit/test_schema_extractor.cpp
     tests/unit/test_generic_subscription_manager.cpp

README.md

@@ -23,6 +23,12 @@ independently.
 - **Multi-Client Support**: Multiple clients can connect simultaneously with shared subscriptions
 - **Runtime Schema Discovery**: Automatic extraction of message schemas from installed ROS2 packages on the server side.
 - **Large Message Stripping** (opt-in): Optional stripping of large array fields (Image, PointCloud2, LaserScan, OccupancyGrid) to reduce bandwidth while preserving metadata. Disabled by default — full message data is forwarded; enable with `strip_large_messages:=true` for low-bandwidth links
+- **Topic Whitelist**: Restrict which topics are visible/subscribable via full-match regex patterns (`topic_whitelist` / `--topic-whitelist`), mirroring foxglove_bridge's option of the same name
+- **QoS Depth Heuristics** (ROS2 only): KEEP_LAST subscription depth is derived from the discovered publishers' depths and clamped to a configurable `[min_qos_depth, max_qos_depth]` range
+- **Pushed Topic Advertisement** (opt-in): Clients can subscribe to a `topics_changed` notification instead of polling `get_topics`, at a configurable `topic_poll_interval`
+- **Slow-Client Backpressure**: A bounded, drop-oldest per-client send queue (`client_backlog_size`) keeps a lagging client from blocking the publish loop or growing an unbounded backlog, instead of disconnecting it
+- **Latched Topic Replay** (ROS2 only): New subscribers to a `TRANSIENT_LOCAL` topic (e.g. `/tf_static`) immediately receive the retained last message instead of waiting for the next publish
+- **TLS / `wss://`** (opt-in): Serve the WebSocket endpoint over TLS with a server certificate + private key (`tls`/`--certfile`+`--keyfile`); clients connect via `wss://` instead of `ws://`
 
 ## CI Status
 
@@ -33,12 +39,39 @@ independently.
 
 ## Configuration Parameters
 
+### ROS2 (via `--ros-args -p`)
+
 | Parameter | Type | Default | Description |
 |-----------|------|---------|-------------|
 | `port` | int | 9090 | WebSocket server port |
 | `publish_rate` | double | 50.0 | Aggregation publish rate in Hz |
 | `session_timeout` | double | 10.0 | Client timeout duration in seconds |
 | `strip_large_messages` | bool | false | Opt-in: strip large arrays from Image, PointCloud2, LaserScan, OccupancyGrid messages |
+| `topic_whitelist` | string array | `[".*"]` | Full-match regex patterns (ECMAScript) restricting visible/subscribable topics |
+| `min_qos_depth` | int | 1 | ROS2 only: minimum KEEP_LAST subscription depth after aggregating publisher depths |
+| `max_qos_depth` | int | 100 | ROS2 only: maximum KEEP_LAST subscription depth after aggregating publisher depths |
+| `topic_poll_interval` | double | 1.0 | Seconds between `topics_changed` notification polls; `0` disables polling |
+| `client_backlog_size` | int | 100 | Max binary frames queued per slow client before the oldest is dropped (must be `> 0`) |
+| `tls` | bool | false | Enable TLS (`wss://`); requires `certfile` and `keyfile` |
+| `certfile` | string | `""` | TLS server certificate file |
+| `keyfile` | string | `""` | TLS private key file |
+
+### FastDDS / RTI (via CLI flags)
+
+| Flag | Type | Default | Description |
+|------|------|---------|-------------|
+| `--domains`, `-d` | int list | (required) | DDS domain IDs |
+| `--port`, `-p` | int | 9090 | WebSocket server port |
+| `--publish-rate` | double | 50.0 | Aggregation publish rate in Hz |
+| `--session-timeout` | double | 10.0 | Client timeout duration in seconds |
+| `--log-level` | string | `info` | Log level (trace, debug, info, warn, error) |
+| `--stats` | flag | off | Print statistics every 5 seconds |
+| `--topic-whitelist` | string list | `.*` | Full-match regex patterns (ECMAScript), repeatable |
+| `--topic-poll-interval` | double | 1.0 | Seconds between `topics_changed` notification polls; `0` disables polling |
+| `--client-backlog-size` | int | 100 | Max binary frames queued per slow client before the oldest is dropped (range `1`-`1000000`) |
+| `--certfile` | string | (none) | TLS server certificate file; enables `wss://`, requires `--keyfile` |
+| `--keyfile` | string | (none) | TLS private key file; enables `wss://`, requires `--certfile` |
+| `--qos-profile` | string | (none) | RTI only: QoS profile XML file path |
 
 ## Just "Download and Run"
 
@@ -73,6 +106,8 @@ chmod +x pj_bridge_ros2-humble-x86_64.AppImage
 
 All dependencies (spdlog, nlohmann_json, ZSTD) are provided by the dependency manager. IXWebSocket is resolved via `find_package` first, with a FetchContent fallback for colcon builds. Only `tl::expected` is vendored.
 
+TLS (`wss://`) support depends on IXWebSocket being built with OpenSSL. The CMake option `PJ_BRIDGE_TLS` (default `ON`) controls this for the FetchContent path (`-DPJ_BRIDGE_TLS=OFF` to disable); a system/conda-provided IXWebSocket must likewise have been built with TLS. See [docs/API.md](docs/API.md#tls--wss) for details.
+
 ### ROS2 — Pixi
 
 [Pixi](https://pixi.sh) manages the full toolchain including ROS2 via [RoboStack](https://robostack.github.io/).

app/include/pj_bridge/bridge_server.hpp

@@ -31,6 +31,7 @@
 #include "pj_bridge/session_manager.hpp"
 #include "pj_bridge/subscription_manager_interface.hpp"
 #include "pj_bridge/topic_source_interface.hpp"
+#include "pj_bridge/whitelist_filter.hpp"
 
 namespace pj_bridge {
 
@@ -63,12 +64,13 @@ class BridgeServer {
    * @param port Server port (default: 9090)
    * @param session_timeout Session timeout in seconds (default: 10.0)
    * @param publish_rate Message aggregation publish rate in Hz (default: 50.0)
+   * @param whitelist Topic whitelist filter (default: matches everything)
    */
   explicit BridgeServer(
       std::shared_ptr<TopicSourceInterface> topic_source,
       std::shared_ptr<SubscriptionManagerInterface> subscription_manager,
       std::shared_ptr<MiddlewareInterface> middleware, int port = 9090, double session_timeout = 10.0,
-      double publish_rate = 50.0);
+      double publish_rate = 50.0, WhitelistFilter whitelist = {});
 
   /// Shuts down middleware before members are destroyed, preventing
   /// disconnect callbacks from firing into a partially destroyed object.
@@ -106,6 +108,18 @@ class BridgeServer {
    */
   void check_session_timeouts();
 
+  /**
+   * @brief Check whether the (whitelist-filtered) topic set has changed and
+   * notify opted-in clients
+   *
+   * Called externally by the event loop (e.g. every topic_poll_interval
+   * seconds). The first call after construction only takes a snapshot and
+   * sends nothing; subsequent calls diff against that snapshot and send a
+   * `topics_changed` notification (added/removed) to every session that
+   * called `subscribe_topic_updates`, but only when the diff is non-empty.
+   */
+  void check_topic_changes();
+
   /**
    * @brief Get the number of active client sessions
    */
@@ -130,11 +144,31 @@ class BridgeServer {
   std::string handle_heartbeat(const std::string& client_id, const nlohmann::json& request);
   std::string handle_pause(const std::string& client_id, const nlohmann::json& request);
   std::string handle_resume(const std::string& client_id, const nlohmann::json& request);
+  std::string handle_subscribe_topic_updates(const std::string& client_id, const nlohmann::json& request);
+  std::string handle_unsubscribe_topic_updates(const std::string& client_id, const nlohmann::json& request);
   std::string create_error_response(
       const std::string& error_code, const std::string& message, const nlohmann::json& request) const;
   void inject_response_fields(nlohmann::json& response, const nlohmann::json& request) const;
   void cleanup_session(const std::string& client_id);
 
+  /// Release one middleware subscription ref for @p topic_name. When the
+  /// last ref is gone (subscription destroyed), also clears the topic's
+  /// latched state in the message buffer: a stale retained sample must not
+  /// be replayed to the next fresh subscriber, who receives the current
+  /// sample from DDS transient_local redelivery instead. Call with
+  /// cleanup_mutex_ held (same requirement as unsubscribe itself).
+  void release_subscription_ref(const std::string& topic_name);
+
+  /// Latched (transient_local) replay bookkeeping for a topic whose
+  /// middleware subscription ref was just (re)acquired for @p client_id —
+  /// shared by handle_subscribe and handle_resume. Records the topic's
+  /// latched flag in the message buffer and, when a retained sample is
+  /// available for replay, serializes it into a single-message binary frame
+  /// queued in pending_replays_ (flushed by process_single_request right
+  /// after the handler's response is sent). Call with cleanup_mutex_ held;
+  /// acquires only the leaf replays_mutex_ and makes no middleware calls.
+  void collect_latched_replay(const std::string& client_id, const std::string& topic_name);
+
   // Backend interfaces (owned)
   std::shared_ptr<TopicSourceInterface> topic_source_;
   std::shared_ptr<SubscriptionManagerInterface> subscription_manager_;
@@ -148,6 +182,7 @@ class BridgeServer {
   int port_;
   double session_timeout_;
   double publish_rate_;
+  WhitelistFilter whitelist_;
 
   // State
   std::atomic<bool> initialized_;
@@ -164,11 +199,34 @@ class BridgeServer {
   //   cleanup_mutex_ > last_sent_mutex_ > stats_mutex_
   // SessionManager::mutex_ may be acquired while holding any of these.
   // Never acquire a higher-order lock while holding a lower-order one.
+  // topics_mutex_ is a LEAF lock: it is held only to compute the
+  // added/removed diff and swap in the new known_topics_ snapshot. Never
+  // call middleware_ or session_manager_ (or acquire any other lock in this
+  // class) while holding topics_mutex_ — release it before sending
+  // notifications.
+  // replays_mutex_ is likewise a LEAF lock: it only guards pending_replays_
+  // map accesses. Never hold it while acquiring any other lock or calling
+  // middleware_ — move the frames out, release, then send.
   std::mutex cleanup_mutex_;
 
   // Per-client per-topic last-sent timestamp (nanoseconds) for rate limiting
   std::unordered_map<std::string, std::unordered_map<std::string, uint64_t>> last_sent_times_;
   std::mutex last_sent_mutex_;
+
+  // Whitelist-filtered topic name -> type snapshot, used by check_topic_changes()
+  // to detect additions/removals/type-changes. LEAF lock (see comment above).
+  std::unordered_map<std::string, std::string> known_topics_;
+  bool topics_snapshot_taken_{false};
+  std::mutex topics_mutex_;
+
+  // Latched (transient_local) replay frames queued by handle_subscribe and
+  // handle_resume (via collect_latched_replay), flushed by
+  // process_single_request right AFTER the handler's response is sent — the
+  // client must receive the response (which carries the topic's schema on
+  // subscribe) before the binary frame that needs it. LEAF lock (see
+  // comment above).
+  std::unordered_map<std::string, std::vector<std::vector<uint8_t>>> pending_replays_;
+  std::mutex replays_mutex_;
 };
 
 }  // namespace pj_bridge

app/include/pj_bridge/message_buffer.hpp

@@ -25,8 +25,10 @@
 #include <functional>
 #include <memory>
 #include <mutex>
+#include <optional>
 #include <string>
 #include <unordered_map>
+#include <unordered_set>
 #include <vector>
 
 namespace pj_bridge {
@@ -78,12 +80,40 @@ class MessageBuffer {
   /// Return the total number of messages across all topics.
   size_t size() const;
 
+  /// Mark (or unmark) @p topic_name as latched (transient_local). While a
+  /// topic is latched, add_message() additionally retains its newest sample
+  /// in a TTL-exempt, one-entry-per-topic store so that later subscribers to
+  /// an already-shared subscription can be replayed the current value even
+  /// after it has aged out of the normal buffer. Setting latched to false
+  /// erases the retained sample.
+  void set_latched(const std::string& topic_name, bool latched);
+
+  /// Return the retained latched sample for @p topic_name, or nullopt if the
+  /// topic is not latched or no sample has been retained yet.
+  std::optional<BufferedMessage> get_latched(const std::string& topic_name) const;
+
+  /// Like get_latched(), but additionally returns nullopt while the topic
+  /// still has messages in the normal buffer. In that case the upcoming
+  /// publish cycle will deliver the sample to the new subscriber anyway —
+  /// replaying it would produce a duplicate. Replay is only needed for
+  /// samples that already left the normal buffer (drained by
+  /// move_messages() or TTL-evicted).
+  std::optional<BufferedMessage> get_latched_for_replay(const std::string& topic_name) const;
+
  private:
   mutable std::mutex mutex_;
   std::unordered_map<std::string, std::deque<BufferedMessage>> topic_buffers_;
   uint64_t max_message_age_ns_;
   ClockFn clock_fn_;
 
+  // Latched (transient_local) topic support: topics currently flagged as
+  // latched, and the single newest retained sample per latched topic.
+  // Bounded at one message per latched topic; entries otherwise persist for
+  // the lifetime of the process (not TTL-cleaned, not drained by
+  // move_messages()).
+  std::unordered_set<std::string> latched_topics_;
+  std::unordered_map<std::string, BufferedMessage> latched_last_;
+
   uint64_t now_ns() const;
   void cleanup_old_messages();
 };