Basic produce.

Author: j7nw4r

src/producer/base_producer.rs

@@ -42,6 +42,7 @@
 //! should wait and try again.
 
 use std::ffi::{CStr, CString};
+use std::fmt;
 use std::marker::PhantomData;
 use std::mem;
 use std::os::raw::c_void;
@@ -340,6 +341,19 @@ where
     _partitioner: PhantomData<Part>,
 }
 
+impl<C, Part> fmt::Debug for BaseProducer<C, Part>
+where
+    Part: Partitioner,
+    C: ProducerContext<Part>,
+{
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("BaseProducer")
+            .field("native_ptr", &self.native_ptr())
+            .field("queue", &self.queue)
+            .finish()
+    }
+}
+
 impl<C, Part> BaseProducer<C, Part>
 where
     Part: Partitioner,

tests/producer.rs

@@ -1,5 +1,9 @@
+use rdkafka::ClientConfig;
 use crate::utils::containers::KafkaContext;
 use crate::utils::logging::init_test_logger;
+use crate::utils::producer::create_base_producer;
+use crate::utils::rand::rand_test_topic;
+use rdkafka::producer::BaseRecord;
 
 mod utils;
 
@@ -14,4 +18,30 @@ pub async fn test_basic_produce() {
             kafka_context_result.unwrap_err()
         );
     };
+
+    let bootstrap_servers_result = _kafka_context.bootstrap_servers().await;
+    let Ok(bootstrap_servers) = bootstrap_servers_result else {
+        panic!("could not create bootstrap servers: {}", bootstrap_servers_result.unwrap_err());
+    };
+    let mut client_config = ClientConfig::default();
+    client_config.set("bootstrap.servers", bootstrap_servers);
+
+    let base_producer_result = create_base_producer(&client_config);
+    let Ok(base_producer) = base_producer_result else {
+        panic!(
+            "could not create based_producer: {}",
+            base_producer_result.unwrap_err()
+        );
+    };
+
+    let test_topic = rand_test_topic("testing-topic");
+    let record = BaseRecord::to(&test_topic) // destination topic
+        .key(&[1, 2, 3, 4]) // message key
+        .payload("content") // message payload
+        .partition(5);
+
+    let send_result = base_producer.send(record);
+    if send_result.is_err() {
+        panic!("could not produce record: {:?}", send_result.unwrap_err());
+    }
 }

tests/utils/mod.rs

@@ -2,6 +2,7 @@
 
 pub mod containers;
 pub mod logging;
+pub mod producer;
 pub mod rand;
 
 use std::collections::HashMap;

tests/utils/producer.rs

@@ -0,0 +1,11 @@
+use rdkafka::ClientConfig;
+use rdkafka::config::FromClientConfig;
+use rdkafka::producer::BaseProducer;
+
+pub fn create_base_producer(config: &ClientConfig) -> anyhow::Result<BaseProducer> {
+    let base_producer_result = BaseProducer::from_config(config);
+    let Ok(base_producer) = base_producer_result else {
+        anyhow::bail!("error creating base producer: {}", base_producer_result.unwrap_err())
+    };
+    Ok(base_producer)
+}