feat(rdkafka): add statistics API (#164)

* add statistics API but leave unimplemented Signed-off-by: Runji Wang <[email protected]> * update CHANGELOG Signed-off-by: Runji Wang <[email protected]> --------- Signed-off-by: Runji Wang <[email protected]>
madsim-rs · Aug 21, 2023 · f01e3ca · f01e3ca
1 parent 5e8c331
commit f01e3ca
Show file tree

Hide file tree

Showing 4 changed files with 24 additions and 2 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -6,6 +6,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## rdkafka
+
+### Added
+
+- Add statistics API.
+- Add future producer API.
+
 ## tonic [0.3.1] - 2023-07-24
 
 ### Added

diff --git a/madsim-rdkafka/src/sim/client.rs b/madsim-rdkafka/src/sim/client.rs
@@ -1,5 +1,17 @@
+use crate::Statistics;
+
+use tracing::info;
+
 /// Client-level context.
 pub trait ClientContext: Send + Sync + 'static {
+    /// Receives the decoded statistics of the librdkafka client. To enable, the
+    /// `statistics.interval.ms` configuration parameter must be specified.
+    ///
+    /// The default implementation logs the statistics at the `info` log level.
+    fn stats(&self, statistics: Statistics) {
+        info!("Client stats: {:?}", statistics);
+    }
+
     fn rewrite_broker_addr(&self, addr: BrokerAddr) -> BrokerAddr {
         addr
     }

diff --git a/madsim-rdkafka/src/sim/mod.rs b/madsim-rdkafka/src/sim/mod.rs
@@ -8,6 +8,8 @@ pub mod message;
 pub mod metadata;
 pub mod producer;
 pub(crate) mod sim_broker;
+#[path = "../std/statistics.rs"]
+pub mod statistics;
 pub mod topic_partition_list;
 pub mod types;
 pub mod util;
@@ -16,6 +18,7 @@ pub use self::client::ClientContext;
 pub use self::config::ClientConfig;
 pub use self::message::{Message, Timestamp};
 pub use self::sim_broker::SimBroker;
+pub use self::statistics::Statistics;
 pub use self::topic_partition_list::{Offset, TopicPartitionList};
 
 // custom deserialize function for serde

diff --git a/madsim-rdkafka/src/std/statistics.rs b/madsim-rdkafka/src/std/statistics.rs
@@ -367,7 +367,7 @@ mod tests {
 
         assert_eq!(stats.brokers.len(), 1);
 
-        let broker = stats.brokers.values().into_iter().collect::<Vec<_>>()[0];
+        let broker = stats.brokers.values().collect::<Vec<_>>()[0];
 
         assert_eq!(
             broker.req,
@@ -391,7 +391,7 @@ mod tests {
     }
 
     // Example from https://github.com/edenhill/librdkafka/wiki/Statistics
-    const EXAMPLE: &'static str = r#"
+    const EXAMPLE: &str = r#"
       {
         "name": "rdkafka#producer-1",
         "client_id": "rdkafka",