datapipe 0.1.4

Stream data from here to there
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

use crate::datapipe_types::{InputReader, good_url};
/// Reader for HTTP
use bytes::Bytes;
use log::{error, trace};
use reqwest::StatusCode;
use std::io::{Error, ErrorKind};

#[derive(Debug)]
pub struct HttpReader {
    client: reqwest::Client,
    url: url::Url,
    update_interval: tokio::time::Interval,
}

impl HttpReader {
    pub const DEFAULT_UPDATE_RATE: u64 = 5000; // milliseconds

    pub fn new(http_input_url: &str, update_rate: u64) -> Result<Self, Error> {
        // HTTP client init and configuration
        let url = good_url(http_input_url, "http://")?;
        Ok(Self {
            client: reqwest::Client::builder()
                .user_agent("datapipe")
                .build()
                .unwrap(),
            url,
            update_interval: tokio::time::interval(std::time::Duration::from_millis(update_rate)),
        })
    }
}

impl InputReader for HttpReader {
    async fn read(&mut self) -> Result<Bytes, Error> {
        self.update_interval.tick().await; // wait until it is time to read
        match self.client.get(self.url.as_str()).send().await {
            Ok(response) => {
                trace!("HttpInput:  Web server response is: {:?}", response);
                match response.status() {
                    StatusCode::OK => {
                        trace!("HttpInput:  Status is OK. Getting response body bytes");
                        match response.bytes().await {
                            Ok(bytes) => Ok(bytes),
                            Err(error) => {
                                let error_message = format!(
                                    "HttpInput:  Error converting response body to bytes: {}",
                                    error
                                );
                                error!("{}", error_message);
                                Err(Error::other(error_message))
                            }
                        }
                    }
                    _ => {
                        error!("HttpInput:  non-Ok status from web server: {:?}", response);
                        match response.error_for_status() {
                            Ok(res) => {
                                let error_message = format!(
                                    "HttpInput:  Failed converting web server response to error: {:?}",
                                    res
                                );
                                error!("{}", error_message);
                                Err(Error::other(error_message))
                            }
                            Err(error) => {
                                let error_message =
                                    format!("HttpInput:  decoded web server status: {}", error);
                                error!("{}", error_message);
                                Err(Error::other(error_message))
                            }
                        }
                    }
                }
            }
            Err(error) => {
                let error_message = format!("HttpInput:  Error getting HTTP input: {}", error);
                error!("{}", error_message);
                Err(Error::new(ErrorKind::NotConnected, error_message))
            }
        }
    }
}