1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

use crate::Frame;
use async_hal::can::{CanReceive, CanTransmit, Frame as _};
use core::{
    marker::PhantomData,
    pin::Pin,
    task::{Context, Poll},
};
use embedded_hal::can::Id;
use futures::{Sink, SinkExt, Stream, StreamExt};
use pin_project_lite::pin_project;

pin_project! {
    pub struct Socket<C, E, F> {
        id: Id,

        #[pin]
        can: C,
        _marker: PhantomData<(E, F)>,
    }

}

impl<C, E, F> Socket<C, E, F> {
    pub fn new(id: impl Into<Id>, can: C) -> Self {
        Self {
            id: id.into(),
            can,
            _marker: PhantomData,
        }
    }
}

impl<C, E, F> Stream for Socket<C, E, F>
where
    C: CanReceive<Error = E>,
{
    type Item = Result<Frame, E>;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
        self.project()
            .can
            .poll_next_unpin(cx)
            .map_ok(|can_frame| Frame::from_bytes(can_frame.data()))
    }
}

impl<C, E, F> Sink<Frame> for Socket<C, E, F>
where
    C: CanTransmit<F>,
    F: async_hal::can::Frame,
{
    type Error = C::Error;

    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
        self.project().can.poll_ready_unpin(cx)
    }

    fn start_send(self: Pin<&mut Self>, item: Frame) -> Result<(), Self::Error> {
        let can_frame = F::new(self.id, item.as_ref()).unwrap();
        self.project().can.start_send_unpin(can_frame)
    }

    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
        self.project().can.poll_flush_unpin(cx)
    }

    fn poll_close(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
        self.project().can.poll_close_unpin(cx)
    }
}