metriqs 0.1.6

Metrics infrastructure framework
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

use std::net::{ToSocketAddrs, UdpSocket};
use std::str;
use std::sync::mpsc::channel;
use std::thread;

use super::parse_metrics;
use super::super::super::collector::Collector;

/// Listens for StatsD UDP datagrams.
pub struct StatsdUdpListener {
    collector: Collector,
}

impl StatsdUdpListener {
    pub fn new(collector: Collector) -> StatsdUdpListener {
        StatsdUdpListener {
            collector,
        }
    }

    /// Spawns a separate thread that listens for StatsD UDP datagrams,
    /// received datagrams are sent back to the calling thread (this will
    /// block) and the parsed metrics are recorded in the store.
    pub fn listen<A: ToSocketAddrs>(&self, addr: A) {
        let (send, recv) = channel();

        let socket = UdpSocket::bind(addr).unwrap();

        thread::spawn(move || {
            // Big enough to hold an ethernet frame:
            //   https://github.com/etsy/statsd/blob/master/docs/metric_types.md#multi-metric-packets
            let mut buf = [0; 1500];
            loop {
                let (bytes_read, _) = match socket.recv_from(&mut buf) {
                    Ok(pair) => pair,
                    Err(_) => return,
                };

                // Get a string from just the amount of bytes read.
                let message: &str = match str::from_utf8(&buf[..bytes_read]) {
                    Ok(s) => s,
                    Err(_) => return,
                };

                send.send(message.to_owned()).unwrap();
            }
        });

        for line in recv {
            match parse_metrics(line.trim_right().as_bytes()) {
                Ok(metrics) => {
                    let aggregated_metrics = metrics.into_iter()
                        .map(|metric| metric.into())
                        .collect();

                    self.collector.push(aggregated_metrics)
                },
                Err(_) => (),
            }
        }
    } // fn listen
} // impl StatsdUdpListener