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
use anyhow::{bail, Result};
use byteorder::{BigEndian, ByteOrder};
fn extract_schema_id<'buf>(buf: &'buf [u8], protocol: &str) -> Result<(i32, &'buf [u8])> {
let expected_len = 5;
if buf.len() < expected_len {
bail!(
"Confluent-style {} datum is too few bytes: expected at least {} bytes, got {}",
protocol,
expected_len,
buf.len()
);
}
let magic = buf[0];
let schema_id = BigEndian::read_i32(&buf[1..5]);
if magic != 0 {
bail!(
"wrong Confluent-style {} serialization magic: expected 0, got {}",
protocol,
magic
);
}
Ok((schema_id, &buf[expected_len..]))
}
pub fn extract_avro_header(buf: &[u8]) -> Result<(i32, &[u8])> {
extract_schema_id(buf, "avro")
}
pub fn extract_protobuf_header(buf: &[u8]) -> Result<(i32, &[u8])> {
let (schema_id, buf) = extract_schema_id(buf, "protobuf")?;
match buf.get(0) {
Some(0) => Ok((schema_id, &buf[1..])),
Some(message_id) => bail!(
"unsupported Confluent-style protobuf message descriptor id: \
expected 0, but found: {}. \
See https://github.com/MaterializeInc/materialize/issues/9250",
message_id
),
None => bail!(
"Confluent-style protobuf datum is too few bytes: expected a message id after magic \
and schema id, got a buffer of length {}",
buf.len()
),
}
}