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
//! Incremental UTF-8 decoding across chunk boundaries (network streams, pipe reads).
use std::borrow::Cow;
/// Incrementally decodes a byte stream into UTF-8 text.
///
/// Chunks can split a multibyte code point (CJK, emoji, accented characters,
/// smart quotes) across boundaries. Decoding each chunk independently with
/// `String::from_utf8_lossy` corrupts such code points into `U+FFFD`
/// replacement characters. This decoder buffers any trailing incomplete
/// sequence until the rest of its bytes arrive, while still replacing genuinely
/// invalid bytes with `U+FFFD` (matching `from_utf8_lossy` semantics). Call
/// [`Utf8StreamDecoder::finish`] at end of stream so a truncated final sequence
/// becomes `U+FFFD` instead of vanishing.
#[derive(Debug, Default)]
pub struct Utf8StreamDecoder {
pending: Vec<u8>,
}
impl Utf8StreamDecoder {
pub fn new() -> Self {
Self::default()
}
/// Appends `bytes` and returns the decodable UTF-8 prefix. A trailing
/// incomplete multibyte sequence is retained for the next call.
pub fn push(&mut self, bytes: &[u8]) -> String {
let mut out = Vec::with_capacity(bytes.len());
self.push_bytes(bytes, &mut out);
// `push_bytes` only ever writes valid UTF-8, so this never allocates a lossy copy.
String::from_utf8(out).unwrap_or_else(|error| String::from_utf8_lossy(error.as_bytes()).into_owned())
}
/// Like [`Utf8StreamDecoder::push`], but borrows `bytes` without copying when
/// nothing is pending and the chunk is valid UTF-8 (the common case for
/// process output), so hot read loops avoid a per-chunk allocation.
pub fn push_cow<'a>(&mut self, bytes: &'a [u8]) -> Cow<'a, str> {
if self.pending.is_empty()
&& let Ok(text) = std::str::from_utf8(bytes)
{
return Cow::Borrowed(text);
}
Cow::Owned(self.push(bytes))
}
/// Appends `bytes` and writes the decodable UTF-8 prefix directly into
/// `out`. A trailing incomplete multibyte sequence is retained for the
/// next call.
///
/// This avoids the intermediate `String` allocation that `push` creates
/// when the caller only needs bytes (e.g. feeding an SSE byte buffer).
pub fn push_bytes(&mut self, bytes: &[u8], out: &mut Vec<u8>) {
self.pending.extend_from_slice(bytes);
loop {
match std::str::from_utf8(&self.pending) {
Ok(text) => {
out.extend_from_slice(text.as_bytes());
self.pending.clear();
break;
}
Err(err) => {
let valid = err.valid_up_to();
if let Some(valid_bytes) = self.pending.get(..valid) {
// `valid_bytes` is guaranteed valid UTF-8 by `valid_up_to`.
out.extend_from_slice(valid_bytes);
}
match err.error_len() {
// Genuinely invalid sequence: emit replacement and skip it.
Some(invalid_len) => {
out.extend_from_slice("\u{FFFD}".as_bytes());
let _ = self.pending.drain(..valid + invalid_len);
}
// Incomplete trailing sequence: keep it for the next push.
None => {
let _ = self.pending.drain(..valid);
break;
}
}
}
}
}
}
/// Ends the stream: returns any retained incomplete sequence lossily
/// decoded (as `U+FFFD`), leaving the decoder empty.
pub fn finish(&mut self) -> String {
let pending = std::mem::take(&mut self.pending);
String::from_utf8_lossy(&pending).into_owned()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn push_bytes_matches_push_byte_by_byte() {
let full = "data: {\"hello\":\"world\"} xin chào 世界\n\n".as_bytes();
let mut dec_str = Utf8StreamDecoder::new();
let mut dec_bytes = Utf8StreamDecoder::new();
let mut joined = String::new();
for (i, &byte) in full.iter().enumerate() {
let mut out = Vec::new();
dec_bytes.push_bytes(std::slice::from_ref(&byte), &mut out);
let s = dec_str.push(std::slice::from_ref(&byte));
assert_eq!(out, s.clone().into_bytes(), "byte {i}: push_bytes must match push");
joined.push_str(&s);
}
assert_eq!(joined.as_bytes(), full);
assert!(dec_str.finish().is_empty());
}
#[test]
fn split_multibyte_is_held_until_complete() {
// U+00E9 = 0xC3 0xA9 split across chunks.
let mut dec = Utf8StreamDecoder::new();
assert!(dec.push(&[0xC3]).is_empty(), "incomplete multibyte should produce no output");
assert_eq!(dec.push(&[0xA9, b'h', b'i']), "\u{00E9}hi");
}
#[test]
fn invalid_bytes_become_replacement_characters() {
// 0xFF is never a valid UTF-8 lead byte.
let mut dec = Utf8StreamDecoder::new();
assert_eq!(dec.push(&[b'a', 0xFF, b'b']), "a\u{FFFD}b");
}
#[test]
fn push_cow_borrows_valid_chunks_and_buffers_splits() {
let mut dec = Utf8StreamDecoder::new();
assert!(matches!(dec.push_cow("plain 世".as_bytes()), Cow::Borrowed("plain 世")));
// `界` = E7 95 8C split across chunks: nothing until complete, then owned.
assert_eq!(dec.push_cow(&[0xE7, 0x95]), "");
let completed = dec.push_cow(&[0x8C, b'!']);
assert!(matches!(completed, Cow::Owned(_)));
assert_eq!(completed, "界!");
assert!(dec.finish().is_empty());
}
#[test]
fn finish_flushes_a_truncated_final_sequence() {
let mut dec = Utf8StreamDecoder::new();
// First two bytes of the three-byte `—` (0xE2 0x80 0x94).
assert_eq!(dec.push(&[b'x', 0xE2, 0x80]), "x");
assert_eq!(dec.finish(), "\u{FFFD}");
assert!(dec.finish().is_empty(), "finish leaves the decoder empty");
}
}