handle socketcan network down

src/can/async_can.rs

@@ -28,7 +28,7 @@ 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().unwrap();
+ let frames: Vec<Frame> = adapter.recv().expect("Failed to Receive CAN Frames");
 
  for frame in frames {
  if DEBUG {

src/error.rs

@@ -12,6 +12,8 @@ pub enum Error {
  MalformedFrame,
  #[error("Timeout")]
  Timeout,
+ #[error("Disconnected")]
+ Disconnected,
  #[error(transparent)]
  IsoTPError(#[from] crate::isotp::Error),
  #[error(transparent)]

src/socketcan/mod.rs

@@ -100,10 +100,21 @@ impl CanAdapter for SocketCan {
  fn recv(&mut self) -> Result<Vec<Frame>> {
  let mut frames = vec![];
 
- while let Ok((frame, meta)) = self.socket.read_frame_with_meta() {
- let mut frame: crate::can::Frame = frame.into();
- frame.loopback = meta.loopback;
- frames.push(frame);
+ loop {
+ match self.socket.read_frame_with_meta() {
+ Ok((frame, meta)) => {
+ let mut frame: crate::can::Frame = frame.into();
+ frame.loopback = meta.loopback;
+ frames.push(frame);
+ }
+ Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
+ break;
+ }
+ Err(e) => {
+ tracing::error!("Error reading frame: {}", e);
+ return Err(crate::error::Error::Disconnected);
+ }
+ }
  }
 
  // Add fake loopback frames to the receive queue

tests/adapter_tests.rs

@@ -7,7 +7,8 @@ use std::time::Duration;
 
 static BULK_NUM_FRAMES_SYNC: u64 = 0x100;
 static BULK_NUM_FRAMES_ASYNC: u64 = 0x1000;
-static BULK_TIMEOUT_MS: u64 = 1000;
+static BULK_SYNC_TIMEOUT_MS: u64 = 1000;
+static BULK_ASYNC_TIMEOUT_MS: u64 = 5000;
 
 /// Sends a large number of frames to a "blocking" adapter, and then reads back all sent messages.
 /// This verified the adapter doesn't drop messages and reads them back in the same order as they are sent,
@@ -27,7 +28,8 @@ fn bulk_send_sync<T: CanAdapter>(adapter: &mut T) {
  let start = std::time::Instant::now();
 
  let mut received: Vec<Frame> = vec![];
- while received.len() < frames.len() && start.elapsed() < Duration::from_millis(BULK_TIMEOUT_MS)
+ while received.len() < frames.len()
+ && start.elapsed() < Duration::from_millis(BULK_SYNC_TIMEOUT_MS)
  {
  let rx = adapter.recv().unwrap();
  let rx: Vec<Frame> = rx.into_iter().filter(|frame| frame.loopback).collect();
@@ -55,7 +57,7 @@ async fn bulk_send(adapter: &AsyncCanAdapter) {
 
  let r = frames.iter().map(|frame| adapter.send(frame));
  tokio::time::timeout(
- Duration::from_millis(BULK_TIMEOUT_MS),
+ Duration::from_millis(BULK_ASYNC_TIMEOUT_MS),
  futures::future::join_all(r),
  )
  .await