First implementation of REST API

Cargo.toml

@@ -7,8 +7,13 @@ edition = "2018"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-clap = "2.33.3"
+async-std = "1.9"
+clap = "2.33"
 toml = "0.5"
 serde = "1.0"
 serde_derive = "1.0"
-function_name = "0.2.0"
+function_name = "0.2"
+tide = "0.16"
+
+lazy_static = "1.4"
+minreq = { version = "2.3", features = ["json-using-serde"] }

src/main.rs

@@ -1,8 +1,11 @@
 #[macro_use]
 extern crate clap;
 
+use async_std::task;
+
 mod cfg_reader;
 mod pulse_counter;
+mod rest_api;
 
 const APP_VERSION: &'static str = env!("CARGO_PKG_VERSION");
 const DEFAULT_CONFIG_FILE_NAME: &str = "/etc/s0_logger.cfg";
@@ -29,4 +32,22 @@ fn main() {
         .unwrap_or(DEFAULT_CONFIG_FILE_NAME);
     println!("Read the config from file '{}'", config_file_name);
 
+    let rest_api_config = rest_api::RestApiConfig {
+        ip_and_port: String::from("127.0.0.1:8080"),
+        get_channels: fake_get_channels,
+        get_pulses_by_channel: fake_get_pulses_by_channel,
+    };
+
+    let _ = task::block_on(rest_api::start(&rest_api_config));
+}
+
+fn fake_get_channels() -> Vec<usize> {
+    vec![]
+}
+
+fn fake_get_pulses_by_channel(_: usize) -> Result<Vec<rest_api::PulseInfo>, std::io::Error> {
+    Err(std::io::Error::new(
+        std::io::ErrorKind::NotFound,
+        "Not implemented",
+    ))
 }

src/rest_api.rs

@@ -0,0 +1,90 @@
+#[allow(dead_code)]
+#[cfg(test)]
+#[path = "./rest_api_test.rs"]
+mod rest_api_test;
+
+use serde::{Deserialize, Serialize};
+use tide;
+use tide::prelude::json;
+
+// struct PinConfiguration {
+//     channel: usize,
+//     gpio: usize,
+// }
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct PulseInfo {
+    timestamp_ns: u64,
+    pin_id: usize,
+    level: bool,
+}
+
+#[derive(Clone)]
+pub struct RestApiConfig {
+    pub ip_and_port: String,
+    pub get_channels: fn() -> Vec<usize>,
+    pub get_pulses_by_channel: fn(channel: usize) -> Result<Vec<PulseInfo>, std::io::Error>,
+}
+
+pub async fn start<'a>(config: &'a RestApiConfig) -> std::io::Result<()> {
+    let mut app = tide::with_state(config.clone());
+    app.at("/api_versions").get(api_versions_get);
+    app.at("/v1/channels").get(v1_channels_get);
+    app.at("/v1/channel/:channel_id/pulses")
+        .get(v1_channel_pulses_get);
+    let ip_and_port = String::from(&config.ip_and_port);
+    match app.listen(ip_and_port).await {
+        Ok(_) => Ok(()),
+        Err(e) => Err(e),
+    }
+}
+
+async fn api_versions_get(_: tide::Request<RestApiConfig>) -> tide::Result {
+    #[derive(Serialize)]
+    struct ApiVersionInfo {
+        version: String,
+        path: String,
+    }
+    let mut version_list = Vec::<ApiVersionInfo>::new();
+    version_list.push(ApiVersionInfo {
+        version: String::from("1.0"),
+        path: String::from("v1"),
+    });
+    Ok(tide::Response::builder(200)
+        .content_type(tide::http::mime::JSON)
+        .body(json!({ "api_versions": &version_list }))
+        .build())
+}
+
+async fn v1_channels_get(req: tide::Request<RestApiConfig>) -> tide::Result {
+    let channel_list = &(req.state().get_channels)();
+    Ok(tide::Response::builder(200)
+        .content_type(tide::http::mime::JSON)
+        .body(json!({ "channels": channel_list }))
+        .build())
+}
+
+async fn v1_channel_pulses_get(req: tide::Request<RestApiConfig>) -> tide::Result {
+    match req.param("channel_id") {
+        Ok(channel_id) => match usize::from_str_radix(channel_id, 10) {
+            Ok(channel_id) => {
+                let info = &req.state().ip_and_port;
+                let channel_pulses_res = &(req.state().get_pulses_by_channel)(channel_id);
+                match channel_pulses_res {
+                    Ok(channel_pulses) => Ok(tide::Response::builder(200)
+                        .content_type(tide::http::mime::JSON)
+                        .body(json!({
+                            "channel": channel_id,
+                            "pulses": channel_pulses,
+                            "question": "channel",
+                            "info": info,
+                        }))
+                        .build()),
+                    Err(_) => Ok(tide::Response::new(404)),
+                }
+            }
+            Err(_) => Ok(tide::Response::new(405)),
+        },
+        Err(_e) => Ok(tide::Response::new(404)),
+    }
+}

src/rest_api_test.rs

@@ -0,0 +1,222 @@
+use super::*;
+
+extern crate lazy_static;
+
+use async_std::task;
+use lazy_static::lazy_static;
+use minreq;
+use serde_derive::Deserialize;
+use std::collections::HashMap;
+use std::io::{Error, ErrorKind};
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::thread;
+
+static IP_AND_PORT: &str = "127.0.0.1:8123";
+
+lazy_static! {
+    static ref TESTEE_THREAD_HANDLE: Arc<Mutex<Option<thread::JoinHandle<()>>>> = {
+        let config = RestApiConfig {
+            ip_and_port: String::from(IP_AND_PORT),
+            get_channels,
+            get_pulses_by_channel,
+        };
+        let hdl = Arc::new(Mutex::new(Some(thread::spawn(move || {
+            task::block_on(start(&config.clone())).unwrap();
+        }))));
+        println!("Server thread spawned");
+        thread::sleep(std::time::Duration::from_millis(5));
+        hdl
+    };
+    static ref CHANNEL_LIST: Mutex<Vec<usize>> = Mutex::new(Vec::new());
+    static ref CHANNEL_PULSES: Mutex<HashMap<usize, Vec<PulseInfo>>> = Mutex::new(HashMap::new());
+}
+
+fn get_channels() -> Vec<usize> {
+    CHANNEL_LIST.lock().unwrap().to_vec()
+}
+
+fn get_pulses_by_channel(channel_id: usize) -> Result<Vec<PulseInfo>, std::io::Error> {
+    let mut pulse_channel_map = CHANNEL_PULSES.lock().unwrap();
+    match pulse_channel_map.get_mut(&channel_id) {
+        Some(pulse_list) => Ok(pulse_list.drain(0..).collect()),
+        None => Err(Error::new(
+            ErrorKind::NotFound,
+            format!("Channel {} does not exist", &channel_id),
+        )),
+    }
+}
+
+fn launch_testee() {
+    let _hdl = &*TESTEE_THREAD_HANDLE.clone();
+}
+
+fn print_response(_context: &str, _response: &minreq::Response) {
+    return;
+    /*
+        let mut r_str = String::new();
+        r_str.push_str(format!("Response in {}:\n", &_context).as_str());
+        r_str.push_str(format!("   status: {}\n", &_response.status_code).as_str());
+        r_str.push_str(format!("   reason: '{}'\n", &_response.reason_phrase).as_str());
+        r_str.push_str(format!("   headers:\n").as_str());
+        for (key, value) in _response.headers.iter() {
+            r_str.push_str(format!("      {}: '{}'\n", key, value).as_str());
+        }
+        r_str.push_str(format!("   body: '{}'\n", &_response.as_str().unwrap()).as_str());
+        println!("{}\n---", &r_str);
+    */
+}
+
+fn set_channel_list(mut new_channel_list: Vec<usize>) {
+    match CHANNEL_LIST.lock() {
+        Ok(mut channel_list) => {
+            channel_list.clear();
+            channel_list.append(&mut new_channel_list);
+        }
+        Err(_) => assert!(false, "ERROR: Could not access CHANNEL_LIST"),
+    }
+}
+
+fn set_channel_pulses(channel_id: usize, mut channel_pulses: Vec<PulseInfo>) {
+    let mut pulse_channel_map = CHANNEL_PULSES.lock().unwrap();
+    let pulse_list = pulse_channel_map.entry(channel_id).or_insert(vec![]);
+    pulse_list.clear();
+    pulse_list.append(&mut channel_pulses)
+}
+
+#[test]
+fn rest_api_fetch_api_versions() {
+    launch_testee();
+    let response = minreq::get(format!("http://{}/api_versions", &IP_AND_PORT)).send();
+    assert!(&response.is_ok());
+    let response = response.unwrap();
+    assert_eq!(response.status_code, 200);
+    print_response("get_api_versions", &response)
+}
+
+#[test]
+fn rest_api_fetch_channels() {
+    #[derive(Debug, Deserialize)]
+    struct ChannelList {
+        channels: Vec<usize>,
+    }
+
+    launch_testee();
+
+    set_channel_list(vec![]);
+    let response = minreq::get(format!("http://{}/v1/channels", &IP_AND_PORT)).send();
+    assert!(&response.is_ok());
+    let response = response.unwrap();
+    // print_response("get_channels", &response);
+    assert_eq!(response.status_code, 200);
+    assert_eq!(response.headers["content-type"], "application/json");
+    let response_json = &response.json::<ChannelList>().unwrap();
+    assert_eq!(&response_json.channels, &*CHANNEL_LIST.lock().unwrap());
+
+    set_channel_list(vec![1, 3]);
+    let response = minreq::get(format!("http://{}/v1/channels", &IP_AND_PORT)).send();
+    assert!(&response.is_ok());
+    let response = response.unwrap();
+    let response_json = &response.json::<ChannelList>().unwrap();
+    assert_eq!(&response_json.channels, &*CHANNEL_LIST.lock().unwrap());
+
+    set_channel_list(vec![2, 3, 5, 6, 7, 8]);
+    let response = minreq::get(format!("http://{}/v1/channels", &IP_AND_PORT)).send();
+    assert!(&response.is_ok());
+    let response = response.unwrap();
+    let response_json = &response.json::<ChannelList>().unwrap();
+    assert_eq!(&response_json.channels, &*CHANNEL_LIST.lock().unwrap());
+}
+
+#[test]
+fn rest_api_fetch_channel() {
+    #[derive(Debug, Deserialize)]
+    struct ChannelPulsesList {
+        channel: usize,
+        pulses: Vec<PulseInfo>,
+    }
+
+    launch_testee();
+
+    let response = minreq::get(format!("http://{}/v1/channel/1/pulses", &IP_AND_PORT)).send();
+    assert!(&response.is_ok());
+    let response = response.unwrap();
+    print_response("rest_api_fetch_channel", &response);
+    assert_eq!(
+        response.status_code,
+        tide::http::StatusCode::NotFound as i32
+    );
+
+    set_channel_pulses(1, vec![]);
+
+    let response = minreq::get(format!("http://{}/v1/channel/2/pulses", &IP_AND_PORT)).send();
+    assert!(&response.is_ok());
+    let response = response.unwrap();
+    print_response("rest_api_fetch_channel", &response);
+    assert_eq!(
+        response.status_code,
+        tide::http::StatusCode::NotFound as i32
+    );
+
+    let response = minreq::get(format!("http://{}/v1/channel/1/pulses", &IP_AND_PORT)).send();
+    assert!(&response.is_ok());
+    let response = response.unwrap();
+    print_response("rest_api_fetch_channel", &response);
+    assert_eq!(response.status_code, tide::http::StatusCode::Ok as i32);
+    let response_json = &response.json::<ChannelPulsesList>().unwrap();
+    assert_eq!(response_json.channel, 1);
+    assert_eq!(response_json.pulses.len(), 0);
+
+    let pulses = vec![
+        PulseInfo {
+            timestamp_ns: 1234u64,
+            pin_id: 1,
+            level: true,
+        },
+        PulseInfo {
+            timestamp_ns: 1256u64,
+            pin_id: 1,
+            level: false,
+        },
+        PulseInfo {
+            timestamp_ns: 1278u64,
+            pin_id: 1,
+            level: true,
+        },
+        PulseInfo {
+            timestamp_ns: 1290u64,
+            pin_id: 1,
+            level: false,
+        },
+    ];
+    set_channel_pulses(1, pulses);
+    let response = minreq::get(format!("http://{}/v1/channel/1/pulses", &IP_AND_PORT)).send();
+    assert!(&response.is_ok());
+    let response = response.unwrap();
+    print_response("rest_api_fetch_channel", &response);
+    assert_eq!(response.status_code, tide::http::StatusCode::Ok as i32);
+    let response_json = &response.json::<ChannelPulsesList>().unwrap();
+    assert_eq!(response_json.channel, 1);
+    assert_eq!(response_json.pulses.len(), 4);
+    assert_eq!(response_json.pulses.get(0).unwrap().timestamp_ns, 1234u64);
+    assert_eq!(response_json.pulses.get(0).unwrap().pin_id, 1);
+    assert_eq!(response_json.pulses.get(0).unwrap().level, true);
+    assert_eq!(response_json.pulses.get(1).unwrap().timestamp_ns, 1256u64);
+    assert_eq!(response_json.pulses.get(1).unwrap().pin_id, 1);
+    assert_eq!(response_json.pulses.get(1).unwrap().level, false);
+    assert_eq!(response_json.pulses.get(2).unwrap().timestamp_ns, 1278u64);
+    assert_eq!(response_json.pulses.get(2).unwrap().pin_id, 1);
+    assert_eq!(response_json.pulses.get(2).unwrap().level, true);
+    assert_eq!(response_json.pulses.get(3).unwrap().timestamp_ns, 1290u64);
+    assert_eq!(response_json.pulses.get(3).unwrap().pin_id, 1);
+    assert_eq!(response_json.pulses.get(3).unwrap().level, false);
+
+    let response = minreq::get(format!("http://{}/v1/channel/1/pulses", &IP_AND_PORT)).send();
+    assert!(&response.is_ok());
+    let response = response.unwrap();
+    print_response("rest_api_fetch_channel", &response);
+    assert_eq!(response.status_code, tide::http::StatusCode::Ok as i32);
+    let response_json = &response.json::<ChannelPulsesList>().unwrap();
+    assert_eq!(response_json.channel, 1);
+    assert_eq!(response_json.pulses.len(), 0);
+}