Add raw mavlink zenoh driver by joaoantoniocardoso · Pull Request #215 · bluerobotics/mavlink-server

joaoantoniocardoso · 2026-06-09T16:58:07Z

patrickelectric · 2026-06-09T17:14:20Z

@@ -1 +1,102 @@
 pub mod json;
+


over having this here, isn't it better to have session.rs ?

patrickelectric · 2026-06-10T12:30:17Z

+                            }
+
+                            loop {
+                                tokio::time::sleep(tokio::time::Duration::from_secs(86400)).await;


Can you explain why is there a 24hs loop here ?

It was a "forever" loop; it feels really odd indeed, I'll simplify it.

patrickelectric · 2026-06-10T12:32:26Z

+async fn init_session_manager() {
+    SESSION_RX
+        .get_or_init(|| async {
+            let (tx, rx) = watch::channel(None);


It would be nice to have a comment explaining the reason behind this task, it took me a while to understand what is doing and why is doing. But it would be nice just to have the reason of the loop itself, the how is indirectly related to the chosen implementation.

patrickelectric · 2026-06-10T12:44:18Z

+            return session;
+        }
+
+        if rx.changed().await.is_err() {


Some things that would improve the code, one of the is to add a log when the session changes and also a log when the session is created.

patrickelectric · 2026-06-10T12:51:22Z

+                        error!("Failed to decode packet: {decode_error:?}");
+                        continue;
+                    }
+                    Ok(None) => break,


It would be nice to have a comment explaining why we are breaking here, having None as "not enough data" is not clear.

I'm gonna move to a Framed adapter, following the same pattern as the other drivers.

patrickelectric · 2026-06-10T12:56:04Z

+                    }
+                };
+
+                let bus_message = Arc::new(Protocol::new("zenohraw", packet));


Since this string needs to match with the one in the sender to avoid loop-back, this should be stored in a static variable.

patrickelectric · 2026-06-10T12:59:57Z

+                        debug!(
+                            "Dropping message: on_message_input callback returned error: {error:?}"
+                        );
+                        continue;


what ? should this be a break ? If so... why are we not printing the others calls status ?

patrickelectric · 2026-06-10T13:00:51Z

+                    debug!(
+                        "Dropping message: on_message_output callback returned error: {error:?}"
+                    );
+                    continue 'mainloop;


wait, how about the other futures calls ? Same as my previous comment.

wait.. This should continue the outer loop, skipping the message transmission.

From patrick from the past: 4b22893

patrickelectric · 2026-06-11T11:05:07Z

        let mut timestamp_bytes = [0u8; 8];

-        loop {
+        'mainloop: loop {


can you give more details about the "fix" in the commit description ?

patrickelectric · 2026-06-11T15:13:49Z

+            let bus_message = Arc::new(Protocol::from_mavlink_raw(
+                content.header.inner,
+                &content.message,
+                DRIVER_IDENTIFIER,


This was added in 9e22904

patrickelectric · 2026-06-11T15:15:49Z

+static SESSION_RX: OnceCell<watch::Receiver<Option<Arc<zenoh::Session>>>> = OnceCell::const_new();
+
+#[instrument(level = "debug")]
+pub(crate) fn build_config() -> anyhow::Result<zenoh::Config> {


why is this here and in session.rs ?

patrickelectric · 2026-06-11T18:10:54Z

+static SESSION_RX: OnceCell<watch::Receiver<Option<Arc<zenoh::Session>>>> = OnceCell::const_new();
+
+#[instrument(level = "debug")]
+pub(crate) fn build_config() -> anyhow::Result<zenoh::Config> {


This is duplicated in session.rs

patrickelectric · 2026-06-11T18:12:05Z

+
+use crate::cli::zenoh_config_file;
+
+static SESSION_RX: OnceCell<watch::Receiver<Option<Arc<zenoh::Session>>>> = OnceCell::const_new();


This is being added here and removed: 1b06fb4

patrickelectric · 2026-06-11T18:17:47Z

+            match decode_result {
+                Ok(Some(item)) => return Poll::Ready(Some(Ok(item))),
+                Ok(None) if self.end_of_stream => return Poll::Ready(None),
+                Ok(None) => { /* need more data */ }


…a singleton

patrickelectric · 2026-06-12T07:58:30Z

+
+    fn start_send(self: Pin<&mut Self>, item: Packet) -> Result<(), Self::Error> {
+        self.publisher
+            .put(item.bytes().clone())


It appears that this is missing .encoding, where the default is ZENOH_BYTES that represents the usage of zenoh specific serialization methods. For this, it appears that https://docs.rs/zenoh/latest/zenoh/bytes/struct.Encoding.html#associatedconstant.APPLICATION_OCTET_STREAM is the best.

great catch, I'll add it

Adjusted for raw and JSON.

…gurations

These on_message_input/on_message_output callbacks are supposed to skip/drop messages when returning an "Err". The previous code was continuing inside the for loop, which won't skip/drop the message.

cargo: update zenoh crate

190e960

joaoantoniocardoso requested a review from patrickelectric June 9, 2026 16:58

joaoantoniocardoso force-pushed the add_raw_mavlink_zenoh_driver branch from 638f44e to 966a05d Compare June 9, 2026 18:33

joaoantoniocardoso marked this pull request as ready for review June 9, 2026 18:33

joaoantoniocardoso mentioned this pull request Jun 9, 2026

Improve local recording: implementation discussion bluerobotics/blueos-recorder#19

Open

patrickelectric requested changes Jun 10, 2026

View reviewed changes

joaoantoniocardoso force-pushed the add_raw_mavlink_zenoh_driver branch from 966a05d to 7dec60c Compare June 11, 2026 01:23

joaoantoniocardoso requested a review from patrickelectric June 11, 2026 01:47

joaoantoniocardoso mentioned this pull request Jun 11, 2026

Improve CI #219

Draft

patrickelectric reviewed Jun 11, 2026

View reviewed changes

joaoantoniocardoso force-pushed the add_raw_mavlink_zenoh_driver branch from 7dec60c to e955c67 Compare June 11, 2026 18:45

joaoantoniocardoso requested a review from patrickelectric June 11, 2026 19:28

joaoantoniocardoso force-pushed the add_raw_mavlink_zenoh_driver branch from e955c67 to 4b49832 Compare June 11, 2026 20:37

joaoantoniocardoso added 2 commits June 11, 2026 17:40

src: lib: drivers: zenoh: Extract zenoh session into its own file as …

56e6b77

…a singleton

src: lib: driver: zenoh: Extract json driver into its own file

6acccfa

joaoantoniocardoso force-pushed the add_raw_mavlink_zenoh_driver branch from 4b49832 to 5a75631 Compare June 11, 2026 20:40

patrickelectric approved these changes Jun 12, 2026

View reviewed changes

joaoantoniocardoso added 12 commits June 12, 2026 14:00

src: lib: drivers: zenoh: Introduce ZenohRaw driver

9e03182

src: lib: drivers: zenoh: json: Add topic prefix constant

3fca2d9

src: lib: drivers: zenoh: json: Align run loop with other drivers

7fafe31

src: lib: drivers: zenoh: json: Cache publishers for hot publish paths

a5fb0d0

src: lib: drivers: zenoh: json: Add schema and adjust publisher confi…

87701fb

…gurations

src: lib: Add missing zenoh to CLI tests

76afccb

src: lib: drivers: Fix 'continue' call

9774de0

These on_message_input/on_message_output callbacks are supposed to skip/drop messages when returning an "Err". The previous code was continuing inside the for loop, which won't skip/drop the message.

cargo: Add zenoh to tests dependencies

17c5892

tests: common: Add ephemeral port and stats wait helpers

139859f

tests: Enable parallel-safe tcp, udp, and websocket integration tests

787a3da

tests: zenoh: Add bidirectional integration tests

4b528a8

.github: workflows: Adopt cargo-nextest for parallel test execution

47b2cdc

joaoantoniocardoso force-pushed the add_raw_mavlink_zenoh_driver branch from 5a75631 to 47b2cdc Compare June 12, 2026 17:10

patrickelectric approved these changes Jun 12, 2026

View reviewed changes

patrickelectric merged commit bfc96df into bluerobotics:master Jun 12, 2026
18 of 19 checks passed


		use crate::cli::zenoh_config_file;

		static SESSION_RX: OnceCell<watch::Receiver<Option<Arc<zenoh::Session>>>> = OnceCell::const_new();

Conversation

joaoantoniocardoso commented Jun 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

joaoantoniocardoso Jun 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

joaoantoniocardoso commented Jun 9, 2026 •

edited

Loading

joaoantoniocardoso Jun 10, 2026 •

edited

Loading