zencan_common/
socketcan.rs

1use std::sync::Arc;
2
3use crate::{
4    messages::{CanError, CanId, CanMessage},
5    traits::{AsyncCanReceiver, AsyncCanSender, CanSendError},
6};
7use snafu::{ResultExt, Snafu};
8use socketcan::{CanFrame, CanSocket, EmbeddedFrame, Frame, ShouldRetry, Socket};
9use tokio::io::{unix::AsyncFd, Interest};
10
11fn socketcan_id_to_zencan_id(id: socketcan::CanId) -> CanId {
12    match id {
13        socketcan::CanId::Standard(id) => CanId::std(id.as_raw()),
14        socketcan::CanId::Extended(id) => CanId::extended(id.as_raw()),
15    }
16}
17
18fn zencan_id_to_socketcan_id(id: CanId) -> socketcan::CanId {
19    match id {
20        CanId::Extended(id) => socketcan::ExtendedId::new(id).unwrap().into(),
21        CanId::Std(id) => socketcan::StandardId::new(id).unwrap().into(),
22    }
23}
24
25fn socketcan_frame_to_zencan_message(frame: socketcan::CanFrame) -> Result<CanMessage, CanError> {
26    let id = socketcan_id_to_zencan_id(frame.can_id());
27
28    match frame {
29        CanFrame::Data(frame) => Ok(CanMessage::new(id, frame.data())),
30        CanFrame::Remote(_) => Ok(CanMessage::new_rtr(id)),
31        CanFrame::Error(frame) => Err(CanError::from_raw(frame.error_bits() as u8)),
32    }
33}
34
35fn zencan_message_to_socket_frame(frame: CanMessage) -> socketcan::CanFrame {
36    let id = zencan_id_to_socketcan_id(frame.id());
37
38    if frame.is_rtr() {
39        socketcan::CanFrame::new_remote(id, 0).unwrap()
40    } else {
41        socketcan::CanFrame::new(id, frame.data()).unwrap()
42    }
43}
44
45#[derive(Debug, Clone)]
46pub struct SocketCanReceiver {
47    socket: Arc<AsyncCanSocket>,
48}
49
50#[derive(Debug, Snafu)]
51pub enum ReceiveError {
52    Io { source: socketcan::IoError },
53    Can { source: CanError },
54}
55
56#[derive(Debug, Snafu)]
57pub struct SendError {
58    source: socketcan::IoError,
59    message: CanMessage,
60}
61
62impl CanSendError for SendError {
63    fn into_can_message(self) -> CanMessage {
64        self.message
65    }
66
67    fn message(&self) -> String {
68        self.source.to_string()
69    }
70}
71
72/// Create an Async socket around a socketcan CanSocket. This is just a reimplemenation of the tokio
73/// socket in the `socketcan` crate, but with support for `try_read_frame` and `try_write_frame`
74/// added.
75#[derive(Debug)]
76struct AsyncCanSocket(AsyncFd<CanSocket>);
77
78#[allow(dead_code)]
79impl AsyncCanSocket {
80    pub fn new(inner: CanSocket) -> Result<Self, std::io::Error> {
81        inner.set_nonblocking(true)?;
82        Ok(Self(AsyncFd::new(inner)?))
83    }
84
85    pub fn open(ifname: &str) -> Result<Self, std::io::Error> {
86        let socket = CanSocket::open(ifname)?;
87        socket.set_nonblocking(true)?;
88        Ok(Self(AsyncFd::new(socket)?))
89    }
90
91    /// Attempt to read a CAN frame from the socket without blocking
92    ///
93    /// If no message is immediately available, a WouldBlock error is returned.
94    pub fn try_read_frame(&self) -> Result<CanFrame, std::io::Error> {
95        self.0.get_ref().read_frame()
96    }
97
98    /// Read a CAN frame from the socket asynchronously
99    pub async fn read_frame(&self) -> Result<CanFrame, std::io::Error> {
100        self.0
101            .async_io(Interest::READABLE, |inner| inner.read_frame())
102            .await
103    }
104
105    pub async fn write_frame(&self, frame: &CanFrame) -> Result<(), std::io::Error> {
106        self.0
107            .async_io(Interest::WRITABLE, |inner| inner.write_frame(frame))
108            .await
109    }
110
111    /// Attempt to write a CAN frame to the socket without blocking
112    pub fn try_write_frame(&self, frame: CanFrame) -> Result<(), std::io::Error> {
113        self.0.get_ref().write_frame(&frame)
114    }
115}
116
117impl AsyncCanReceiver for SocketCanReceiver {
118    type Error = ReceiveError;
119
120    fn try_recv(&mut self) -> Option<CanMessage> {
121        match self.socket.try_read_frame() {
122            Ok(frame) => Some(socketcan_frame_to_zencan_message(frame).unwrap()),
123            _ => None,
124        }
125    }
126
127    async fn recv(&mut self) -> Result<CanMessage, ReceiveError> {
128        loop {
129            match self.socket.read_frame().await {
130                Ok(frame) => return socketcan_frame_to_zencan_message(frame).context(CanSnafu),
131                Err(e) => {
132                    if !e.should_retry() {
133                        return Err(ReceiveError::Io { source: e });
134                    }
135                }
136            }
137        }
138    }
139}
140
141#[derive(Debug, Clone)]
142pub struct SocketCanSender {
143    socket: Arc<AsyncCanSocket>,
144}
145
146impl AsyncCanSender for SocketCanSender {
147    type Error = SendError;
148    async fn send(&mut self, msg: CanMessage) -> Result<(), Self::Error> {
149        let socketcan_frame = zencan_message_to_socket_frame(msg);
150
151        self.socket
152            .write_frame(&socketcan_frame)
153            .await
154            .context(SendSnafu { message: msg })
155    }
156}
157
158/// Open a socketcan device and split it into a sender and receiver object for use with zencan
159/// library
160///
161/// # Arguments
162/// * `device` - The name of the socketcan device to open, e.g. "vcan0", or "can0"
163///
164/// A key benefit of this is that by creating both sender and receiver objects from a shared socket,
165/// the receiver will not receive messages sent by the sender.
166#[cfg_attr(docsrs, doc(cfg(feature = "socketcan")))]
167pub fn open_socketcan<S: AsRef<str>>(
168    device: S,
169) -> Result<(SocketCanSender, SocketCanReceiver), socketcan::IoError> {
170    let device: &str = device.as_ref();
171    let socket = Arc::new(AsyncCanSocket::open(device)?);
172    let receiver = SocketCanReceiver {
173        socket: socket.clone(),
174    };
175    let sender = SocketCanSender { socket };
176    Ok((sender, receiver))
177}
zencan_common/socketcan.rs

zencan_common/
socketcan.rs
