ubmsc 0.1.1

A client for Battery Management Systems (BMS).
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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115

use crate::{log, Error, Exporter, Main, Result};
use std::net::SocketAddr;

use http_body_util::Full;
use hyper::{
    body::Bytes,
    client::conn::http1::handshake,
    header::{CONTENT_TYPE, HOST},
    Request, Uri,
};
use hyper_util::rt::TokioIo;
use tokio::{net::TcpStream, select, time::interval};

#[derive(thiserror::Error, Debug)]
enum PushError {
    /// Client error
    #[error("Client error: {0}")]
    Client(#[from] Error),

    /// HttpError
    #[error("HTTP error: {0}")]
    Http(#[from] http::Error),

    /// HyperError
    #[error("Hyper error: {0}")]
    Hyper(#[from] hyper::Error),

    /// Invalid status
    #[error("Invalid response status: {0}")]
    BadStatus(u16),
}

impl From<std::io::Error> for PushError {
    fn from(error: std::io::Error) -> Self {
        Self::Client(error.into())
    }
}

impl Exporter {
    async fn do_request(
        &self,
        addr: &SocketAddr,
        url: &Uri,
    ) -> core::result::Result<(), PushError> {
        log::info!("Connect to {addr}");

        let stream = TcpStream::connect(addr).await?;

        let io = TokioIo::new(stream);

        let (mut sender, conn) = handshake(io).await?;

        tokio::task::spawn(async move {
            if let Err(error) = conn.await {
                log::error!("Connection failed: {error:?}");
            }
        });

        let mut data = Vec::with_capacity(4096);

        let content_type = self.encode(None, &mut data)?;

        let request = Request::put(url)
            .header(HOST, url.host().unwrap())
            .header(CONTENT_TYPE, content_type)
            .body(Full::<Bytes>::from(data))?;

        log::debug!("Request: {request:?}");

        let response = sender.send_request(request).await?;

        log::debug!("Response: {response:?}");

        if !response.status().is_success() {
            return Err(PushError::BadStatus(response.status().as_u16()));
        }

        Ok(())
    }
}

impl Main {
    pub async fn run_exporter_client(&self) -> Result<()> {
        let addr = self.url_addr().await?;

        let exporter = Exporter::new(self.default_encoding(), &self.clients)?;

        let mut poller = interval(self.args.scrape_interval);

        if self.exporter {
            log::info!("Start pusher for: {addr}");

            loop {
                select! {
                    _ = poller.tick() => (),
                    _ = self.intr.notified() => break,
                }

                if exporter.scrape(&self.clients).await.is_ok() {
                    if let Err(error) = exporter.do_request(&addr, &self.url).await {
                        log::error!("Error while pushing metrics: {error}");
                    }
                }
            }

            log::info!("Stop pusher for: {addr}");
        } else if exporter.scrape(&self.clients).await.is_ok() {
            if let Err(error) = exporter.do_request(&addr, &self.url).await {
                log::error!("Error while pushing metrics: {error}");
            }
        }

        Ok(())
    }
}