ocpp-client 0.4.0

OCPP Client Implementation. Use this library to implement an OCPP charge point
Documentation 
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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93

use crate::transport::{TransportError, TransportEvent, TransportSink, TransportStream};
use alloc::boxed::Box;
use core::future::Future;
use core::pin::Pin;
use futures::stream::{SplitSink, SplitStream};
use futures::{SinkExt, StreamExt};
use tokio::net::TcpStream;
use tokio_tungstenite::tungstenite::Message;
use tokio_tungstenite::{MaybeTlsStream, WebSocketStream};

type WsStream = WebSocketStream<MaybeTlsStream<TcpStream>>;

pub(crate) struct WebSocketSink(SplitSink<WsStream, Message>);
pub(crate) struct WebSocketSource(SplitStream<WsStream>);

fn boxed(err: tokio_tungstenite::tungstenite::Error) -> TransportError {
    Box::new(err)
}

impl TransportSink for WebSocketSink {
    fn send<'a>(
        &'a mut self,
        frame: String,
    ) -> Pin<Box<dyn Future<Output = Result<(), TransportError>> + Send + 'a>> {
        Box::pin(async move { self.0.send(Message::text(frame)).await.map_err(boxed) })
    }

    fn ping<'a>(
        &'a mut self,
        payload: Vec<u8>,
    ) -> Pin<Box<dyn Future<Output = Result<(), TransportError>> + Send + 'a>> {
        Box::pin(async move {
            self.0
                .send(Message::Ping(payload.into()))
                .await
                .map_err(boxed)
        })
    }

    fn pong<'a>(
        &'a mut self,
        payload: Vec<u8>,
    ) -> Pin<Box<dyn Future<Output = Result<(), TransportError>> + Send + 'a>> {
        Box::pin(async move {
            self.0
                .send(Message::Pong(payload.into()))
                .await
                .map_err(boxed)
        })
    }

    fn close<'a>(
        &'a mut self,
    ) -> Pin<Box<dyn Future<Output = Result<(), TransportError>> + Send + 'a>> {
        Box::pin(async move { self.0.close().await.map_err(boxed) })
    }
}

impl TransportStream for WebSocketSource {
    fn recv<'a>(
        &'a mut self,
    ) -> Pin<Box<dyn Future<Output = Result<Option<TransportEvent>, TransportError>> + Send + 'a>>
    {
        Box::pin(async move {
            loop {
                return match self.0.next().await {
                    None => Ok(None),
                    Some(Err(err)) => Err(boxed(err)),
                    Some(Ok(Message::Text(text))) => {
                        Ok(Some(TransportEvent::Frame(text.to_string())))
                    }
                    Some(Ok(Message::Ping(payload))) => {
                        Ok(Some(TransportEvent::Ping(payload.into())))
                    }
                    Some(Ok(Message::Pong(payload))) => {
                        Ok(Some(TransportEvent::Pong(payload.into())))
                    }
                    Some(Ok(Message::Close(_))) => Ok(None),
                    // Binary/frame frames aren't valid OCPP-J; skip and keep reading.
                    Some(Ok(_)) => continue,
                };
            }
        })
    }
}

pub(crate) fn split(stream: WsStream) -> (Box<dyn TransportSink>, Box<dyn TransportStream>) {
    let (sink, stream) = stream.split();
    (
        Box::new(WebSocketSink(sink)),
        Box::new(WebSocketSource(stream)),
    )
}