J2534 Support (#101)

* Add J2534 support

* fix comment style, do not set stmin_tx ioctl unless requested

* run cargo fmt

* dont use 5 year old winreg version

* remove some stray comments

* move constants to separate file with proper types

* move dll magic into own file, might reuse later

* Simplify J2534CanAdapter, make fully blocking

* loopback flag already set

* some more refactoring of common stuff into J2534Channel struct

* return disconnected error

* remove comment

* move stmin helper into isotp code, ensure rounding up

* refactor J2534NativeIsoTpTransport to take an IsoTPConfig

* parse more fields out of the IsoTPConfig

* unified error handling

* add isotp example

* attempt to fix 29 bit ids

* handle isotp rx timeouts

* this should fix RX of extended IDs

* Turns those logs into errors

* Those don't result in valid rx ids

* Improve query_fw_versions for adapters without proper ACK support

* assert on invalid 11 bit id

* remove testing change

---------

Co-authored-by: Willem Melching <willem.melching@gmail.com>

Author: bri3d

Cargo.lock

@@ -13,11 +13,12 @@ repository = "https://github.com/I-CAN-hack/automotive"
 default = ["default-adapters"]
 all = ["all-adapters", "serde"]
 default-adapters = ["panda", "socketcan"]
-all-adapters = ["default-adapters", "vector-xl"]
+all-adapters = ["default-adapters", "vector-xl", "j2534"]
 serde = ["dep:serde"]
 
 # adapters
 vector-xl = ["dep:bindgen", "dep:libloading"]
+j2534 = ["dep:libloading", "dep:winreg"]
 panda = []
 socketcan = []
 
@@ -46,6 +47,7 @@ libc = "0.2.154"
 
 [target.'cfg(target_os = "windows")'.dependencies]
 libloading = { version = "0.8", optional = true }
+winreg = { version = "0.55", optional = true }
 
 [dev-dependencies]
 futures = "0.3.30"

Cargo.toml

@@ -1,5 +1,7 @@
+use std::time::Duration;
+
 use automotive::can::AsyncCanAdapter;
-use automotive::isotp::IsoTPAdapter;
+use automotive::isotp::{IsoTPAdapter, IsoTPConfig};
 use automotive::Result;
 
 use automotive::uds::DataIdentifier;
@@ -9,7 +11,13 @@ use bstr::ByteSlice;
 use strum::IntoEnumIterator;
 
 async fn get_version(adapter: &AsyncCanAdapter, identifier: u32) -> Result<()> {
-    let isotp = IsoTPAdapter::from_id(adapter, identifier);
+    let mut config = IsoTPConfig::new(0, identifier.into());
+
+    // Increased timeout for adapters not supporting real ACKs
+    // We send a lot of frames, so the timeout might start counting before the relevant frame is sent
+    config.timeout = Duration::from_secs(1);
+
+    let isotp = IsoTPAdapter::new(adapter, config);
     let uds = UDSClient::new(&isotp);
 
     uds.tester_present().await?;
@@ -35,7 +43,7 @@ async fn main() {
 
     let adapter = automotive::can::get_adapter().unwrap();
 
-    let standard_ids = 0x700..=0x7ff;
+    let standard_ids = 0x700..=0x7f7;
     let extended_ids = (0x00..=0xff).map(|i| 0x18da0000 + (i << 8) + 0xf1);
 
     let ids: Vec<u32> = standard_ids.chain(extended_ids).collect();

examples/query_fw_versions.rs

@@ -0,0 +1,74 @@
+#[cfg(all(target_os = "windows", feature = "j2534"))]
+mod app {
+    use automotive::isotp::IsoTPConfig;
+    use automotive::j2534::J2534NativeIsoTpTransport;
+    use automotive::uds::{DataIdentifier, UDSClient};
+    use automotive::Result;
+
+    use bstr::ByteSlice;
+    use strum::IntoEnumIterator;
+
+    async fn get_version(isotp: &J2534NativeIsoTpTransport, identifier: u32) -> Result<()> {
+        let uds = UDSClient::new(isotp);
+
+        uds.tester_present().await?;
+
+        for did in DataIdentifier::iter() {
+            if let Ok(resp) = uds.read_data_by_identifier(did as u16).await {
+                println!(
+                    "{:x} 0x{:x} {:?}: {:?}",
+                    identifier,
+                    did as u16,
+                    did,
+                    resp.as_bstr()
+                );
+            }
+        }
+
+        Ok(())
+    }
+
+    pub async fn run() -> Result<()> {
+        tracing_subscriber::fmt::init();
+
+        let dll_path = None;
+        let bitrate = 500_000;
+
+        let standard_ids = 0x700..=0x7f7;
+        let extended_ids = (0x00..=0xff).map(|i| 0x18da0000 + (i << 8) + 0xf1);
+        let mut ids = standard_ids.chain(extended_ids);
+
+        let Some(first_id) = ids.next() else {
+            return Ok(());
+        };
+
+        let config = IsoTPConfig::new(0, first_id.into());
+        let isotp = J2534NativeIsoTpTransport::new(dll_path, bitrate, config)?;
+        let _ = get_version(&isotp, first_id).await;
+        let mut device = isotp.into_device();
+
+        for id in ids {
+            let config = IsoTPConfig::new(0, id.into());
+            let isotp = match J2534NativeIsoTpTransport::new_on_device(device, bitrate, config) {
+                Ok(isotp) => isotp,
+                Err(_) => J2534NativeIsoTpTransport::new(dll_path, bitrate, config)?,
+            };
+
+            let _ = get_version(&isotp, id).await;
+            device = isotp.into_device();
+        }
+
+        Ok(())
+    }
+}
+
+#[cfg(all(target_os = "windows", feature = "j2534"))]
+#[tokio::main]
+async fn main() -> automotive::Result<()> {
+    app::run().await
+}
+
+#[cfg(not(all(target_os = "windows", feature = "j2534")))]
+fn main() {
+    eprintln!("This example requires Windows and the `j2534` feature.");
+}

examples/query_fw_versions_j2534.rs

@@ -28,5 +28,12 @@ pub fn get_adapter() -> Result<crate::can::AsyncCanAdapter, crate::error::Error>
         };
     }
 
+    #[cfg(all(target_os = "windows", feature = "j2534"))]
+    {
+        if let Ok(adapter) = crate::j2534::J2534CanAdapter::new_async(None, 500_000) {
+            return Ok(adapter);
+        };
+    }
+
     Err(crate::error::Error::NotFound)
 }

src/can/adapter.rs

@@ -12,7 +12,12 @@ use tokio::sync::{broadcast, mpsc, oneshot};
 use tracing::debug;
 
 const CAN_TX_BUFFER_SIZE: usize = 128;
-const CAN_RX_BUFFER_SIZE: usize = 1024;
+// Must be large enough to absorb all CAN frames generated during a
+// multi-frame ISO-TP transfer without the recv subscriber being polled.
+// A 4 KiB UDS payload at 7 bytes/CF ≈ 585 CFs; each produces a loopback
+// echo plus the real frame.  Back-to-back transfers or mixed TX/RX traffic
+// can double that.  8192 gives ample headroom.
+const CAN_RX_BUFFER_SIZE: usize = 8192;
 const DEBUG: bool = false;
 
 type BusIdentifier = (u8, Identifier);
@@ -28,7 +33,13 @@ fn process<T: CanAdapter>(
     let mut callbacks: HashMap<BusIdentifier, VecDeque<FrameCallback>> = HashMap::new();
 
     while shutdown_receiver.try_recv().is_err() {
-        let frames: Vec<Frame> = adapter.recv().expect("Failed to Receive CAN Frames");
+        let frames: Vec<Frame> = match adapter.recv() {
+            Ok(f) => f,
+            Err(e) => {
+                debug!("Adapter recv error: {:?} — shutting down process loop", e);
+                break;
+            }
+        };
 
         for frame in frames {
             if DEBUG {
@@ -98,7 +109,7 @@ pub struct AsyncCanAdapter {
 }
 
 impl AsyncCanAdapter {
-    pub fn new<T: CanAdapter + Send + Sync + 'static>(adapter: T) -> Self {
+    pub fn new<T: CanAdapter + Send + 'static>(adapter: T) -> Self {
         let (shutdown_sender, shutdown_receiver) = oneshot::channel();
         let (send_sender, send_receiver) = mpsc::channel(CAN_TX_BUFFER_SIZE);
         let (recv_sender, recv_receiver) = broadcast::channel(CAN_RX_BUFFER_SIZE);
@@ -145,7 +156,10 @@ impl AsyncCanAdapter {
                             yield frame
                         }
                     },
-                    Err(RecvError::Closed) => panic!("Adapter thread has exited"),
+                    Err(RecvError::Closed) => {
+                        tracing::debug!("Adapter broadcast closed — ending recv stream");
+                        return;
+                    },
                     Err(RecvError::Lagged(n)) => {
                         tracing::warn!("Receive too slow, dropping {} frame(s).", n)
                     },
@@ -158,9 +172,15 @@ impl AsyncCanAdapter {
 impl Drop for AsyncCanAdapter {
     fn drop(&mut self) {
         if let Some(handle) = self.processing_handle.take() {
-            // Send shutdown signal to background tread
-            self.shutdown.take().unwrap().send(()).unwrap();
-            handle.join().unwrap();
+            // Send shutdown signal to background thread.
+            // Use `ok()` instead of `unwrap()` because the receiver may already
+            // be dropped if the process thread exited early (e.g. adapter error).
+            if let Some(shutdown) = self.shutdown.take() {
+                let _ = shutdown.send(());
+            }
+            // Join the thread; use `ok()` to avoid panicking inside Drop if the
+            // process thread panicked (double-panic would abort the process).
+            let _ = handle.join();
         }
     }
 }

src/can/async_can.rs

@@ -31,6 +31,10 @@ pub enum Error {
     #[error("Error loading DLL: {0}")]
     Libloading(std::sync::Arc<libloading::Error>),
 
+    #[cfg(all(target_os = "windows", feature = "j2534"))]
+    #[error(transparent)]
+    J2534Error(#[from] crate::j2534::Error),
+
     #[cfg(feature = "panda")]
     #[error(transparent)]
     PandaError(#[from] crate::panda::Error),

src/error.rs

@@ -30,6 +30,38 @@ use tracing::debug;
 use self::types::FlowControlConfig;
 
 const DEFAULT_TIMEOUT_MS: u64 = 100;
+
+/// Abstraction over anything that can exchange ISO-TP PDUs.
+///
+/// Two implementations ship with this workspace:
+/// * [`IsoTPAdapter`] — software ISO-TP framing on top of any [`crate::can::CanAdapter`].
+/// * `J2534NativeIsoTpTransport` — hardware ISO 15765 via a J2534 PassThru device; the
+///   adapter firmware handles all framing, flow-control, and STmin timing.
+pub trait IsoTpTransport {
+    /// Transmit a single UDS PDU. Resolves once the transport has accepted the
+    /// payload for transmission (not necessarily once it has been ACKed on the bus).
+    fn send<'a>(
+        &'a self,
+        data: &'a [u8],
+    ) -> impl std::future::Future<Output = crate::Result<()>> + 'a;
+
+    /// Infinite stream of received UDS PDUs. Each item is one complete PDU.
+    fn recv(&self) -> impl crate::Stream<Item = crate::Result<Vec<u8>>> + Unpin + '_;
+}
+
+impl IsoTpTransport for IsoTPAdapter<'_> {
+    fn send<'a>(
+        &'a self,
+        data: &'a [u8],
+    ) -> impl std::future::Future<Output = crate::Result<()>> + 'a {
+        IsoTPAdapter::send(self, data)
+    }
+
+    fn recv(&self) -> impl crate::Stream<Item = crate::Result<Vec<u8>>> + Unpin + '_ {
+        IsoTPAdapter::recv(self)
+    }
+}
+
 const DEFAULT_PADDING_BYTE: u8 = 0xAA;
 
 /// N_WFTmax in ISO 15765-2
@@ -39,7 +71,34 @@ const CAN_MAX_DLEN: usize = 8;
 const CAN_FD_MAX_DLEN: usize = 64;
 
 const ISO_TP_MAX_DLEN: usize = (1 << 12) - 1;
-const ISO_TP_FD_MAX_DLEN: usize = (1 << 32) - 1;
+const ISO_TP_FD_MAX_DLEN: usize = u32::MAX as usize;
+
+/// Encode a separation-time value to the ISO 15765-2 STmin byte.
+///
+/// Values that do not land exactly on a representable STmin step round up to
+/// the next valid encoding so the requested delay is never undershot.
+///
+/// Encoding (ISO 15765-2 §9.6.5.4 / Table 5):
+/// - 0 µs -> `0x00` (no delay)
+/// - 1-900 µs -> `0xF1`-`0xF9` (100 µs steps, rounded up)
+/// - 901-127_000 µs -> `0x01`-`0x7F` (1 ms steps, rounded up)
+/// - Values above 127 ms clamp to `0x7F`
+pub fn duration_to_stmin_byte(duration: std::time::Duration) -> u8 {
+    let us = duration.as_micros().min(u128::from(u32::MAX)) as u32;
+    if us == 0 {
+        0x00
+    } else if us < 1_000 {
+        let steps = us.saturating_add(99) / 100;
+        if steps <= 9 {
+            0xF0 + steps as u8
+        } else {
+            0x01
+        }
+    } else {
+        let ms = us.saturating_add(999) / 1_000;
+        ms.min(127) as u8
+    }
+}
 
 /// Configuring passed to the IsoTPAdapter.
 #[derive(Debug, Clone, Copy)]
@@ -537,3 +596,36 @@ impl<'a> IsoTPAdapter<'a> {
         })
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::duration_to_stmin_byte;
+    use std::time::Duration;
+
+    #[test]
+    fn duration_to_stmin_byte_rounds_up_to_supported_microsecond_steps() {
+        assert_eq!(duration_to_stmin_byte(Duration::ZERO), 0x00);
+        assert_eq!(duration_to_stmin_byte(Duration::from_micros(1)), 0xF1);
+        assert_eq!(duration_to_stmin_byte(Duration::from_micros(99)), 0xF1);
+        assert_eq!(duration_to_stmin_byte(Duration::from_micros(100)), 0xF1);
+        assert_eq!(duration_to_stmin_byte(Duration::from_micros(101)), 0xF2);
+        assert_eq!(duration_to_stmin_byte(Duration::from_micros(900)), 0xF9);
+        assert_eq!(duration_to_stmin_byte(Duration::from_micros(901)), 0x01);
+        assert_eq!(duration_to_stmin_byte(Duration::from_micros(999)), 0x01);
+    }
+
+    #[test]
+    fn duration_to_stmin_byte_rounds_up_to_supported_millisecond_steps() {
+        assert_eq!(duration_to_stmin_byte(Duration::from_micros(1_000)), 0x01);
+        assert_eq!(duration_to_stmin_byte(Duration::from_micros(1_001)), 0x02);
+        assert_eq!(duration_to_stmin_byte(Duration::from_micros(1_999)), 0x02);
+        assert_eq!(duration_to_stmin_byte(Duration::from_micros(2_000)), 0x02);
+    }
+
+    #[test]
+    fn duration_to_stmin_byte_clamps_to_maximum_supported_delay() {
+        assert_eq!(duration_to_stmin_byte(Duration::from_micros(127_000)), 0x7F);
+        assert_eq!(duration_to_stmin_byte(Duration::from_micros(127_001)), 0x7F);
+        assert_eq!(duration_to_stmin_byte(Duration::from_secs(u64::MAX)), 0x7F);
+    }
+}