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
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
//! Tests for concurrent parsing: shared parser behind Arc<Mutex<>> and
//! independent per-thread parsers.
use netflow_parser::{NetflowParser, RouterScopedParser};
use std::sync::{Arc, Mutex};
use std::thread;
/// A valid V5 packet for testing.
const V5_PACKET: [u8; 72] = [
0, 5, 0, 1, 3, 0, 4, 0, 5, 0, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5,
6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5,
6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7,
];
/// Verify that a RouterScopedParser behind Arc<Mutex<>> can be shared across threads.
#[test]
fn test_shared_scoped_parser_across_threads() {
let parser = Arc::new(Mutex::new(RouterScopedParser::<String>::new()));
let mut handles = Vec::new();
for i in 0..4 {
let parser = Arc::clone(&parser);
handles.push(thread::spawn(move || {
let source = format!("router-{}", i);
for _ in 0..10 {
let mut p = parser.lock().unwrap();
let packets = p.parse_from_source(source.clone(), &V5_PACKET).packets;
assert_eq!(packets.len(), 1);
}
}));
}
for h in handles {
h.join().expect("thread panicked");
}
let p = parser.lock().unwrap();
assert_eq!(p.source_count(), 4);
}
/// Verify that independent NetflowParser instances work correctly in parallel threads.
#[test]
fn test_independent_parsers_per_thread() {
let mut handles = Vec::new();
for _ in 0..4 {
handles.push(thread::spawn(|| {
let mut parser = NetflowParser::default();
for _ in 0..20 {
let result = parser.parse_bytes(&V5_PACKET);
assert!(result.is_ok());
assert_eq!(result.packets.len(), 1);
}
}));
}
for h in handles {
h.join().expect("thread panicked");
}
}
/// Verify concurrent V9 template caching and data parsing through a shared parser.
/// Each thread registers a unique template ID, then parses data against it.
#[test]
fn test_shared_parser_concurrent_v9_templates() {
let parser = Arc::new(Mutex::new(NetflowParser::default()));
let mut handles = Vec::new();
for i in 0u16..4 {
let parser = Arc::clone(&parser);
handles.push(thread::spawn(move || {
let template_id = 256 + i;
let tid_bytes = template_id.to_be_bytes();
// V9 template packet: template_id with 1 field (IN_BYTES=1, len=4)
let template_packet: Vec<u8> = vec![
0,
9,
0,
1, // version=9, count=1
0,
0,
0,
0, // sys_up_time
0,
0,
0,
0, // unix_secs
0,
0,
0,
(i + 1) as u8, // sequence
0,
0,
0,
1, // source_id
0,
0, // flowset_id=0 (template)
0,
12, // length=12
tid_bytes[0],
tid_bytes[1], // template_id
0,
1, // field_count=1
0,
1, // field_type=IN_BYTES
0,
4, // field_length=4
];
// V9 data packet using this template
let data_packet: Vec<u8> = vec![
0,
9,
0,
1, // version=9, count=1
0,
0,
0,
0, // sys_up_time
0,
0,
0,
0, // unix_secs
0,
0,
0,
(i + 10) as u8, // sequence
0,
0,
0,
1, // source_id
tid_bytes[0],
tid_bytes[1], // flowset_id = template_id
0,
8, // length = 8 (4 header + 4 data)
0,
0,
0,
(i + 1) as u8, // IN_BYTES value
];
// Register template
{
let mut p = parser.lock().unwrap();
let result = p.parse_bytes(&template_packet);
assert_eq!(
result.packets.len(),
1,
"Template parse should produce 1 packet"
);
}
// Parse data (separate lock acquisition to allow interleaving)
{
let mut p = parser.lock().unwrap();
let result = p.parse_bytes(&data_packet);
assert_eq!(
result.packets.len(),
1,
"Data parse should produce 1 packet"
);
assert!(result.error.is_none(), "Data parse should not error");
}
}));
}
for h in handles {
h.join().expect("thread panicked");
}
// Verify all 4 templates were cached
let p = parser.lock().unwrap();
for i in 0u16..4 {
assert!(
p.has_v9_template(256 + i),
"Template {} should be cached",
256 + i
);
}
}