socketcan 3.6.0

//! # tokio-socketcan
//!
//! Connective plumbing between the socketcan crate
//! and the tokio asynchronous I/O system
//!
//! # Usage
//!
//! The [socketcan](https://docs.rs/socketcan/1.7.0/socketcan/)
//! crate's documentation is valuable as the api used by
//! tokio-socketcan is largely identical to the socketcan one.
//!
//! An example echo server:
//!
//! ```no_run
//! use futures_util::stream::StreamExt;
//! use socketcan::{Error, tokio::CanSocket};
//!
//! #[tokio::main]
//! async fn main() -> Result<(), Error> {
//!     let mut socket_rx = CanSocket::open("vcan0")?;
//!     let socket_tx = CanSocket::open("vcan0")?;
//!
//!     while let Some(Ok(frame)) = socket_rx.next().await {
//!         socket_tx.write_frame(frame).await;
//!     }
//!     Ok(())
//! }
//! ```
use crate::{
    frame::AsPtr, timestamp::CanTimestamps, CanAddr, CanAnyFrame, CanFrame, Error, IoResult,
    Result, Socket, SocketOptions,
};
use futures::{prelude::*, ready, task::Context};
use std::{
    io::{Read, Write},
    os::unix::{
        io::{AsRawFd, OwnedFd},
        prelude::RawFd,
    },
    pin::Pin,
    task::Poll,
    time::{Duration, SystemTime},
};
use tokio::io::unix::AsyncFd;
use tokio::io::Interest;
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};

/// An asynchronous I/O wrapped CanSocket
#[derive(Debug)]
pub struct AsyncCanSocket<T: Socket>(AsyncFd<T>);

impl<T: Socket + From<OwnedFd>> AsyncCanSocket<T> {
    /// Open a named CAN device such as "can0, "vcan0", etc
    pub fn open(ifname: &str) -> IoResult<Self> {
        let sock = T::open(ifname)?;
        sock.set_nonblocking(true)?;
        Ok(Self(AsyncFd::new(sock)?))
    }

    /// Open CAN device by kernel interface number
    pub fn open_if(ifindex: u32) -> IoResult<Self> {
        let sock = T::open_iface(ifindex)?;
        sock.set_nonblocking(true)?;
        Ok(Self(AsyncFd::new(sock)?))
    }

    /// Open a CAN socket by address
    pub fn open_addr(addr: &CanAddr) -> IoResult<Self> {
        let sock = T::open_addr(addr)?;
        sock.set_nonblocking(true)?;
        Ok(Self(AsyncFd::new(sock)?))
    }
}

impl<T: Socket> SocketOptions for AsyncCanSocket<T> {}

impl<T: Socket> AsRawFd for AsyncCanSocket<T> {
    fn as_raw_fd(&self) -> RawFd {
        self.0.as_raw_fd()
    }
}

/// Asynchronous Can Socket
pub type CanSocket = AsyncCanSocket<crate::CanSocket>;

impl CanSocket {
    /// Write a CAN frame to the socket asynchronously
    pub async fn write_frame(&self, frame: CanFrame) -> IoResult<()> {
        self.0
            .async_io(Interest::WRITABLE, |inner| inner.write_frame(&frame))
            .await
    }

    /// Attempt to write a CAN frame to the socket without blocking.
    ///
    /// Returns `WouldBlock` if the send buffer is full.
    ///
    /// Bypasses the tokio readiness reactor: this call goes straight to
    /// the underlying non-blocking fd, so no `AsyncFd` readiness event is
    /// consumed. Mixing with [`write_frame`](Self::write_frame) on the same
    /// socket is safe — the async path will simply re-check OS-level
    /// readiness on its next poll and may briefly round-trip through
    /// `WouldBlock` before re-arming.
    pub fn try_write_frame(&self, frame: CanFrame) -> IoResult<()> {
        self.0.get_ref().write_frame(&frame)
    }

    /// Read a CAN frame from the socket asynchronously
    pub async fn read_frame(&self) -> IoResult<CanFrame> {
        self.0
            .async_io(Interest::READABLE, |inner| inner.read_frame())
            .await
    }

    /// Attempt to read a CAN frame from the socket without blocking.
    ///
    /// Returns `WouldBlock` if no frame is immediately available.
    ///
    /// Bypasses the tokio readiness reactor: this call goes straight to
    /// the underlying non-blocking fd, so no `AsyncFd` readiness event is
    /// consumed. Mixing with [`read_frame`](Self::read_frame) on the same
    /// socket is safe — the async path will simply re-check OS-level
    /// readiness on its next poll and may briefly round-trip through
    /// `WouldBlock` before re-arming.
    pub fn try_read_frame(&self) -> IoResult<CanFrame> {
        self.0.get_ref().read_frame()
    }

    /// Returns `true` if the bound interface supports hardware receive timestamps.
    pub fn has_hw_timestamps(&self) -> bool {
        self.0.get_ref().has_hw_timestamps()
    }

    /// Read a CAN frame and its socket-layer arrival timestamp asynchronously.
    ///
    /// Requires [`SocketOptions::set_recv_timestamp`] to be called first.
    pub async fn read_frame_with_timestamp(&self) -> IoResult<(CanFrame, SystemTime)> {
        self.0
            .async_io(Interest::READABLE, |inner| {
                inner.read_frame_with_timestamp()
            })
            .await
    }

    /// Read a CAN frame and all available timestamps asynchronously.
    ///
    /// Timestamp fields are `None` for modes not enabled on the socket.
    pub async fn read_frame_with_timestamps(&self) -> IoResult<(CanFrame, CanTimestamps)> {
        self.0
            .async_io(Interest::READABLE, |inner| {
                inner.read_frame_with_timestamps()
            })
            .await
    }

    /// Read a CAN frame and its raw hardware clock timestamp asynchronously.
    ///
    /// Requires [`SocketOptions::set_timestamping`] with
    /// `SOF_TIMESTAMPING_RX_HARDWARE | SOF_TIMESTAMPING_OPT_CMSG` to be called first.
    pub async fn read_frame_with_hw_timestamp(&self) -> IoResult<(CanFrame, Duration)> {
        self.0
            .async_io(Interest::READABLE, |inner| {
                inner.read_frame_with_hw_timestamp()
            })
            .await
    }
}

impl Stream for CanSocket {
    type Item = Result<CanFrame>;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
        loop {
            let mut ready_guard = ready!(self.0.poll_read_ready(cx))?;
            match ready_guard.try_io(|inner| inner.get_ref().read_frame()) {
                Ok(result) => return Poll::Ready(Some(result.map_err(|e| e.into()))),
                Err(_would_block) => continue,
            }
        }
    }
}

impl Sink<CanFrame> for CanSocket {
    type Error = Error;

    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
        let _ = ready!(self.0.poll_write_ready(cx))?;
        Poll::Ready(Ok(()))
    }

    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<()>> {
        Poll::Ready(Ok(()))
    }

    fn poll_close(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<()>> {
        // Nothing to flush; the underlying fd closes on drop.
        Poll::Ready(Ok(()))
    }

    fn start_send(self: Pin<&mut Self>, item: CanFrame) -> Result<()> {
        // `poll_ready` already cleared write-readiness, so a single
        // non-blocking write is sufficient — no need to busy-retry.
        self.0.get_ref().write_frame(&item)?;
        Ok(())
    }
}

impl AsyncRead for CanSocket {
    fn poll_read(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<IoResult<()>> {
        loop {
            let mut guard = ready!(self.0.poll_read_ready_mut(cx))?;

            let unfilled = buf.initialize_unfilled();
            match guard.try_io(|inner| inner.get_mut().read(unfilled)) {
                Ok(Ok(len)) => {
                    buf.advance(len);
                    return Poll::Ready(Ok(()));
                }
                Ok(Err(err)) => return Poll::Ready(Err(err)),
                Err(_would_block) => continue,
            }
        }
    }
}

impl AsyncWrite for CanSocket {
    fn poll_write(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<IoResult<usize>> {
        loop {
            let mut guard = ready!(self.0.poll_write_ready_mut(cx))?;

            match guard.try_io(|inner| inner.get_mut().write(buf)) {
                Ok(result) => return Poll::Ready(result),
                Err(_would_block) => continue,
            }
        }
    }

    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<IoResult<()>> {
        Poll::Ready(Ok(()))
    }

    fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<IoResult<()>> {
        Poll::Ready(Ok(()))
    }
}

/// An Asynchronous CAN FD Socket
pub type CanFdSocket = AsyncCanSocket<crate::CanFdSocket>;

impl CanFdSocket {
    /// Write a CAN FD frame to the socket asynchronously
    pub async fn write_frame<F>(&self, frame: &F) -> IoResult<()>
    where
        F: Into<CanAnyFrame> + AsPtr,
    {
        self.0
            .async_io(Interest::WRITABLE, |inner| inner.write_frame(frame))
            .await
    }

    /// Attempt to write a CAN frame to the socket without blocking.
    ///
    /// Returns `WouldBlock` if the send buffer is full.
    ///
    /// Bypasses the tokio readiness reactor: this call goes straight to
    /// the underlying non-blocking fd, so no `AsyncFd` readiness event is
    /// consumed. Mixing with [`write_frame`](Self::write_frame) on the same
    /// socket is safe — the async path will simply re-check OS-level
    /// readiness on its next poll and may briefly round-trip through
    /// `WouldBlock` before re-arming.
    pub fn try_write_frame<F>(&self, frame: &F) -> IoResult<()>
    where
        F: Into<CanAnyFrame> + AsPtr,
    {
        self.0.get_ref().write_frame(frame)
    }

    /// Reads a CAN FD frame from the socket asynchronously
    pub async fn read_frame(&self) -> IoResult<CanAnyFrame> {
        self.0
            .async_io(Interest::READABLE, |inner| inner.read_frame())
            .await
    }

    /// Attempt to read a CAN frame from the socket without blocking.
    ///
    /// Returns `WouldBlock` if no frame is immediately available.
    ///
    /// Bypasses the tokio readiness reactor: this call goes straight to
    /// the underlying non-blocking fd, so no `AsyncFd` readiness event is
    /// consumed. Mixing with [`read_frame`](Self::read_frame) on the same
    /// socket is safe — the async path will simply re-check OS-level
    /// readiness on its next poll and may briefly round-trip through
    /// `WouldBlock` before re-arming.
    pub fn try_read_frame(&self) -> IoResult<CanAnyFrame> {
        self.0.get_ref().read_frame()
    }

    /// Returns `true` if the bound interface supports hardware receive timestamps.
    pub fn has_hw_timestamps(&self) -> bool {
        self.0.get_ref().has_hw_timestamps()
    }

    /// Read a CAN frame and its socket-layer arrival timestamp asynchronously.
    ///
    /// Requires [`SocketOptions::set_recv_timestamp`] to be called first.
    pub async fn read_frame_with_timestamp(&self) -> IoResult<(CanAnyFrame, SystemTime)> {
        self.0
            .async_io(Interest::READABLE, |inner| {
                inner.read_frame_with_timestamp()
            })
            .await
    }

    /// Read a CAN frame and all available timestamps asynchronously.
    ///
    /// Timestamp fields are `None` for modes not enabled on the socket.
    pub async fn read_frame_with_timestamps(&self) -> IoResult<(CanAnyFrame, CanTimestamps)> {
        self.0
            .async_io(Interest::READABLE, |inner| {
                inner.read_frame_with_timestamps()
            })
            .await
    }

    /// Read a CAN frame and its raw hardware clock timestamp asynchronously.
    ///
    /// Requires [`SocketOptions::set_timestamping`] with
    /// `SOF_TIMESTAMPING_RX_HARDWARE | SOF_TIMESTAMPING_OPT_CMSG` to be called first.
    pub async fn read_frame_with_hw_timestamp(&self) -> IoResult<(CanAnyFrame, Duration)> {
        self.0
            .async_io(Interest::READABLE, |inner| {
                inner.read_frame_with_hw_timestamp()
            })
            .await
    }
}

impl Stream for CanFdSocket {
    type Item = Result<CanAnyFrame>;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
        loop {
            let mut ready_guard = ready!(self.0.poll_read_ready(cx))?;
            match ready_guard.try_io(|inner| inner.get_ref().read_frame()) {
                Ok(result) => return Poll::Ready(Some(result.map_err(|e| e.into()))),
                Err(_would_block) => continue,
            }
        }
    }
}

impl Sink<CanAnyFrame> for CanFdSocket {
    type Error = Error;

    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
        let _ = ready!(self.0.poll_write_ready(cx))?;
        Poll::Ready(Ok(()))
    }

    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<()>> {
        Poll::Ready(Ok(()))
    }

    fn poll_close(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<()>> {
        // Nothing to flush; the underlying fd closes on drop.
        Poll::Ready(Ok(()))
    }

    fn start_send(self: Pin<&mut Self>, item: CanAnyFrame) -> Result<()> {
        // `poll_ready` already cleared write-readiness, so a single
        // non-blocking write is sufficient — no need to busy-retry.
        self.0.get_ref().write_frame(&item)?;
        Ok(())
    }
}

impl AsyncRead for CanFdSocket {
    fn poll_read(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<IoResult<()>> {
        loop {
            let mut guard = ready!(self.0.poll_read_ready_mut(cx))?;

            let unfilled = buf.initialize_unfilled();
            match guard.try_io(|inner| inner.get_mut().read(unfilled)) {
                Ok(Ok(len)) => {
                    buf.advance(len);
                    return Poll::Ready(Ok(()));
                }
                Ok(Err(err)) => return Poll::Ready(Err(err)),
                Err(_would_block) => continue,
            }
        }
    }
}

impl AsyncWrite for CanFdSocket {
    fn poll_write(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<IoResult<usize>> {
        loop {
            let mut guard = ready!(self.0.poll_write_ready_mut(cx))?;

            match guard.try_io(|inner| inner.get_mut().write(buf)) {
                Ok(result) => return Poll::Ready(result),
                Err(_would_block) => continue,
            }
        }
    }

    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<IoResult<()>> {
        Poll::Ready(Ok(()))
    }

    fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<IoResult<()>> {
        Poll::Ready(Ok(()))
    }
}

/////////////////////////////////////////////////////////////////////////////

#[cfg(feature = "vcan_tests")]
#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        frame::{can_frame_default, AsPtr},
        CanFdFrame, CanFrame, Frame, IoErrorKind, StandardId,
    };
    use embedded_can::Frame as EmbeddedFrame;
    use futures::{select, try_join};
    use futures_timer::Delay;
    use serial_test::serial;
    use std::time::Duration;
    use tokio::io::{AsyncReadExt, AsyncWriteExt};

    const TIMEOUT: Duration = Duration::from_millis(100);

    /// Receive a frame from the CanSocket
    async fn recv_frame(socket: CanSocket) -> Result<CanSocket> {
        select!(
            frame = socket.read_frame().fuse() => if let Ok(_frame) = frame { Ok(socket) } else { panic!("unexpected") },
            _timeout = Delay::new(TIMEOUT).fuse() => Err(IoErrorKind::TimedOut.into()),
        )
    }

    /// Receive a frame from the CanSocket using the `Stream` trait
    async fn recv_frame_with_stream(mut socket: CanSocket) -> Result<CanSocket> {
        select!(
            frame = socket.next().fuse() => if let Some(_frame) = frame { Ok(socket) } else { panic!("unexpected") },
            _timeout = Delay::new(TIMEOUT).fuse() => Err(IoErrorKind::TimedOut.into()),
        )
    }

    /// Receive a frame from the CanSocket using the `tokio::io::AsyncRead` trait
    async fn recv_frame_with_async_read(mut socket: CanSocket) -> Result<CanSocket> {
        let mut frame = can_frame_default();
        select!(
            // SAFETY: `frame` is fully zero-initialised by `can_frame_default`.
            frame = socket.read_exact(unsafe { crate::as_bytes_mut(&mut frame) }).fuse() => if let Ok(_bytes_read) = frame { Ok(socket) } else { panic!("unexpected") },
            _timeout = Delay::new(TIMEOUT).fuse() => Err(IoErrorKind::TimedOut.into()),
        )
    }

    /// Write a test frame to the CanSocket
    async fn write_frame(socket: &CanSocket) -> Result<()> {
        let test_frame = CanFrame::new(StandardId::new(0x1).unwrap(), &[0]).unwrap();
        socket.write_frame(test_frame).await?;
        Ok(())
    }

    fn try_write_frame(socket: &CanSocket) -> Result<()> {
        let test_frame = CanFrame::new(StandardId::new(0x1).unwrap(), &[0]).unwrap();
        socket.try_write_frame(test_frame)?;
        Ok(())
    }

    /// Write a test frame to the CanSocket using the `tokio::io::AsyncWrite` trait
    async fn write_frame_with_async_write(socket: &mut CanSocket) -> Result<()> {
        let test_frame = CanFrame::new(StandardId::new(0x1).unwrap(), &[0]).unwrap();
        socket.write(test_frame.as_bytes()).await?;
        Ok(())
    }

    /// Receive a frame from the CanFdSocket
    async fn recv_frame_fd(socket: CanFdSocket) -> Result<CanFdSocket> {
        select!(
            frame = socket.read_frame().fuse() => if let Ok(_frame) = frame { Ok(socket) } else { panic!("unexpected") },
            _timeout = Delay::new(TIMEOUT).fuse() => Err(IoErrorKind::TimedOut.into()),
        )
    }

    /// Receive a frame from the CanFdSocket using the `Stream` trait
    async fn recv_frame_fd_with_stream(mut socket: CanFdSocket) -> Result<CanFdSocket> {
        select!(
            frame = socket.next().fuse() => if let Some(_frame) = frame { Ok(socket) } else { panic!("unexpected") },
            _timeout = Delay::new(TIMEOUT).fuse() => Err(IoErrorKind::TimedOut.into()),
        )
    }

    /// Receive a frame from the CanFdSocket using the `tokio::io::AsyncWrite` trait
    async fn recv_frame_fd_with_async_read(mut socket: CanFdSocket) -> Result<CanFdSocket> {
        let mut frame = can_frame_default();
        select!(
            // SAFETY: `frame` is fully zero-initialised by `can_frame_default`.
            frame = socket.read_exact(unsafe { crate::as_bytes_mut(&mut frame) }).fuse() => if let Ok(_bytes_read) = frame { Ok(socket) } else { panic!("unexpected") },
            _timeout = Delay::new(TIMEOUT).fuse() => Err(IoErrorKind::TimedOut.into()),
        )
    }

    /// Write a test CANFD frame to the CanSocket
    async fn write_frame_fd_canfd(socket: &CanFdSocket) -> Result<()> {
        let test_frame =
            CanFdFrame::new(StandardId::new(0x1).unwrap(), &[0, 0, 0, 0, 0, 0, 0, 0, 0]).unwrap();
        socket.write_frame(&test_frame).await?;
        Ok(())
    }

    fn try_write_frame_fd_canfd(socket: &CanFdSocket) -> Result<()> {
        let test_frame =
            CanFdFrame::new(StandardId::new(0x1).unwrap(), &[0, 0, 0, 0, 0, 0, 0, 0, 0]).unwrap();
        socket.try_write_frame(&test_frame)?;
        Ok(())
    }

    /// Write a test CAN frame to the CanSocket
    async fn write_frame_fd_can(socket: &CanFdSocket) -> Result<()> {
        let test_frame = CanFrame::new(StandardId::new(0x1).unwrap(), &[0]).unwrap();
        socket.write_frame(&test_frame).await?;
        Ok(())
    }

    /// Write a test frame to the CanSocket using the `tokio::io::AsyncWrite` trait
    async fn write_frame_fd_with_async_write(socket: &mut CanFdSocket) -> Result<()> {
        let test_frame = CanFdFrame::new(StandardId::new(0x1).unwrap(), &[0]).unwrap();
        socket.write(test_frame.as_bytes()).await?;
        Ok(())
    }

    #[serial]
    #[tokio::test]
    async fn test_receive() -> Result<()> {
        let socket1 = CanSocket::open("vcan0").unwrap();
        let socket2 = CanSocket::open("vcan0").unwrap();

        let send_frames = future::try_join(write_frame(&socket1), write_frame(&socket1));

        let recv_frames = async {
            let socket2 = recv_frame(socket2).await?;
            let _socket2 = recv_frame(socket2).await;
            Ok(())
        };

        try_join!(recv_frames, send_frames)?;

        Ok(())
    }

    #[serial]
    #[tokio::test]
    async fn test_receive_with_stream() -> Result<()> {
        let socket1 = CanSocket::open("vcan0").unwrap();
        let socket2 = CanSocket::open("vcan0").unwrap();

        let send_frames = future::try_join(write_frame(&socket1), write_frame(&socket1));

        let recv_frames = async {
            let socket2 = recv_frame_with_stream(socket2).await?;
            let _socket2 = recv_frame_with_stream(socket2).await;
            Ok(())
        };

        try_join!(recv_frames, send_frames)?;

        Ok(())
    }

    #[serial]
    #[tokio::test]
    async fn test_asyncread_and_asyncwrite() -> Result<()> {
        let mut socket1 = CanSocket::open("vcan0").unwrap();
        let socket2 = CanSocket::open("vcan0").unwrap();

        let send_frames = write_frame_with_async_write(&mut socket1);

        let recv_frames = async {
            let _socket2 = recv_frame_with_async_read(socket2).await?;
            Ok(())
        };

        try_join!(recv_frames, send_frames)?;

        Ok(())
    }

    #[serial]
    #[tokio::test]
    async fn test_tryread_and_trywrite() -> Result<()> {
        let socket1 = CanSocket::open("vcan0").unwrap();
        let socket2 = CanSocket::open("vcan0").unwrap();

        let result = socket2.try_read_frame();
        assert!(result.is_err(), "Expected no frames available");
        assert_eq!(result.unwrap_err().kind(), std::io::ErrorKind::WouldBlock);

        try_write_frame(&socket1)?;
        tokio::time::sleep(Duration::from_millis(100)).await;

        socket2.try_read_frame()?;

        let result = socket2.try_read_frame();
        assert!(result.is_err(), "Expected no frames available");
        assert_eq!(result.unwrap_err().kind(), std::io::ErrorKind::WouldBlock);

        Ok(())
    }

    #[serial]
    #[tokio::test]
    async fn test_receive_can_fd_canfd() -> Result<()> {
        let socket1 = CanFdSocket::open("vcan0").unwrap();
        let socket2 = CanFdSocket::open("vcan0").unwrap();

        let send_frames = future::try_join(
            write_frame_fd_canfd(&socket1),
            write_frame_fd_canfd(&socket1),
        );

        let recv_frames = async {
            let socket2 = recv_frame_fd(socket2).await?;
            let _socket2 = recv_frame_fd(socket2).await;
            Ok(())
        };

        try_join!(recv_frames, send_frames)?;

        Ok(())
    }

    #[serial]
    #[tokio::test]
    async fn test_receive_can_fd_can() -> Result<()> {
        let socket1 = CanFdSocket::open("vcan0").unwrap();
        let socket2 = CanFdSocket::open("vcan0").unwrap();

        let send_frames =
            future::try_join(write_frame_fd_can(&socket1), write_frame_fd_can(&socket1));

        let recv_frames = async {
            let socket2 = recv_frame_fd(socket2).await?;
            let _socket2 = recv_frame_fd(socket2).await;
            Ok(())
        };

        try_join!(recv_frames, send_frames)?;

        Ok(())
    }

    #[serial]
    #[tokio::test]
    async fn test_receive_can_fd_canfd_with_stream() -> Result<()> {
        let socket1 = CanFdSocket::open("vcan0").unwrap();
        let socket2 = CanFdSocket::open("vcan0").unwrap();

        let send_frames = future::try_join(
            write_frame_fd_canfd(&socket1),
            write_frame_fd_canfd(&socket1),
        );

        let recv_frames = async {
            let socket2 = recv_frame_fd_with_stream(socket2).await?;
            let _socket2 = recv_frame_fd_with_stream(socket2).await;
            Ok(())
        };

        try_join!(recv_frames, send_frames)?;

        Ok(())
    }

    #[serial]
    #[tokio::test]
    async fn test_receive_can_fd_can_with_stream() -> Result<()> {
        let socket1 = CanFdSocket::open("vcan0").unwrap();
        let socket2 = CanFdSocket::open("vcan0").unwrap();

        let send_frames =
            future::try_join(write_frame_fd_can(&socket1), write_frame_fd_can(&socket1));

        let recv_frames = async {
            let socket2 = recv_frame_fd_with_stream(socket2).await?;
            let _socket2 = recv_frame_fd_with_stream(socket2).await;
            Ok(())
        };

        try_join!(recv_frames, send_frames)?;

        Ok(())
    }

    #[serial]
    #[tokio::test]
    async fn test_asyncread_and_asyncwrite_fd() -> Result<()> {
        let mut socket1 = CanFdSocket::open("vcan0").unwrap();
        let socket2 = CanFdSocket::open("vcan0").unwrap();

        let send_frames = write_frame_fd_with_async_write(&mut socket1);

        let recv_frames = async {
            let _socket2 = recv_frame_fd_with_async_read(socket2).await?;
            Ok(())
        };

        try_join!(recv_frames, send_frames)?;

        Ok(())
    }

    #[serial]
    #[tokio::test]
    async fn test_sink_stream() -> Result<()> {
        let socket1 = CanSocket::open("vcan0").unwrap();
        let socket2 = CanSocket::open("vcan0").unwrap();

        let frame_id_1 = CanFrame::from_raw_id(0x01, &[0u8]).unwrap();
        let frame_id_2 = CanFrame::from_raw_id(0x02, &[0u8]).unwrap();
        let frame_id_3 = CanFrame::from_raw_id(0x03, &[0u8]).unwrap();

        let (mut sink, _stream) = socket1.split();
        let (_sink, stream) = socket2.split();

        let count_ids_less_than_3 = stream
            .map(|x| x.unwrap())
            .take_while(|frame| future::ready(frame.raw_id() < 3))
            .fold(0u8, |acc, _frame| async move { acc + 1 });

        let send_frames = async {
            let _frame_1 = sink.send(frame_id_1).await?;
            let _frame_2 = sink.send(frame_id_2).await?;
            let _frame_3 = sink.send(frame_id_3).await?;
            println!("Sent 3 frames");
            Ok::<(), Error>(())
        };

        let (x, frame_send_r) = future::join(count_ids_less_than_3, send_frames).await;
        frame_send_r?;

        assert_eq!(x, 2);

        Ok(())
    }

    #[serial]
    #[tokio::test]
    async fn test_sink_stream_fd_canfd() -> Result<()> {
        let socket1 = CanFdSocket::open("vcan0").unwrap();
        let socket2 = CanFdSocket::open("vcan0").unwrap();

        let frame_id_1 = CanFdFrame::from_raw_id(0x01, &[0u8]).unwrap();
        let frame_id_2 = CanFdFrame::from_raw_id(0x02, &[0u8]).unwrap();
        let frame_id_3 = CanFdFrame::from_raw_id(0x03, &[0u8]).unwrap();

        let (mut sink, _stream) = socket1.split();
        let (_sink, stream) = socket2.split();

        let count_ids_less_than_3 = stream
            .map(|x| x.unwrap())
            .take_while(|frame| {
                if let CanAnyFrame::Fd(frame) = frame {
                    future::ready(frame.raw_id() < 3)
                } else {
                    future::ready(false)
                }
            })
            .fold(0u8, |acc, _frame| async move { acc + 1 });

        let send_frames = async {
            let _frame_1 = sink.send(frame_id_1.into()).await?;
            let _frame_2 = sink.send(frame_id_2.into()).await?;
            let _frame_3 = sink.send(frame_id_3.into()).await?;
            println!("Sent 3 frames");
            Ok::<(), Error>(())
        };

        let (x, frame_send_r) = future::join(count_ids_less_than_3, send_frames).await;
        frame_send_r?;

        assert_eq!(x, 2);

        Ok(())
    }

    #[serial]
    #[tokio::test]
    async fn test_sink_stream_fd_can() -> Result<()> {
        let socket1 = CanFdSocket::open("vcan0").unwrap();
        let socket2 = CanFdSocket::open("vcan0").unwrap();

        let frame_id_1 = CanFrame::from_raw_id(0x01, &[0u8]).unwrap();
        let frame_id_2 = CanFrame::from_raw_id(0x02, &[0u8]).unwrap();
        let frame_id_3 = CanFrame::from_raw_id(0x03, &[0u8]).unwrap();

        let (mut sink, _stream) = socket1.split();
        let (_sink, stream) = socket2.split();

        let count_ids_less_than_3 = stream
            .map(|x| x.unwrap())
            .take_while(|frame| {
                if let CanAnyFrame::Normal(frame) = frame {
                    future::ready(frame.raw_id() < 3)
                } else {
                    future::ready(false)
                }
            })
            .fold(0u8, |acc, _frame| async move { acc + 1 });

        let send_frames = async {
            let _frame_1 = sink.send(frame_id_1.into()).await?;
            let _frame_2 = sink.send(frame_id_2.into()).await?;
            let _frame_3 = sink.send(frame_id_3.into()).await?;
            println!("Sent 3 frames");
            Ok::<(), Error>(())
        };

        let (x, frame_send_r) = future::join(count_ids_less_than_3, send_frames).await;
        frame_send_r?;

        assert_eq!(x, 2);

        Ok(())
    }

    #[serial]
    #[tokio::test]
    async fn test_tryread_and_trywrite_fd() -> Result<()> {
        let socket1 = CanFdSocket::open("vcan0").unwrap();
        let socket2 = CanFdSocket::open("vcan0").unwrap();

        let result = socket2.try_read_frame();
        assert!(result.is_err(), "Expected no frames available");
        assert_eq!(result.unwrap_err().kind(), std::io::ErrorKind::WouldBlock);

        try_write_frame_fd_canfd(&socket1)?;
        tokio::time::sleep(Duration::from_millis(100)).await;

        socket2.try_read_frame()?;

        let result = socket2.try_read_frame();
        assert!(result.is_err(), "Expected no frames available");
        assert_eq!(result.unwrap_err().kind(), std::io::ErrorKind::WouldBlock);

        Ok(())
    }
}