risingwave_connector/schema/pulsar_schema/
schema.rs

1// Copyright 2026 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use anyhow::Context;
16use serde::Deserialize;
17
18#[derive(Debug, Deserialize)]
19pub struct PulsarSchemaInfo {
20    pub version: i64,
21    pub r#type: String,
22    pub data: String,
23}
24
25#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
26pub struct PulsarSchemaVersion(pub i64);
27
28impl TryFrom<&[u8]> for PulsarSchemaVersion {
29    type Error = crate::error::ConnectorError;
30
31    fn try_from(version: &[u8]) -> Result<Self, Self::Error> {
32        let bytes: [u8; 8] = version.try_into().with_context(|| {
33            format!(
34                "expected 8-byte Pulsar LongSchemaVersion, got {} bytes",
35                version.len()
36            )
37        })?;
38        Ok(Self(i64::from_be_bytes(bytes)))
39    }
40}
41
42#[cfg(test)]
43mod tests {
44    use super::*;
45
46    #[test]
47    fn pulsar_schema_info_deserializes_type() {
48        let schema: PulsarSchemaInfo =
49            serde_json::from_str(r#"{"version":0,"type":"AVRO","data":"{}"}"#).unwrap();
50        assert_eq!(schema.r#type, "AVRO");
51    }
52
53    #[test]
54    fn pulsar_schema_version_is_signed_big_endian() {
55        assert_eq!(
56            PulsarSchemaVersion::try_from(1_i64.to_be_bytes().as_slice())
57                .unwrap()
58                .0,
59            1
60        );
61        assert_eq!(
62            PulsarSchemaVersion::try_from((-1_i64).to_be_bytes().as_slice())
63                .unwrap()
64                .0,
65            -1
66        );
67    }
68
69    #[test]
70    fn pulsar_schema_version_requires_exactly_eight_bytes() {
71        assert!(PulsarSchemaVersion::try_from([].as_slice()).is_err());
72        assert!(PulsarSchemaVersion::try_from([1, 2, 3].as_slice()).is_err());
73        assert!(PulsarSchemaVersion::try_from([0; 9].as_slice()).is_err());
74    }
75}
risingwave_connector/schema/pulsar_schema/schema.rs

risingwave_connector/schema/pulsar_schema/
schema.rs
