Docs in cloudevents-sdk-rdkafka

Signed-off-by: Francesco Guardiani <[email protected]>

Author: slinkydeveloper

cloudevents-sdk-rdkafka/Cargo.toml

@@ -8,14 +8,15 @@ description = "CloudEvents official Rust SDK - Kafka integration"
 documentation = "https://docs.rs/cloudevents-sdk-rdkafka"
 repository = "https://github.com/cloudevents/sdk-rust"
 readme = "README.md"
+categories = ["web-programming", "encoding"]
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
 bytes = "^0.5"
 cloudevents-sdk = { version = "0.2.0", path = ".." }
 lazy_static = "1.4.0"
-rdkafka = { version = "^0.24", features = ["cmake-build"] }
+rdkafka = { version = "^0.24", default-features = false }
 
 
 [dev-dependencies]

cloudevents-sdk-rdkafka/src/kafka_consumer_record.rs

@@ -134,8 +134,10 @@ pub fn record_to_event(msg: &impl Message) -> Result<Event> {
     MessageDeserializer::into_event(ConsumerRecordDeserializer::new(msg)?)
 }
 
-/// Extension Trait for [`Message`] which acts as a wrapper for the function [`record_to_event()`]
-pub trait MessageExt {
+/// Extension Trait for [`Message`] which acts as a wrapper for the function [`record_to_event()`].
+///
+/// This trait is sealed and cannot be implemented for types outside of this crate.
+pub trait MessageExt: private::Sealed {
     /// Generates [`Event`] from [`BorrowedMessage`].
     fn to_event(&self) -> Result<Event>;
 }
@@ -152,6 +154,13 @@ impl MessageExt for OwnedMessage {
     }
 }
 
+mod private {
+    // Sealing the MessageExt
+    pub trait Sealed {}
+    impl Sealed for rdkafka::message::OwnedMessage {}
+    impl Sealed for rdkafka::message::BorrowedMessage<'_> {}
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

cloudevents-sdk-rdkafka/src/kafka_producer_record.rs

@@ -88,7 +88,9 @@ impl StructuredSerializer<MessageRecord> for MessageRecord {
 }
 
 /// Extension Trait for [`BaseRecord`] that fills the record with a [`MessageRecord`].
-pub trait BaseRecordExt<'a, K: ToBytes + ?Sized> {
+///
+/// This trait is sealed and cannot be implemented for types outside of this crate.
+pub trait BaseRecordExt<'a, K: ToBytes + ?Sized>: private::Sealed {
     /// Fill this [`BaseRecord`] with a [`MessageRecord`].
     fn message_record(
         self,
@@ -112,7 +114,9 @@ impl<'a, K: ToBytes + ?Sized> BaseRecordExt<'a, K> for BaseRecord<'a, K, Vec<u8>
 }
 
 /// Extension Trait for [`FutureRecord`] that fills the record with a [`MessageRecord`].
-pub trait FutureRecordExt<'a, K: ToBytes + ?Sized> {
+///
+/// This trait is sealed and cannot be implemented for types outside of this crate.
+pub trait FutureRecordExt<'a, K: ToBytes + ?Sized>: private::Sealed {
     /// Fill this [`FutureRecord`] with a [`MessageRecord`].
     fn message_record(self, message_record: &'a MessageRecord) -> FutureRecord<'a, K, Vec<u8>>;
 }
@@ -128,3 +132,10 @@ impl<'a, K: ToBytes + ?Sized> FutureRecordExt<'a, K> for FutureRecord<'a, K, Vec
         self
     }
 }
+
+mod private {
+    // Sealing the FutureRecordExt and BaseRecordExt
+    pub trait Sealed {}
+    impl <K: rdkafka::message::ToBytes + ?Sized, V: rdkafka::message::ToBytes> Sealed for rdkafka::producer::FutureRecord<'_, K, V> {}
+    impl <K: rdkafka::message::ToBytes + ?Sized, V: rdkafka::message::ToBytes> Sealed for rdkafka::producer::BaseRecord<'_, K, V> {}
+}

cloudevents-sdk-rdkafka/src/lib.rs

@@ -10,17 +10,16 @@
 //! use rdkafka::util::Timeout;
 //! use cloudevents_sdk_rdkafka::{MessageRecord, FutureRecordExt};
 //!
-//! # async fn produce(producer: &FutureProducer, event: Event) {
-//! let message_record = MessageRecord::from_event(event)
-//!   .expect("error while serializing the event");
+//! # async fn produce(producer: &FutureProducer, event: Event) -> Result<(), Box<dyn std::error::Error>> {
+//! let message_record = MessageRecord::from_event(event)?;
 //!
 //! producer.send(
 //!     FutureRecord::to("topic")
 //!         .key("some_event")
 //!         .message_record(&message_record),
 //!     Timeout::Never
 //! ).await;
-//!
+//! # Ok(())
 //! # }
 //!
 //! ```
@@ -32,22 +31,25 @@
 //! use cloudevents_sdk_rdkafka::MessageExt;
 //! use futures::StreamExt;
 //!
-//! # async fn consume(consumer: StreamConsumer<DefaultConsumerContext>) {
+//! # async fn consume(consumer: StreamConsumer<DefaultConsumerContext>) -> Result<(), Box<dyn std::error::Error>> {
 //! let mut message_stream = consumer.start();
 //!
 //! while let Some(message) = message_stream.next().await {
 //!     match message {
 //!         Err(e) => println!("Kafka error: {}", e),
 //!         Ok(m) => {
-//!             let event = m.to_event().expect("error while deserializing record to CloudEvent");
-//!             println!("Received Event: {:#?}", event);
-//!             consumer.commit_message(&m, CommitMode::Async).unwrap();
+//!             let event = m.to_event()?;
+//!             println!("Received Event: {}", event);
+//!             consumer.commit_message(&m, CommitMode::Async)?;
 //!         }
 //!     };
 //! }
+//! # Ok(())
 //! # }
 //! ```
 
+#![deny(broken_intra_doc_links)]
+
 #[macro_use]
 mod headers;
 mod kafka_consumer_record;