add send_batch method to kafka mod

crates/mod-kafka/Cargo.toml

@@ -4,9 +4,11 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
-anyhow = {workspace=true}
+anyhow.workspace=true
 config = {path="../config"}
 duration-serde = {path="../duration-serde"}
+futures.workspace=true
 mlua = {workspace=true, features=["vendored", "lua54", "async", "send", "serialize"]}
-rdkafka = {workspace=true}
-serde = {workspace=true}
+rdkafka.workspace=true
+serde.workspace=true
+tokio.workspace=true

crates/mod-kafka/src/lib.rs

@@ -1,4 +1,6 @@
 use config::{any_err, get_or_create_sub_module};
+use futures::stream::FuturesOrdered;
+use futures::StreamExt;
 use mlua::prelude::LuaUserData;
 use mlua::{Lua, LuaSerdeExt, UserDataMethods, Value};
 use rdkafka::message::{Header, OwnedHeaders};
@@ -94,6 +96,66 @@ impl LuaUserData for Producer {
             Ok((partition, offset))
         });
 
+        methods.add_async_method("send_batch", |lua, this, values: Vec<Value>| async move {
+            let mut tasks = FuturesOrdered::new();
+            let producer = this.get_producer()?;
+
+            for value in values {
+                let record: Record = lua.from_value(value)?;
+
+                let headers = if record.headers.is_empty() {
+                    None
+                } else {
+                    let mut headers = OwnedHeaders::new();
+                    for (key, v) in &record.headers {
+                        headers = headers.insert(Header {
+                            key,
+                            value: Some(v),
+                        });
+                    }
+                    Some(headers)
+                };
+
+                let producer = producer.clone();
+
+                tasks.push_back(tokio::spawn(async move {
+                    producer
+                        .send(
+                            FutureRecord {
+                                topic: &record.topic,
+                                partition: record.partition,
+                                payload: record.payload.as_ref(),
+                                key: record.key.as_ref(),
+                                headers,
+                                timestamp: None,
+                            },
+                            Timeout::After(record.timeout.unwrap_or(Duration::from_secs(60))),
+                        )
+                        .await
+                }));
+            }
+
+            let failed_indexes = lua.create_table()?;
+            let errors = lua.create_table()?;
+            let mut index = 1;
+
+            while let Some(result) = tasks.next().await {
+                match result {
+                    Ok(Ok(_)) => {}
+                    Ok(Err((error, _msg))) => {
+                        failed_indexes.push(index)?;
+                        errors.push(format!("{error:#}"))?;
+                    }
+                    Err(error) => {
+                        failed_indexes.push(index)?;
+                        errors.push(format!("{error:#}"))?;
+                    }
+                }
+                index += 1;
+            }
+            Ok((failed_indexes, errors))
+        });
+
         methods.add_method("close", |_lua, this, _: ()| {
             this.producer.lock().unwrap().take();
             Ok(())

docs/reference/kumo.kafka/_index.md

@@ -1,4 +1,4 @@
-# Module `kumo.amqp`
+# Module `kumo.kafka`
 
 This module provides Apache Kafka client functionality.
 

docs/reference/kumo.kafka/build_producer.md

@@ -40,6 +40,43 @@ producer:send {
 }
 ```
 
+### client:send_batch({PARAMS})
+
+{{ since('dev') }}
+
+Sends a batch of messages. `PARAMS` is a table of object style table with the
+following keys:
+
+* `topic` - required string; the name of the queue to which to send the message
+* `payload` - required string; the message to send
+* `timeout` - how long to wait for a response.
+
+The result from send_batch is a tuple of tables: local failed_items, errors = producer:send_batch(...).
+
+```lua
+local producer = kumo.kafka.build_producer {
+  ['bootstrap.servers'] = 'localhost:9092',
+}
+
+local failed_items, errors = producer:send_batch {{
+  topic = 'my.topic',
+  payload = 'payload 1',
+  timeout = '1 minute',
+},
+{
+  topic = 'my.other.topic',
+  payload = 'payload 2',
+  timeout = '1 minute',
+}}
+if #failed_items > 0 then
+   -- some items failed
+   for i, item_idx in ipairs(failed_items) do
+      local error = errors[i]
+      print(string.format("item idx %d failed: %s", item_idx, error))
+   end
+end
+```
+
 ### client:close()
 
 {{since('2024.09.02-c5476b89')}}