feature gate std stuff

Cargo.lock

@@ -1,6 +1,6 @@
 # This file is automatically @generated by Cargo.
 # It is not intended for manual editing.
-version = 3
+version = 4
 
 [[package]]
 name = "addr2line"
@@ -80,6 +80,8 @@ version = "0.1.0"
 dependencies = [
  "derive-getters",
  "derive-new",
+ "enumflags2",
+ "itertools",
  "log",
  "tokio",
  "tokio-util",
@@ -249,6 +251,32 @@ dependencies = [
  "syn 2.0.48",
 ]
 
+[[package]]
+name = "either"
+version = "1.15.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719"
+
+[[package]]
+name = "enumflags2"
+version = "0.7.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1027f7680c853e056ebcec683615fb6fbbc07dbaa13b4d5d9442b146ded4ecef"
+dependencies = [
+ "enumflags2_derive",
+]
+
+[[package]]
+name = "enumflags2_derive"
+version = "0.7.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "67c78a4d8fdf9953a5c9d458f9efe940fd97a0cab0941c075a813ac594733827"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.48",
+]
+
 [[package]]
 name = "env_logger"
 version = "0.9.0"
@@ -323,6 +351,15 @@ version = "0.3.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ce23b50ad8242c51a442f3ff322d56b02f08852c77e4c0b4d3fd684abc89c683"
 
+[[package]]
+name = "itertools"
+version = "0.14.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2b192c782037fadd9cfa75548310488aabdbf3d2da73885b31bd0abd03351285"
+dependencies = [
+ "either",
+]
+
 [[package]]
 name = "lazy_static"
 version = "1.4.0"

atem-connection-rs/Cargo.toml

@@ -1,11 +1,18 @@
 [package]
 name = "atem-connection-rs"
 version = "0.1.0"
-edition = "2021"
+edition = "2024"
 
 [dependencies]
 derive-getters = "0.2.0"
 derive-new = "0.6.0"
+itertools = {version = "0.14.0", default-features = false}
 log = "0.4.14"
-tokio = { version = "1.13.0", features = ["full"] }
+tokio = { version = "1.13.0", features = ["full"], optional = true }
-tokio-util = "0.7.10"
+tokio-util = { version = "0.7.10", optional = true }
+enumflags2 = { version = "0.7.12", default-features = false }
+
+[features]
+default = ["std"]
+
+std = ["dep:tokio", "dep:tokio-util", "itertools/use_std"]

atem-connection-rs/src/atem.rs

@@ -1,9 +1,11 @@
 use std::{
+    boxed::Box,
     collections::{HashMap, VecDeque},
     net::SocketAddr,
     ops::DerefMut,
     sync::Arc,
     time::Duration,
+    vec::Vec,
 };
 
 use tokio::{select, sync::Semaphore};

atem-connection-rs/src/atem_lib/atem_field.rs

@@ -0,0 +1,87 @@
+//! Definitions and decoding of ATEM protocol fields
+
+/// An uninterpreted ATEM protocol field - a 4-ascii character type plus variable length data
+pub struct RawField<'a> {
+    pub r#type: &'a str,
+    pub data: &'a [u8],
+}
+
+#[derive(Debug)]
+pub struct _Ver {
+    pub major: u16,
+    pub minor: u16,
+}
+
+impl<'a> Field<'a> for _Ver {
+    const TYPE: &'static str = "_ver";
+
+    fn decode(data: &'a [u8]) -> Result<Self, FieldParsingError> {
+        let data = checked_len::<4>(data)?;
+        Ok(Self {
+            major: u16::from_be_bytes(data[0..=1].try_into().unwrap()),
+            minor: u16::from_be_bytes(data[2..=3].try_into().unwrap()),
+        })
+    }
+}
+
+#[derive(Debug)]
+pub struct PrvI {
+    pub m_e_index: u8,
+    pub source_index: u16,
+    pub pvw_in_pgm: bool,
+}
+
+impl<'a> Field<'a> for PrvI {
+    const TYPE: &'static str = "PrvI";
+    fn decode(data: &'a [u8]) -> Result<Self, FieldParsingError> {
+        let data = checked_len::<8>(data)?;
+        Ok(Self {
+            m_e_index: data[0],
+            source_index: u16::from_be_bytes(data[2..=3].try_into().unwrap()),
+            pvw_in_pgm: data[4] != 0,
+        })
+    }
+}
+
+#[derive(Debug)]
+pub struct PrgI {
+    pub m_e_index: u8,
+    pub source_index: u16,
+}
+
+impl<'a> Field<'a> for PrgI {
+    const TYPE: &'static str = "PrgI";
+    fn decode(data: &'a [u8]) -> Result<Self, FieldParsingError> {
+        let data = checked_len::<4>(data)?;
+        Ok(Self {
+            m_e_index: data[0],
+            source_index: u16::from_be_bytes(data[2..=3].try_into().unwrap()),
+        })
+    }
+}
+
+pub trait Field<'a>: Sized {
+    const TYPE: &'static str;
+    fn decode(data: &'a [u8]) -> Result<Self, FieldParsingError>;
+    fn try_from_raw(raw: RawField<'a>) -> Result<Self, FieldParsingError> {
+        if Self::TYPE != raw.r#type {
+            Err(FieldParsingError::MismatchedFieldType)
+        } else {
+            Self::decode(&raw.data)
+        }
+    }
+}
+
+#[derive(Debug)]
+pub enum FieldParsingError {
+    UnexpectedLength { expected: usize, got: usize },
+    MismatchedFieldType,
+}
+
+fn checked_len<'a, const LEN: usize>(data: &'a [u8]) -> Result<&'a [u8; LEN], FieldParsingError> {
+    data.try_into()
+        .map_err(|_| FieldParsingError::UnexpectedLength {
+            expected: LEN,
+            got: data.len(),
+        })
+}

atem-connection-rs/src/atem_lib/atem_packet.rs

@@ -1,115 +1,185 @@
+use core::{fmt::Display, str};
+use enumflags2::{BitFlags, bitflags};
+
+use super::atem_field::{Field, FieldParsingError, RawField};
+
+/// The "hello" packet to start communication with the ATEM
+pub const COMMAND_CONNECT_HELLO: [u8; 20] = [
+    0x10, 0x14, 0x53, 0xab, 0x00, 0x00, 0x00, 0x00, 0x00, 0x3a, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00,
+    0x00, 0x00, 0x00, 0x00,
+];
+
 #[derive(Debug)]
-pub struct AtemPacket<'packet_buffer> {
+pub struct AtemPacket<T: AsRef<[u8]>> {
-    flags: u8,
+    buf: T,
-    session_id: u16,
-    remote_packet_id: u16,
-    retransmit_requested_from_packet_id: Option<u16>,
-    ack_reply: Option<u16>,
-    body: Option<&'packet_buffer [u8]>,
 }
 
+impl<'a> TryFrom<&'a [u8]> for AtemPacket<&'a [u8]> {
+    type Error = AtemPacketErr;
+    fn try_from(buf: &'a [u8]) -> Result<Self, Self::Error> {
+        AtemPacket::new_checked(buf)
+    }
+}
+
+#[derive(Debug)]
 pub enum AtemPacketErr {
-    TooShort(String),
+    /// The packet was too short
-    LengthDiffers(String),
+    TooShort {
+        got: usize,
+    },
+    /// The packet's stated and actual lengths were different
+    LengthDiffers {
+        expected: u16,
+        got: usize,
+    },
+    InvalidFlags,
 }
 
-#[derive(PartialEq)]
+#[bitflags]
+#[repr(u8)]
+#[derive(PartialEq, Copy, Clone, Debug)]
 pub enum PacketFlag {
-    AckRequest,
+    AckRequest = 0x1,
-    NewSessionId,
+    NewSessionId = 0x2,
-    IsRetransmit,
+    IsRetransmit = 0x4,
-    RetransmitRequest,
+    RetransmitRequest = 0x8,
-    AckReply,
+    AckReply = 0x10,
 }
 
-impl From<PacketFlag> for u8 {
+impl<T: AsRef<[u8]>> AtemPacket<T> {
-    fn from(flag: PacketFlag) -> Self {
+    pub fn new_checked(buf: T) -> Result<Self, AtemPacketErr> {
-        match flag {
+        let len = buf.as_ref().len();
-            PacketFlag::AckRequest => 0x01,
+        if len < 12 {
-            PacketFlag::NewSessionId => 0x02,
+            return Err(AtemPacketErr::TooShort {
-            PacketFlag::IsRetransmit => 0x04,
+                got: buf.as_ref().len(),
-            PacketFlag::RetransmitRequest => 0x08,
+            });
-            PacketFlag::AckReply => 0x10,
         }
+        let p = Self { buf };
+        if p.length() as usize != len {
+            return Err(AtemPacketErr::LengthDiffers {
+                expected: p.length(),
+                got: len,
+            });
+        }
+        // Check flags are valid
+        let _: BitFlags<PacketFlag> = p
+            .flags_raw()
+            .try_into()
+            .map_err(|_| AtemPacketErr::InvalidFlags)?;
+        Ok(p)
+    }
+    pub fn length(&self) -> u16 {
+        u16::from_be_bytes(self.buf.as_ref()[0..=1].try_into().unwrap()) & 0x07ff
+    }
+
+    fn flags_raw(&self) -> u8 {
+        self.buf.as_ref()[0] >> 3
+    }
+
+    pub fn flags(&self) -> BitFlags<PacketFlag> {
+        // We `unwrap` here, but given we check the flags in the constructor this should never panic.
+        self.flags_raw().try_into().unwrap()
     }
-}
 
-impl<'packet_buffer> AtemPacket<'packet_buffer> {
     pub fn session_id(&self) -> u16 {
-        self.session_id
+        u16::from_be_bytes(self.buf.as_ref()[2..=3].try_into().unwrap())
     }
 
-    pub fn remote_packet_id(&self) -> u16 {
+    pub fn ack_number(&self) -> u16 {
-        self.remote_packet_id
+        u16::from_be_bytes(self.buf.as_ref()[4..=5].try_into().unwrap())
     }
 
-    pub fn body(&self) -> Option<&[u8]> {
+    pub fn remote_sequence_number(&self) -> u16 {
-        self.body
+        u16::from_be_bytes(self.buf.as_ref()[9..=10].try_into().unwrap())
+    }
+
+    pub fn local_sequence_number(&self) -> u16 {
+        u16::from_be_bytes(self.buf.as_ref()[10..=11].try_into().unwrap())
     }
 
     pub fn retransmit_request(&self) -> Option<u16> {
-        self.retransmit_requested_from_packet_id
+        self.flags()
+            .contains(PacketFlag::RetransmitRequest)
+            .then_some(u16::from_be_bytes([
+                self.buf.as_ref()[6],
+                self.buf.as_ref()[7],
+            ]))
     }
 
     pub fn ack_reply(&self) -> Option<u16> {
-        self.ack_reply
+        self.flags()
+            .contains(PacketFlag::AckReply)
+            .then_some(self.ack_number())
     }
 
-    pub fn has_flag(&self, flag: PacketFlag) -> bool {
+    /// Get an iterator over the `Field`s in this packet.
-        self.flags & u8::from(flag) > 0
+    ///
+    /// Returns None if this is a packet without fields.
+    pub fn raw_fields(&self) -> RawFields {
+        // TODO: do we only ever get newsessionid during the handshake (i.e. not in a packet with fields)?
+        let has_fields = !self.flags().contains(PacketFlag::NewSessionId);
+        RawFields::new(if has_fields { self.body() } else { &[] })
     }
-}
 
-impl<'packet_buffer> TryFrom<&'packet_buffer [u8]> for AtemPacket<'packet_buffer> {
+    pub fn fields<'a, F: Field<'a>>(
-    type Error = AtemPacketErr;
+        &'a self,
-
+    ) -> impl Iterator<Item = Result<F, FieldParsingError>> + use<'a, F, T> {
-    fn try_from(buffer: &'packet_buffer [u8]) -> Result<Self, Self::Error> {
+        self.raw_fields().filter_map(|f| match F::try_from_raw(f) {
-        if buffer.len() < 12 {
+            Err(FieldParsingError::MismatchedFieldType) => None,
-            return Err(AtemPacketErr::TooShort(format!(
+            x => Some(x),
-                "Invalid packet from ATEM {:x?}",
-                buffer
-            )));
-        }
-
-        let length = u16::from_be_bytes([buffer[0], buffer[1]]) & 0x07ff;
-        if length as usize != buffer.len() {
-            return Err(AtemPacketErr::LengthDiffers(format!(
-                "Length of message differs, expected {} got {}",
-                length,
-                buffer.len()
-            )));
-        }
-
-        let flags = buffer[0] >> 3;
-        let session_id = u16::from_be_bytes([buffer[2], buffer[3]]);
-        let remote_packet_id = u16::from_be_bytes([buffer[10], buffer[11]]);
-
-        let body = if buffer.len() > 12 {
-            Some(&buffer[12..])
-        } else {
-            None
-        };
-
-        let retransmit_requested_from_packet_id =
-            if flags & u8::from(PacketFlag::RetransmitRequest) > 0 {
-                Some(u16::from_be_bytes([buffer[6], buffer[7]]))
-            } else {
-                None
-            };
-
-        let ack_reply = if flags & u8::from(PacketFlag::AckReply) > 0 {
-            Some(u16::from_be_bytes([buffer[4], buffer[5]]))
-        } else {
-            None
-        };
-
-        Ok(AtemPacket {
-            flags,
-            session_id,
-            remote_packet_id,
-            body,
-            retransmit_requested_from_packet_id,
-            ack_reply,
         })
     }
+
+    pub fn body(&self) -> &[u8] {
+        &self.buf.as_ref()[12..]
+    }
+}
+
+impl<T: AsRef<[u8]>> Display for AtemPacket<T> {
+    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+        write!(
+            f,
+            "len: {}, sid: {}, flags: {}, ack: {}, localseq: {}, remoteseq: {}",
+            self.length(),
+            self.session_id(),
+            self.flags(),
+            self.ack_number(),
+            self.remote_sequence_number(),
+            self.local_sequence_number(),
+        )
+    }
+}
+
+/// Created by [`AtemPacket::raw_fields`]
+pub struct RawFields<'a> {
+    data: &'a [u8],
+    // The offset of the next field in the packet
+    offset: usize,
+}
+
+impl<'a> RawFields<'a> {
+    pub(crate) fn new(data: &'a [u8]) -> Self {
+        Self { data, offset: 0 }
+    }
+}
+
+impl<'a> Iterator for RawFields<'a> {
+    type Item = RawField<'a>;
+    fn next(&mut self) -> Option<Self::Item> {
+        let remain = self.data.len() - self.offset;
+        if remain == 0 {
+            return None;
+        } else if remain < 8 {
+            // TODO: is 8 indeed the minimum size for something here? (i.e. no field data)
+            panic!("Oh no");
+        }
+
+        let length =
+            u16::from_be_bytes(self.data[self.offset..=self.offset + 1].try_into().unwrap());
+        // TODO: sanity check length
+        let r#type = str::from_utf8(&self.data[self.offset + 4..=self.offset + 7]).unwrap();
+        let data = &self.data[self.offset + 8..self.offset + (length as usize)];
+        self.offset += length as usize;
+        Some(RawField { r#type, data })
+    }
 }

atem-connection-rs/src/atem_lib/atem_socket.rs

@@ -1,12 +1,16 @@
 use std::{
-    collections::VecDeque,
+    borrow::ToOwned,
     fmt::Display,
     io,
     net::SocketAddr,
+    string::{String, ToString},
     sync::Arc,
-    time::{Duration, SystemTime},
+    time::{Duration, Instant},
+    vec,
+    vec::Vec,
 };
 
+use itertools::Itertools;
 use tokio::{
     net::UdpSocket,
     select,
@@ -14,7 +18,7 @@ use tokio::{
 };
 
 use crate::{
-    atem_lib::{atem_packet::AtemPacket, atem_util},
+    atem_lib::atem_packet::{self, AtemPacket, COMMAND_CONNECT_HELLO},
     commands::{
         command_base::{BasicWritableCommand, DeserializedCommand},
         parse_commands::deserialize_commands,
@@ -24,10 +28,10 @@ use crate::{
 
 use super::atem_packet::PacketFlag;
 
-const IN_FLIGHT_TIMEOUT: u64 = 60;
+const IN_FLIGHT_TIMEOUT: Duration = Duration::from_millis(60);
-const CONNECTION_TIMEOUT: u64 = 5000;
+const CONNECTION_TIMEOUT: Duration = Duration::from_millis(5000);
-const CONNECTION_RETRY_INTERVAL: u64 = 1000;
+const CONNECTION_RETRY_INTERVAL: Duration = Duration::from_millis(1000);
-const RETRANSMIT_CHECK_INTERVAL: u64 = 1000;
+const RETRANSMIT_CHECK_INTERVAL: Duration = Duration::from_millis(1000);
 const MAX_PACKET_RETRIES: u16 = 10;
 const MAX_PACKET_ID: u16 = 1 << 15;
 const MAX_PACKET_PER_ACK: u16 = 16;
@@ -92,8 +96,8 @@ impl AtemSocketCommand {
 
 pub struct AtemSocket {
     connection_state: ConnectionState,
-    reconnect_timer: Option<SystemTime>,
+    reconnect_timer: Option<Instant>,
-    retransmit_timer: Option<SystemTime>,
+    retransmit_timer: Option<Instant>,
 
     next_tracking_id: u64,
 
@@ -105,10 +109,10 @@ pub struct AtemSocket {
 
     protocol_version: ProtocolVersion,
 
-    last_received_at: SystemTime,
+    last_received_at: Instant,
     last_received_packed_id: u16,
     in_flight: Vec<InFlightPacket>,
-    ack_timer: Option<SystemTime>,
+    ack_timer: Option<Instant>,
     received_without_ack: u16,
 
     atem_message_rx: tokio::sync::mpsc::Receiver<AtemSocketMessage>,
@@ -141,7 +145,7 @@ struct InFlightPacket {
     packet_id: u16,
     tracking_id: u64,
     payload: Vec<u8>,
-    pub last_sent: SystemTime,
+    pub last_sent: Instant,
     pub resent: u16,
 }
 
@@ -170,7 +174,7 @@ impl AtemSocket {
 
             protocol_version: ProtocolVersion::V7_2,
 
-            last_received_at: SystemTime::now(),
+            last_received_at: Instant::now(),
             last_received_packed_id: 0,
             in_flight: vec![],
             ack_timer: None,
@@ -258,7 +262,7 @@ impl AtemSocket {
         self.in_flight = vec![];
         log::debug!("Reconnect");
 
-        self.send_packet(&atem_util::COMMAND_CONNECT_HELLO).await;
+        self.send_packet(&COMMAND_CONNECT_HELLO).await;
         self.connection_state = ConnectionState::SynSent;
 
         Ok(())
@@ -299,7 +303,7 @@ impl AtemSocket {
             self.next_send_packet_id = 0;
         }
 
-        let opcode = u16::from(u8::from(PacketFlag::AckRequest)) << 11;
+        let opcode = u16::from(PacketFlag::AckRequest as u8) << 11;
 
         let mut buffer = vec![0; 20 + payload.len()];
 
@@ -320,7 +324,7 @@ impl AtemSocket {
             packet_id,
             tracking_id,
             payload: buffer,
-            last_sent: SystemTime::now(),
+            last_sent: Instant::now(),
             resent: 0,
         })
     }
@@ -338,17 +342,15 @@ impl AtemSocket {
             }
         }
         if let Some(ack_time) = self.ack_timer {
-            if ack_time <= SystemTime::now() {
+            if ack_time <= Instant::now() {
                 self.ack_timer = None;
                 self.received_without_ack = 0;
                 self.send_ack(self.last_received_packed_id).await;
             }
         }
         if let Some(reconnect_time) = self.reconnect_timer {
-            if reconnect_time <= SystemTime::now() {
+            if reconnect_time <= Instant::now() {
-                if self.last_received_at + Duration::from_millis(CONNECTION_TIMEOUT)
+                if self.last_received_at + CONNECTION_TIMEOUT <= Instant::now() {
-                    <= SystemTime::now()
-                {
                     log::debug!("{:?}", self.last_received_at);
                     log::debug!("Connection timed out, restarting");
                     self.restart_connection().await;
@@ -357,7 +359,7 @@ impl AtemSocket {
             }
         }
         if let Some(retransmit_time) = self.retransmit_timer {
-            if retransmit_time <= SystemTime::now() {
+            if retransmit_time <= Instant::now() {
                 self.check_for_retransmit().await;
                 self.start_retransmit_timer();
             }
@@ -386,18 +388,23 @@ impl AtemSocket {
     }
 
     async fn recieved_packet(&mut self, packet: &[u8]) {
-        let Ok(atem_packet): Result<AtemPacket, _> = packet.try_into() else {
+        let Ok(atem_packet): Result<AtemPacket<_>, _> = packet.try_into() else {
             return;
         };
 
-        log::debug!("Received {:x?}", atem_packet);
+        log::debug!("Received {}", atem_packet,);
+        log::debug!(
+            "fields: {}",
+            atem_packet.raw_fields().map(|f| f.r#type).join(",")
+        );
 
-        self.last_received_at = SystemTime::now();
+        self.last_received_at = Instant::now();
 
         self.session_id = atem_packet.session_id();
-        let remote_packet_id = atem_packet.remote_packet_id();
+        // TODO: naming seems rather off here
+        let remote_packet_id = atem_packet.local_sequence_number();
 
-        if atem_packet.has_flag(PacketFlag::NewSessionId) {
+        if atem_packet.flags().contains(PacketFlag::NewSessionId) {
             log::debug!("New session");
             self.connection_state = ConnectionState::Established;
             self.last_received_packed_id = remote_packet_id;
@@ -413,14 +420,12 @@ impl AtemSocket {
                 self.retransmit_from(from_packet_id).await;
             }
 
-            if atem_packet.has_flag(PacketFlag::AckRequest) {
+            if atem_packet.flags().contains(PacketFlag::AckRequest) {
                 if remote_packet_id == (self.last_received_packed_id + 1) % MAX_PACKET_ID {
                     self.last_received_packed_id = remote_packet_id;
                     self.send_or_queue_ack().await;
 
-                    if let Some(body) = atem_packet.body() {
+                    self.on_commands_received(atem_packet.body());
-                        self.on_commands_received(body);
-                    }
                 } else if self
                     .is_packet_covered_by_ack(self.last_received_packed_id, remote_packet_id)
                 {
@@ -428,7 +433,7 @@ impl AtemSocket {
                 }
             }
 
-            if atem_packet.has_flag(PacketFlag::IsRetransmit) {
+            if atem_packet.flags().contains(PacketFlag::IsRetransmit) {
                 log::debug!("ATEM retransmitted packet {:x?}", remote_packet_id);
             }
 
@@ -472,13 +477,13 @@ impl AtemSocket {
             self.ack_timer = None;
             self.send_ack(self.last_received_packed_id).await;
         } else if self.ack_timer.is_none() {
-            self.ack_timer = Some(SystemTime::now() + Duration::from_millis(5));
+            self.ack_timer = Some(Instant::now() + Duration::from_millis(5));
         }
     }
 
     async fn send_ack(&mut self, packet_id: u16) {
         log::debug!("Sending ack for packet {:x?}", packet_id);
-        let flag: u8 = PacketFlag::AckReply.into();
+        let flag: u8 = PacketFlag::AckReply as u8;
         let opcode = u16::from(flag) << 11;
         let mut buffer: [u8; ACK_PACKET_LENGTH as _] = [0; 12];
         buffer[0..2].copy_from_slice(&u16::to_be_bytes(opcode | ACK_PACKET_LENGTH));
@@ -505,7 +510,7 @@ impl AtemSocket {
                 if sent_packet.packet_id == from_id
                     || !self.is_packet_covered_by_ack(from_id, sent_packet.packet_id)
                 {
-                    sent_packet.last_sent = SystemTime::now();
+                    sent_packet.last_sent = Instant::now();
                     sent_packet.resent += 1;
 
                     self.send_packet(&sent_packet.payload).await;
@@ -519,8 +524,7 @@ impl AtemSocket {
 
     async fn check_for_retransmit(&mut self) {
         for sent_packet in self.in_flight.clone() {
-            if sent_packet.last_sent + Duration::from_millis(IN_FLIGHT_TIMEOUT) < SystemTime::now()
+            if sent_packet.last_sent + IN_FLIGHT_TIMEOUT < Instant::now() {
-            {
                 if sent_packet.resent <= MAX_PACKET_RETRIES
                     && self
                         .is_packet_covered_by_ack(self.next_send_packet_id, sent_packet.packet_id)
@@ -537,9 +541,11 @@ impl AtemSocket {
     }
 
     fn on_commands_received(&mut self, payload: &[u8]) {
-        let _ = self
+        if !payload.is_empty() {
-            .atem_event_tx
+            let _ = self
-            .send(AtemSocketEvent::ReceivedCommands(payload.to_vec()));
+                .atem_event_tx
+                .send(AtemSocketEvent::ReceivedCommands(payload.to_vec()));
+        }
     }
 
     fn on_command_acknowledged(&mut self, packets: Vec<AckedPacket>) {
@@ -574,13 +580,11 @@ impl AtemSocket {
     }
 
     fn start_reconnect_timer(&mut self) {
-        self.reconnect_timer =
+        self.reconnect_timer = Some(Instant::now() + CONNECTION_RETRY_INTERVAL);
-            Some(SystemTime::now() + Duration::from_millis(CONNECTION_RETRY_INTERVAL));
     }
 
     fn start_retransmit_timer(&mut self) {
-        self.retransmit_timer =
+        self.retransmit_timer = Some(Instant::now() + RETRANSMIT_CHECK_INTERVAL);
-            Some(SystemTime::now() + Duration::from_millis(RETRANSMIT_CHECK_INTERVAL));
     }
 
     fn next_packet_tracking_id(&mut self) -> u64 {

atem-connection-rs/src/atem_lib/atem_util.rs

@@ -1,4 +0,0 @@
-pub const COMMAND_CONNECT_HELLO: [u8; 20] = [
-    0x10, 0x14, 0x53, 0xab, 0x00, 0x00, 0x00, 0x00, 0x00, 0x3a, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00,
-    0x00, 0x00, 0x00, 0x00,
-];

atem-connection-rs/src/atem_lib/mod.rs

@@ -1,3 +1,4 @@
-mod atem_packet;
+pub mod atem_field;
+pub mod atem_packet;
+#[cfg(feature = "std")]
 pub mod atem_socket;
-pub mod atem_util;

atem-connection-rs/src/commands/command_base.rs

@@ -1,4 +1,7 @@
-use std::{collections::HashMap, fmt::Debug, process::Command, sync::Arc};
+use std::{
+    boxed::Box, collections::HashMap, fmt::Debug, process::Command, string::String, sync::Arc,
+    vec::Vec,
+};
 
 use crate::{enums::ProtocolVersion, state::AtemState};
 
@@ -9,7 +12,7 @@ pub trait DeserializedCommand: Send + Sync + Debug {
 
 pub trait CommandDeserializer: Send + Sync {
     fn deserialize(&self, buffer: &[u8], version: &ProtocolVersion)
-        -> Arc<dyn DeserializedCommand>;
+    -> Arc<dyn DeserializedCommand>;
 }
 
 pub trait SerializableCommand: Send + Sync {

atem-connection-rs/src/commands/device_profile/product_identifier.rs

@@ -1,4 +1,9 @@
-use std::{ffi::CString, sync::Arc};
+use std::{
+    ffi::CString,
+    string::{String, ToString},
+    sync::Arc,
+    vec,
+};
 
 use crate::{
     commands::command_base::{CommandDeserializer, DeserializedCommand},

atem-connection-rs/src/commands/device_profile/topology.rs

@@ -1,4 +1,4 @@
-use std::sync::Arc;
+use std::{sync::Arc, vec};
 
 use crate::{
     commands::command_base::{CommandDeserializer, DeserializedCommand},

atem-connection-rs/src/commands/mix_effects/program_input.rs

@@ -1,4 +1,5 @@
-use std::sync::Arc;
+use derive_new::new;
+use std::{sync::Arc, vec, vec::Vec};
 
 use crate::{
     commands::command_base::{

atem-connection-rs/src/commands/parse_commands.rs

@@ -1,7 +1,8 @@
-use std::{collections::VecDeque, sync::Arc};
+use std::{boxed::Box, collections::VecDeque, sync::Arc};
 
 use crate::{
-    commands::device_profile::version::{deserialize_version, DESERIALIZE_VERSION_RAW_NAME},
+    atem_lib::atem_packet::{AtemPacket, RawFields},
+    commands::device_profile::version::{DESERIALIZE_VERSION_RAW_NAME, deserialize_version},
     enums::ProtocolVersion,
 };
 
@@ -9,59 +10,44 @@ use super::{
     command_base::{CommandDeserializer, DeserializedCommand},
     device_profile::{
         audio_mixer_config::{AudioMixerConfigDeserializer, DESERIALIZE_AUDIO_MIXER_CONFIG_NAME},
-        media_pool_config::{MediaPoolConfigDeserializer, DESERIALIZE_MEDIA_POOL_CONFIG_NAME},
+        media_pool_config::{DESERIALIZE_MEDIA_POOL_CONFIG_NAME, MediaPoolConfigDeserializer},
         mix_effect_block_config::{
-            MixEffectBlockConfigDeserializer, DESERIALIZE_MIX_EFFECT_BLOCK_CONFIG_NAME,
+            DESERIALIZE_MIX_EFFECT_BLOCK_CONFIG_NAME, MixEffectBlockConfigDeserializer,
         },
-        multiviewer_config::{MultiviewerConfigDeserializer, DESERIALIZE_MULTIVIEWER_NAME},
+        multiviewer_config::{DESERIALIZE_MULTIVIEWER_NAME, MultiviewerConfigDeserializer},
         product_identifier::{
-            ProductIdentifierDeserializer, DESERIALIZE_PRODUCT_IDENTIFIER_RAW_NAME,
+            DESERIALIZE_PRODUCT_IDENTIFIER_RAW_NAME, ProductIdentifierDeserializer,
         },
-        topology::{TopologyDeserializer, DESERIALIZE_TOPOLOGY_RAW_NAME},
+        topology::{DESERIALIZE_TOPOLOGY_RAW_NAME, TopologyDeserializer},
     },
-    init_complete::{InitCompleteDeserializer, DESERIALIZE_INIT_COMPLETE_RAW_NAME},
+    init_complete::{DESERIALIZE_INIT_COMPLETE_RAW_NAME, InitCompleteDeserializer},
-    mix_effects::program_input::{ProgramInputDeserializer, DESERIALIZE_PROGRAM_INPUT_RAW_NAME},
+    mix_effects::program_input::{DESERIALIZE_PROGRAM_INPUT_RAW_NAME, ProgramInputDeserializer},
-    tally_by_source::{TallyBySourceDeserializer, DESERIALIZE_TALLY_BY_SOURCE_RAW_NAME},
+    tally_by_source::{DESERIALIZE_TALLY_BY_SOURCE_RAW_NAME, TallyBySourceDeserializer},
-    time::{TimeDeserializer, DESERIALIZE_TIME_RAW_NAME},
+    time::{DESERIALIZE_TIME_RAW_NAME, TimeDeserializer},
 };
 
-pub fn deserialize_commands(
+pub fn deserialize_commands<T: AsRef<[u8]>>(
-    payload: &[u8],
+    payload: T,
     version: &mut ProtocolVersion,
 ) -> VecDeque<Arc<dyn DeserializedCommand>> {
     let mut parsed_commands: VecDeque<Arc<dyn DeserializedCommand>> = VecDeque::new();
-    let mut head = 0;
 
-    while payload.len() > head + 8 {
+    for field in RawFields::new(payload.as_ref()) {
-        let length = u16::from_be_bytes([payload[head], payload[head + 1]]) as usize;
+        let name: &str = field.r#type.try_into().unwrap();
-        let Ok(name) = String::from_utf8(payload[(head + 4)..(head + 8)].to_vec()) else {
+        log::debug!("Received command {} with length {}", name, field.data.len(),);
-            break;
-        };
 
-        if length < 8 {
+        if field.r#type == DESERIALIZE_VERSION_RAW_NAME {
-            break;
+            let version_command = deserialize_version(field.data);
-        }
-
-        log::debug!("Received command {} with length {}", name, length);
-
-        let command_buffer = &payload[head + 8..head + length];
-
-        if name == DESERIALIZE_VERSION_RAW_NAME {
-            let version_command = deserialize_version(command_buffer);
             *version = version_command.version.clone();
             log::info!("Switched to protocol version {}", version);
             parsed_commands.push_back(Arc::new(version_command));
-        } else if let Some(deserializer) = command_deserializer_from_string(name.as_str()) {
+        } else if let Some(deserializer) = command_deserializer_from_string(name) {
-            let deserialized_command = deserializer.deserialize(command_buffer, version);
+            let deserialized_command = deserializer.deserialize(field.data, version);
             log::debug!("Received {:?}", deserialized_command);
             parsed_commands.push_back(deserialized_command);
         } else {
             log::warn!("Received command {name} for which there is no deserializer.");
-            // TODO: Remove!
-            todo!("Write deserializer for {name}.");
         }
-
-        head += length;
     }
 
     parsed_commands

atem-connection-rs/src/commands/tally_by_source.rs

@@ -1,3 +1,4 @@
+use derive_new::new;
 use std::{collections::HashMap, sync::Arc};
 
 use crate::enums::ProtocolVersion;

atem-connection-rs/src/lib.rs

@@ -1,11 +1,16 @@
-#[macro_use]
+#![no_std]
-extern crate derive_new;
-#[macro_use]
-extern crate derive_getters;
 
+#[cfg(feature = "std")]
+extern crate std;
+
+#[cfg(feature = "std")]
 pub mod atem;
 pub mod atem_lib;
+#[cfg(feature = "std")]
 pub mod commands;
+#[cfg(feature = "std")]
 pub mod enums;
+#[cfg(feature = "std")]
 pub mod state;
+#[cfg(feature = "std")]
 pub mod tally;

atem-connection-rs/src/state/atem_macro.rs

@@ -1,3 +1,7 @@
+use std::{string::String, vec::Vec};
+
+use derive_getters::Getters;
+use derive_new::new;
 #[derive(Clone, PartialEq, Getters, new, Default)]
 pub struct MacroPlayerState {
     pub is_running: bool,

atem-connection-rs/src/state/audio.rs

@@ -1,6 +1,8 @@
 use std::collections::HashMap;
 
 use crate::enums::{AudioMixOption, AudioSourceType, ExternalPortType};
+use derive_getters::Getters;
+use derive_new::new;
 
 pub type AudioChannel = ClassicAudioChannel;
 pub type AudioMasterChannel = ClassicAudioMasterChannel;

atem-connection-rs/src/state/color.rs

@@ -1,3 +1,5 @@
+use derive_getters::Getters;
+use derive_new::new;
 #[derive(Clone, PartialEq, Getters, new)]
 pub struct ColorGeneratorState {
     pub hue: u64,

atem-connection-rs/src/state/common.rs

@@ -1,3 +1,5 @@
+use derive_getters::Getters;
+use derive_new::new;
 #[derive(Clone, PartialEq, Getters, new)]
 pub struct Timecode {
     pub hours: u64,

atem-connection-rs/src/state/fairlight.rs

@@ -1,4 +1,6 @@
-use std::collections::HashMap;
+use derive_getters::Getters;
+use derive_new::new;
+use std::{collections::HashMap, string::String, vec::Vec};
 
 use crate::enums::{
     ExternalPortType, FairlightAnalogInputLevel, FairlightAudioMixOption, FairlightAudioSourceType,

atem-connection-rs/src/state/info.rs

@@ -1,4 +1,8 @@
+use std::{string::String, vec::Vec};
+
 use crate::enums::{Model, ProtocolVersion};
+use derive_getters::Getters;
+use derive_new::new;
 
 #[derive(Clone, PartialEq, Getters, new)]
 pub struct AtemCapabilites {

atem-connection-rs/src/state/input.rs

@@ -1,4 +1,8 @@
 use crate::enums::{ExternalPortType, InternalPortType, MeAvailability, SourceAvailability};
+use derive_getters::Getters;
+use derive_new::new;
+use std::string::String;
+use std::vec::Vec;
 
 #[derive(Clone, PartialEq, Getters, new)]
 pub struct InputChannel {

atem-connection-rs/src/state/media.rs

@@ -1,4 +1,8 @@
 use crate::enums;
+use derive_getters::Getters;
+use derive_new::new;
+use std::string::String;
+use std::vec::Vec;
 
 #[derive(Clone, PartialEq, Getters, new)]
 pub struct MediaPlayer {

atem-connection-rs/src/state/recording.rs

@@ -1,4 +1,7 @@
+use derive_getters::Getters;
+use derive_new::new;
 use std::collections::HashMap;
+use std::string::String;
 
 use crate::enums::{RecordingDiskStatus, RecordingError, RecordingStatus};
 

atem-connection-rs/src/state/settings.rs

@@ -1,4 +1,7 @@
 use crate::enums::{MultiViewerLayout, VideoMode};
+use derive_getters::Getters;
+use derive_new::new;
+use std::vec::Vec;
 
 pub trait MultiViewerSourceState {
     fn get_source(&self) -> u64;

atem-connection-rs/src/state/streaming.rs

@@ -1,4 +1,7 @@
 use crate::enums::{StreamingError, StreamingStatus};
+use derive_getters::Getters;
+use derive_new::new;
+use std::string::String;
 
 use super::common::Timecode;
 

atem-connection-rs/src/state/util.rs

@@ -1,9 +1,10 @@
 use crate::enums::{TransitionSelection, TransitionStyle};
+use std::vec;
 
 use super::{
+    AtemState,
     settings::MultiViewer,
     video::{MixEffect, TransitionPosition, TransitionProperties, TransitionSettings},
-    AtemState,
 };
 
 pub fn create() -> AtemState {

atem-connection-rs/src/state/video/downstream_keyers.rs

@@ -1,3 +1,5 @@
+use derive_getters::Getters;
+use derive_new::new;
 pub trait DownstreamKeyerBase {
     fn get_in_transition(&self) -> bool;
     fn get_remaining_frames(&self) -> f64;

atem-connection-rs/src/state/video/mod.rs

@@ -1,4 +1,7 @@
 use crate::enums;
+use derive_getters::Getters;
+use derive_new::new;
+use std::vec::Vec;
 
 mod downstream_keyers;
 mod super_source;

atem-connection-rs/src/state/video/super_source.rs

@@ -1,4 +1,6 @@
 use crate::enums;
+use derive_getters::Getters;
+use derive_new::new;
 
 #[derive(Clone, PartialEq, Getters, new)]
 pub struct SuperSourceBox {

atem-connection-rs/src/tally.rs

@@ -1,3 +1,4 @@
+use std::string::String;
 #[derive(Debug)]
 pub struct TallyEvent {
     tally_state: TallyState,

atem-test/Cargo.toml

@@ -1,9 +1,7 @@
 [package]
 name = "atem-test"
 version = "0.1.0"
-edition = "2021"
+edition = "2024"
-
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
 atem-connection-rs = { path = "../atem-connection-rs" }

atem-test/src/main.rs

@@ -69,12 +69,16 @@ async fn main() {
 
 fn setup_logging() -> Result<(), Report> {
     if std::env::var("RUST_LIB_BACKTRACE").is_err() {
-        std::env::set_var("RUST_LIB_BACKTRACE", "1");
+        unsafe {
+            std::env::set_var("RUST_LIB_BACKTRACE", "1");
+        }
     }
     color_eyre::install()?;
 
     if std::env::var("RUST_LOG").is_err() {
-        std::env::set_var("RUST_LOG", "debug");
+        unsafe {
+            std::env::set_var("RUST_LOG", "debug");
+        }
     }
     env_logger::init();
 

flake.lock

@@ -1,26 +1,8 @@
 {
   "nodes": {
-    "devshell": {
-      "inputs": {
-        "nixpkgs": "nixpkgs"
-      },
-      "locked": {
-        "lastModified": 1741473158,
-        "narHash": "sha256-kWNaq6wQUbUMlPgw8Y+9/9wP0F8SHkjy24/mN3UAppg=",
-        "owner": "numtide",
-        "repo": "devshell",
-        "rev": "7c9e793ebe66bcba8292989a68c0419b737a22a0",
-        "type": "github"
-      },
-      "original": {
-        "owner": "numtide",
-        "repo": "devshell",
-        "type": "github"
-      }
-    },
     "naersk": {
       "inputs": {
-        "nixpkgs": "nixpkgs_2"
+        "nixpkgs": "nixpkgs"
       },
       "locked": {
         "lastModified": 1745925850,
@@ -38,11 +20,11 @@
     },
     "nixpkgs": {
       "locked": {
-        "lastModified": 1722073938,
+        "lastModified": 1749619289,
-        "narHash": "sha256-OpX0StkL8vpXyWOGUD6G+MA26wAXK6SpT94kLJXo6B4=",
+        "narHash": "sha256-qX6gXVjaCXXbcn6A9eSLUf8Fm07MgPGe5ir3++y2O1Q=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "e36e9f57337d0ff0cf77aceb58af4c805472bfae",
+        "rev": "f72be405a10668b8b00937b452f2145244103ebc",
         "type": "github"
       },
       "original": {
@@ -54,25 +36,10 @@
     },
     "nixpkgs_2": {
       "locked": {
-        "lastModified": 1749401433,
+        "lastModified": 1733412085,
-        "narHash": "sha256-HXIQzULIG/MEUW2Q/Ss47oE3QrjxvpUX7gUl4Xp6lnc=",
+        "narHash": "sha256-FillH0qdWDt/nlO6ED7h4cmN+G9uXwGjwmCnHs0QVYM=",
-        "owner": "NixOS",
+        "path": "/nix/store/5wvrlpfrwlmg24ywkybvidyzcvki6bwx-source",
-        "repo": "nixpkgs",
+        "rev": "4dc2fc4e62dbf62b84132fe526356fbac7b03541",
-        "rev": "08fcb0dcb59df0344652b38ea6326a2d8271baff",
-        "type": "github"
-      },
-      "original": {
-        "owner": "NixOS",
-        "ref": "nixpkgs-unstable",
-        "repo": "nixpkgs",
-        "type": "github"
-      }
-    },
-    "nixpkgs_3": {
-      "locked": {
-        "lastModified": 0,
-        "narHash": "sha256-DDe16FJk18sadknQKKG/9FbwEro7A57tg9vB5kxZ8kY=",
-        "path": "/nix/store/2d1ahim48jhzg4bbm97mvjlb4p7fpan3-source",
         "type": "path"
       },
       "original": {
@@ -80,7 +47,7 @@
         "type": "indirect"
       }
     },
-    "nixpkgs_4": {
+    "nixpkgs_3": {
       "locked": {
         "lastModified": 1744536153,
         "narHash": "sha256-awS2zRgF4uTwrOKwwiJcByDzDOdo3Q1rPZbiHQg/N38=",
@@ -98,23 +65,22 @@
     },
     "root": {
       "inputs": {
-        "devshell": "devshell",
         "naersk": "naersk",
-        "nixpkgs": "nixpkgs_3",
+        "nixpkgs": "nixpkgs_2",
         "rust-overlay": "rust-overlay",
         "utils": "utils"
       }
     },
     "rust-overlay": {
       "inputs": {
-        "nixpkgs": "nixpkgs_4"
+        "nixpkgs": "nixpkgs_3"
       },
       "locked": {
-        "lastModified": 1749436897,
+        "lastModified": 1749695868,
-        "narHash": "sha256-OkDtaCGQQVwVFz5HWfbmrMJR99sFIMXHCHEYXzUJEJY=",
+        "narHash": "sha256-debjTLOyqqsYOUuUGQsAHskFXH5+Kx2t3dOo/FCoNRA=",
         "owner": "oxalica",
         "repo": "rust-overlay",
-        "rev": "e7876c387e35dc834838aff254d8e74cf5bd4f19",
+        "rev": "55f914d5228b5c8120e9e0f9698ed5b7214d09cd",
         "type": "github"
       },
       "original": {

flake.nix

@@ -3,7 +3,6 @@
 
   inputs = {
     utils.url = "github:numtide/flake-utils";
-    devshell.url = "github:numtide/devshell";
     naersk.url = "github:nix-community/naersk";
     rust-overlay.url = "github:oxalica/rust-overlay";
   };
@@ -13,7 +12,6 @@
     nixpkgs,
     utils,
     naersk,
-    devshell,
     rust-overlay,
   }:
     utils.lib.eachDefaultSystem (system: let
@@ -35,18 +33,10 @@
 
       apps.default = utils.lib.mkApp {drv = packages.default;};
 
-      # Provide a dev env with rust and rust-analyzer
+      devShells.default = pkgs.mkShell {
-      devShells.default = let
-        pkgs = import nixpkgs {
-          inherit system;
-          overlays = [devshell.overlays.default];
-        };
-      in
-        pkgs.devshell.mkShell {
-          motd = "Hello you wonderful person, I hope you are having a lovely day 💜";
-
           packages = with pkgs; [(rust.override {extensions = ["rust-src"];}) rust-analyzer gcc];
+
         };
-      formatter = pkgs.alejandra;
+      formatter = pkgs.nixfmt-rfc-style;
     });
 }