feat: support external schema registry (PIP-420)

Cargo.toml

@@ -1,3 +1,6 @@
+[workspace]
+members = ["pulsar-schema-confluent"]
+
 [package]
 name = "pulsar"
 version = "6.4.1"

PulsarApi.proto

@@ -45,6 +45,7 @@ message Schema {
     LocalTime = 18;
     LocalDateTime = 19;
     ProtobufNative = 20;
+    External = 22;
   }
 
   required string name = 1;
@@ -162,6 +163,7 @@ message MessageMetadata {
 
   // Indicate if the message partition key is set
   optional bool null_partition_key = 30 [default = false];
+  optional bytes schema_id = 32;
 }
 
 message SingleMessageMetadata {

examples/batching.rs

@@ -18,10 +18,7 @@ struct TestData {
 impl SerializeMessage for TestData {
     fn serialize_message(input: Self) -> Result<producer::Message, PulsarError> {
         let payload = serde_json::to_vec(&input).map_err(|e| PulsarError::Custom(e.to_string()))?;
-        Ok(producer::Message {
-            payload,
-            ..Default::default()
-        })
+        Ok(producer::Message::new(payload))
     }
 }
 

examples/producer.rs

@@ -16,10 +16,7 @@ struct TestData {
 impl SerializeMessage for TestData {
     fn serialize_message(input: Self) -> Result<producer::Message, PulsarError> {
         let payload = serde_json::to_vec(&input).map_err(|e| PulsarError::Custom(e.to_string()))?;
-        Ok(producer::Message {
-            payload,
-            ..Default::default()
-        })
+        Ok(producer::Message::new(payload))
     }
 }
 

examples/round_trip.rs

@@ -16,10 +16,7 @@ struct TestData {
 impl SerializeMessage for TestData {
     fn serialize_message(input: Self) -> Result<producer::Message, PulsarError> {
         let payload = serde_json::to_vec(&input).map_err(|e| PulsarError::Custom(e.to_string()))?;
-        Ok(producer::Message {
-            payload,
-            ..Default::default()
-        })
+        Ok(producer::Message::new(payload))
     }
 }
 

pulsar-schema-confluent/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "pulsar-schema-confluent"
+version = "0.1.0"
+edition = "2021"
+license = "MIT OR Apache-2.0"
+description = "Confluent Schema Registry integration for pulsar-rs"
+repository = "https://github.com/streamnative/pulsar-rs"
+
+[dependencies]
+pulsar = { path = "..", default-features = false, features = ["tokio-runtime"] }
+async-trait = "0.1"
+schema_registry_converter = { version = "4.8", features = ["json"] }
+serde = { version = "1", features = ["derive"] }
+serde_json = "1"
+log = "0.4"
+
+[dev-dependencies]
+tokio = { version = "1", features = ["macros", "rt-multi-thread"] }