reduct_base/batch/
v1.rs

1// Copyright 2021-2026 ReductSoftware UG
2// Licensed under the Apache License, Version 2.0
3
4use crate::error::ReductError;
5use crate::{unprocessable_entity, Labels};
6use http::{HeaderMap, HeaderValue};
7
8#[derive(Debug, Clone, PartialEq, Eq)]
9pub struct RecordHeader {
10    pub content_length: u64,
11    pub content_type: String,
12    pub labels: Labels,
13}
14
15/// Parse a batched header into a content length, content type, and labels.
16///
17/// # Arguments
18///
19/// * `header` - The batched header to parse.
20///
21/// # Returns
22///
23/// * `content_length` - The content length of the batched header.
24/// * `content_type` - The content type of the batched header.
25/// * `labels` - The labels of the batched header.
26pub fn parse_batched_header(header: &str) -> Result<RecordHeader, ReductError> {
27    let (content_length, rest) = header
28        .split_once(',')
29        .ok_or(unprocessable_entity!("Invalid batched header"))?;
30    let content_length = content_length
31        .trim()
32        .parse::<u64>()
33        .map_err(|_| unprocessable_entity!("Invalid content length"))?;
34
35    let (content_type, rest) = rest
36        .split_once(',')
37        .unwrap_or((rest, "application/octet-stream"));
38
39    let content_type = if content_type.is_empty() {
40        "application/octet-stream".to_string()
41    } else {
42        content_type.trim().to_string()
43    };
44
45    let mut labels = Labels::new();
46    let mut rest = rest.to_string();
47    while let Some(pair) = rest.split_once('=') {
48        let (key, value) = pair;
49        let key = key.trim();
50
51        rest = if value.starts_with('\"') {
52            let value = value[1..].to_string();
53            let (value, rest) = value
54                .split_once('\"')
55                .ok_or(unprocessable_entity!("Invalid batched header"))?;
56            labels.insert(key.trim().to_string(), value.trim().to_string());
57            rest.trim_start_matches(',').trim().to_string()
58        } else if let Some(ret) = value.split_once(',') {
59            let (value, rest) = ret;
60            labels.insert(key.trim().to_string(), value.trim().to_string());
61            rest.trim().to_string()
62        } else {
63            labels.insert(key.to_string(), value.trim().to_string());
64            break;
65        };
66    }
67
68    Ok(RecordHeader {
69        content_length,
70        content_type,
71        labels,
72    })
73}
74
75pub fn sort_headers_by_time(headers: &HeaderMap) -> Result<Vec<(u64, HeaderValue)>, ReductError> {
76    let sorted_headers: Vec<_> = headers
77        .clone()
78        .into_iter()
79        .filter(|(name, _)| name.is_some())
80        .map(|(name, value)| (name.unwrap().to_string(), value))
81        .filter(|(name, _)| name.starts_with("x-reduct-time-"))
82        .map(|(key, value)| (key[14..].parse::<u64>().ok(), (key, value)))
83        .collect();
84
85    for (time, (key, _)) in &sorted_headers {
86        if time.is_none() {
87            return Err(unprocessable_entity!(
88                "Invalid header '{}': must be an unix timestamp in microseconds",
89                key
90            ));
91        }
92    }
93
94    let mut sorted_headers: Vec<(u64, HeaderValue)> = sorted_headers
95        .into_iter()
96        .map(|(time, (_key, value))| (time.unwrap(), value))
97        .collect();
98    sorted_headers.sort_by(|(ts1, _), (ts2, _)| ts1.cmp(ts2));
99    Ok(sorted_headers)
100}
101
102#[cfg(test)]
103mod tests {
104    use super::*;
105    use rstest::*;
106
107    #[rstest]
108    fn test_parse_batched_header_row() {
109        let header = "123, text/plain, label1=value1, label2=value2";
110        let RecordHeader {
111            content_length,
112            content_type,
113            labels,
114        } = parse_batched_header(header).unwrap();
115        assert_eq!(content_length, 123);
116        assert_eq!(content_type, "text/plain");
117        assert_eq!(labels.len(), 2);
118        assert_eq!(labels.get("label1"), Some(&"value1".to_string()));
119        assert_eq!(labels.get("label2"), Some(&"value2".to_string()));
120    }
121
122    #[rstest]
123    fn test_parse_batched_header_row_quotes() {
124        let header = "123, text/plain, label1=\"[1, 2, 3]\", label2=\"value2\"";
125        let RecordHeader {
126            content_length,
127            content_type,
128            labels,
129        } = parse_batched_header(header).unwrap();
130        assert_eq!(content_length, 123);
131        assert_eq!(content_type, "text/plain");
132        assert_eq!(labels.len(), 2);
133        assert_eq!(labels.get("label1"), Some(&"[1, 2, 3]".to_string()));
134        assert_eq!(labels.get("label2"), Some(&"value2".to_string()));
135    }
136
137    #[rstest]
138    fn test_parse_header_no_labels() {
139        let header = "123, text/plain";
140        let RecordHeader {
141            content_length,
142            content_type,
143            labels,
144        } = parse_batched_header(header).unwrap();
145        assert_eq!(content_length, 123);
146        assert_eq!(content_type, "text/plain");
147        assert_eq!(labels.len(), 0);
148    }
149
150    #[rstest]
151    #[case("")]
152    #[case("xxx")]
153    fn test_parse_header_bad_header(#[case] header: &str) {
154        let err = parse_batched_header(header).err().unwrap();
155        assert_eq!(err, unprocessable_entity!("Invalid batched header"));
156    }
157
158    #[rstest]
159    fn test_parse_header_with_computed_label() {
160        let RecordHeader { labels, .. } =
161            parse_batched_header("123, text/plain, @label1=value1, label2=value2").unwrap();
162
163        assert_eq!(labels.get("@label1"), Some(&"value1".to_string()));
164        assert_eq!(labels.get("label2"), Some(&"value2".to_string()));
165    }
166}
reduct_base/batch/v1.rs

reduct_base/batch/
v1.rs
