influx_db.rs

use chrono::{DateTime, Local, Utc};
use failure::{err_msg, Error};
use hyper::{Client, Request};
use std::fmt::Write as _;
use std::marker::PhantomData;

pub struct Done;
pub struct Tags;
pub struct Fields;

pub struct Lines<S>(String, PhantomData<S>);

impl Lines<Done> {
    pub fn new() -> Self {
        Lines(String::new(), PhantomData)
    }

    pub fn measurement(mut self, measurement: &str) -> Lines<Tags> {
        format_measurement(measurement, &mut self.0);
        Lines(self.0, PhantomData)
    }

    pub fn len(&self) -> usize {
        self.0.len()
    }

    pub fn is_empty(&self) -> bool {
        self.0.is_empty()
    }

    pub async fn write(self) -> Result<(), Error> {
        let req = Request::post("http://xxxx/api/upload")
            .header("Authorization", "DATAAPI apikey=\"XXXX\"")
            .header("Content-Type", "text/plain")
            .body(self.0.into())?;

        let resp = Client::new()
            .request(req)
            .await?;

        if resp.status().is_success() {
            Ok(())
        } else {
            Err(err_msg("Request to monster returned non-200"))
        }
    }
}

impl Lines<Tags> {
    pub fn tag(mut self, key: &str, value: &str) -> Self {
        self.0.push(',');
        format_key(key, &mut self.0);
        self.0.push('=');
        format_tag_value(value, &mut self.0);
        Lines(self.0, PhantomData)
    }

    pub fn field(mut self, key: &str, value: f64) -> Lines<Fields> {
        self.0.push(' ');
        format_key(key, &mut self.0);
        self.0.push('=');
        format_field_value(value, &mut self.0);
        Lines(self.0, PhantomData)
    }
}

impl Lines<Fields> {
    pub fn field(mut self, key: &str, value: f64) -> Self {
        self.0.push(',');
        format_key(key, &mut self.0);
        self.0.push('=');
        format_field_value(value, &mut self.0);
        Lines(self.0, PhantomData)
    }

    pub fn stamp(mut self, stamp: DateTime<Utc>) -> Lines<Done> {
        self.0.push(' ');
        format_stamp(stamp, &mut self.0);
        self.0.push('\n');
        Lines(self.0, PhantomData)
    }
}

fn format_measurement(s: &str, buf: &mut String) {
    buf.reserve(s.len());

    for ch in s.chars() {
        if ch == '\\' || ch == ',' || ch == ' ' {
            buf.push('\\');
        }
        buf.push(ch);
    }
}

fn format_key(s: &str, buf: &mut String) {
    buf.reserve(s.len());

    for ch in s.chars() {
        if ch == '\\' || ch == ',' || ch == '=' || ch == ' ' {
            buf.push('\\');
        }
        buf.push(ch);
    }
}

fn format_tag_value(s: &str, buf: &mut String) {
    // Same as above, alias for clarity
    format_key(s, buf)
}

fn format_field_value(f: f64, buf: &mut String) {
    write!(buf, "{}", f)
        .unwrap()
}

fn format_stamp(stamp: DateTime<Utc>, buf: &mut String) {
    // Monster takes ET timestamps ("local" on all our hosts)
    write!(buf, "{}", DateTime::<Local>::from(stamp).timestamp_nanos())
        .unwrap()
}