mymq 0.1.0

Broker and message queues
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
116

// NOTE/TODO: We don't yet know the penalty of tracing. Defer this integration.

use tracing::span::{self, Attributes, Record};
use tracing::{debug, field::Field, Event, Level, Metadata};

use std::sync::atomic::{AtomicU64, Ordering::SeqCst};
use std::sync::RwLock;

// metadata
//     span  - id, parent
//           - name, target, level, fields, call_site, module_path, file, line
//     event - name, target, level, fields, call_site, module_path, file, line
//
// span - id, parent

pub struct TraceRecorder {
    name: String,
    ids: RwLock<Vec<Id>>,
    next_span: AtomicU64,
}

struct Id {
    is_span: bool,
    name: String,
    target: String,
    level: Level,
    fields: Vec<Field>,
    file: Option<String>,
    line: Option<u32>,
    module_path: Option<String>,
}

impl Default for Id {
    fn default() -> Id {
        Id {
            is_span: bool::default(),
            name: String::default(),
            target: String::default(),
            level: Level::INFO,
            fields: Vec::default(),
            file: None,
            line: None,
            module_path: None,
        }
    }
}

impl<'a, 'b> From<&'b Metadata<'a>> for Id {
    fn from(val: &'b Metadata<'a>) -> Self {
        let val = Id {
            is_span: val.is_span(),
            name: val.name().to_string(),
            target: val.target().to_string(),
            level: val.level().clone(),
            fields: val.fields().iter().collect(),
            file: val.file().map(|s| s.to_string()),
            line: val.line(),
            module_path: val.module_path().map(|s| s.to_string()),
        };
        val
    }
}

impl Default for TraceRecorder {
    fn default() -> Self {
        let ids = vec![Id::default()];
        let val = TraceRecorder {
            name: "mqtt-recorder".to_string(),
            ids: RwLock::new(ids),
            next_span: AtomicU64::new(1),
        };
        println!("start tracing {:?} ...", val.name);
        val
    }
}

impl tracing::Subscriber for TraceRecorder {
    fn enabled(&self, _metadata: &Metadata<'_>) -> bool {
        true
    }

    fn new_span(&self, attrs: &Attributes<'_>) -> span::Id {
        let md = attrs.metadata();
        println!("new_span {}", md.name());
        if md.is_span() {
            debug!("enabling span for {:?}", md.name());
        } else if md.is_event() {
            debug!("enabling event for {:?}", md.name());
        } else {
            unreachable!()
        }
        let id = span::Id::from_u64(self.next_span.fetch_add(1, SeqCst));
        self.ids.write().unwrap().push(md.into());
        id
    }

    fn record(&self, id: &span::Id, _values: &Record<'_>) {
        println!("record {}", id.into_u64());
    }

    fn record_follows_from(&self, id: &span::Id, _follows: &span::Id) {
        println!("record_follows_from {}", id.into_u64());
    }

    fn enter(&self, id: &span::Id) {
        println!("enter {}", id.into_u64());
    }

    fn event(&self, _event: &Event<'_>) {
        println!("event");
    }

    fn exit(&self, id: &span::Id) {
        println!("exit {}", id.into_u64());
    }
}