Implementation:Fede1024 Rust rdkafka AsyncStdRuntime Adapter
| Knowledge Sources | |
|---|---|
| Domains | Async_Runtime, Runtime_Abstraction, Integration |
| Last Updated | 2026-02-07 19:00 GMT |
Overview
Example implementation of the AsyncRuntime trait that integrates rust-rdkafka with the async-std runtime, demonstrating pluggable async runtime support.
Description
The runtime_async_std example defines a custom AsyncStdRuntime struct that implements rdkafka::util::AsyncRuntime. It maps spawn to async_std::task::spawn and delay_for to async_std::task::sleep, using Pin<Box<dyn Future<Output = ()> + Send>> as the Delay associated type. The main function, annotated with #[async_std::main], sends a single message through a FutureProducer<_, AsyncStdRuntime> and receives it with a StreamConsumer<_, AsyncStdRuntime>, verifying the roundtrip works under async-std.
Usage
Use this as a reference when integrating rust-rdkafka with the async-std runtime instead of the default Tokio. Copy the AsyncStdRuntime struct and trait implementation into your project, then parameterize producers and consumers with it.
Code Reference
Source Location
- Repository: Fede1024_Rust_rdkafka
- File: examples/runtime_async_std.rs
- Lines: 1-108
Signature
pub struct AsyncStdRuntime;
impl AsyncRuntime for AsyncStdRuntime {
type Delay = Pin<Box<dyn Future<Output = ()> + Send>>;
fn spawn<T>(task: T)
where
T: Future<Output = ()> + Send + 'static,
{
async_std::task::spawn(task);
}
fn delay_for(duration: Duration) -> Self::Delay {
Box::pin(async_std::task::sleep(duration))
}
}
Import
use rdkafka::util::AsyncRuntime;
use std::future::Future;
use std::pin::Pin;
use std::time::Duration;
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| task | impl Future<Output = ()> + Send + 'static | Yes | Async task to spawn on async-std |
| duration | Duration | Yes | Sleep duration for delay_for |
Outputs
| Name | Type | Description |
|---|---|---|
| Delay | Pin<Box<dyn Future<Output = ()> + Send>> | Boxed sleep future from async-std |
Usage Examples
Using AsyncStdRuntime with Producer and Consumer
use rdkafka::config::ClientConfig;
use rdkafka::consumer::StreamConsumer;
use rdkafka::producer::FutureProducer;
// Define the runtime adapter
pub struct AsyncStdRuntime;
impl rdkafka::util::AsyncRuntime for AsyncStdRuntime {
type Delay = Pin<Box<dyn Future<Output = ()> + Send>>;
fn spawn<T: Future<Output = ()> + Send + 'static>(task: T) {
async_std::task::spawn(task);
}
fn delay_for(duration: Duration) -> Self::Delay {
Box::pin(async_std::task::sleep(duration))
}
}
// Create producer with async-std runtime
let producer: FutureProducer<_, AsyncStdRuntime> = ClientConfig::new()
.set("bootstrap.servers", "localhost:9092")
.create()
.expect("Producer creation failed");
// Create consumer with async-std runtime
let consumer: StreamConsumer<_, AsyncStdRuntime> = ClientConfig::new()
.set("bootstrap.servers", "localhost:9092")
.set("group.id", "my-group")
.create()
.expect("Consumer creation failed");