Client Rust

Skill: Rust Client — danube-client

Prerequisites

• Rust toolchain installed (rustc, cargo)

Installation

cargo add danube-client
cargo add tokio --features full
cargo add serde_json  # for JSON messages

Client Creation

use danube_client::DanubeClient;

#[tokio::main]
async fn main() -> Result> {
    let client = DanubeClient::builder()
        .service_url("http://127.0.0.1:6650")
        .build()
        .await?;

    Ok(())
}

Producer

Basic Producer

let mut producer = client
    .new_producer()
    .with_topic("/default/my-topic")
    .with_name("my-producer")
    .build()?;

producer.create().await?;

let message_id = producer
    .send("Hello Danube!".as_bytes().to_vec(), None)
    .await?;

println!("Sent message ID: {}", message_id);

Send with Attributes

use std::collections::HashMap;

let mut attributes = HashMap::new();
attributes.insert("source".to_string(), "app-1".to_string());
attributes.insert("priority".to_string(), "high".to_string());

let message_id = producer
    .send(b"Important message".to_vec(), Some(attributes))
    .await?;

Partitioned Producer

let mut producer = client
    .new_producer()
    .with_topic("/default/high-throughput")
    .with_name("partitioned-producer")
    .with_partitions(3)
    .build()?;

producer.create().await?;

Reliable Dispatch Producer

let mut producer = client
    .new_producer()
    .with_topic("/default/critical-events")
    .with_name("reliable-producer")
    .with_reliable_dispatch()
    .build()?;

producer.create().await?;

Schema-Linked Producer

let mut producer = client
    .new_producer()
    .with_topic("/default/events")
    .with_name("schema-producer")
    .with_schema_subject("event-schema")
    .build()?;

producer.create().await?;

let event = serde_json::to_vec(&serde_json::json!({
    "user_id": "user-123",
    "event": "login",
    "timestamp": 1234567890
}))?;

producer.send(event, None).await?;

Consumer

Basic Consumer (Exclusive)

use danube_client::SubType;

let mut consumer = client
    .new_consumer()
    .with_topic("/default/my-topic")
    .with_consumer_name("my-consumer")
    .with_subscription("my-subscription")
    .with_subscription_type(SubType::Exclusive)
    .build()?;

consumer.subscribe().await?;

let mut stream = consumer.receive().await?;

while let Some(message) = stream.recv().await {
    let payload = String::from_utf8_lossy(&message.payload);
    println!("Received: {}", payload);
    consumer.ack(&message).await?;
}

Subscription Types

// Exclusive: single consumer, ordered
.with_subscription_type(SubType::Exclusive)

// Shared: load balanced across consumers
.with_subscription_type(SubType::Shared)

// Failover: active/standby
.with_subscription_type(SubType::FailOver)

// Key-Shared: per-key ordering with parallelism
.with_subscription_type(SubType::KeyShared)

Accessing Message Fields

while let Some(message) = stream.recv().await {
    let payload = String::from_utf8_lossy(&message.payload);
    println!("Payload: {}", payload);

    if let Some(attributes) = &message.attributes {
        for (key, value) in attributes {
            println!("  {}: {}", key, value);
        }
    }

    consumer.ack(&message).await?;
}

NACK with Retry

while let Some(message) = stream.recv().await {
    match process(&message) {
        Ok(_) => consumer.ack(&message).await?,
        Err(e) => {
            consumer.nack(
                &message,
                Some(1000),                           // retry after 1s
                Some(format!("processing failed: {}", e)),
            ).await?;
        }
    }
}

Partitioned Consumer

// Consumer auto-discovers all partitions
let mut consumer = client
    .new_consumer()
    .with_topic("/default/my-topic")  // Parent topic name
    .with_consumer_name("partition-consumer")
    .with_subscription("partition-sub")
    .with_subscription_type(SubType::Exclusive)
    .build()?;

consumer.subscribe().await?;
// Automatically receives from all partitions

Key-Shared

Producer with Routing Keys

// All "payment" messages go to the same consumer
producer.send_with_key(b"Payment for #1001".to_vec(), None, "payment").await?;

// "shipping" goes to (potentially) a different consumer
producer.send_with_key(b"Order #1001 shipped".to_vec(), None, "shipping").await?;

Key-Shared Consumer

let mut consumer = client
    .new_consumer()
    .with_topic("/default/orders")
    .with_consumer_name("worker_1")
    .with_subscription("orders_sub")
    .with_subscription_type(SubType::KeyShared)
    .build()?;

consumer.subscribe().await?;
let mut stream = consumer.receive().await?;

while let Some(message) = stream.recv().await {
    let key = message.routing_key.as_deref().unwrap_or("");
    let payload = String::from_utf8_lossy(&message.payload);
    println!("key={:= 2)
.with_schema_min_version("user-events", 2)

Complete Example: Simple Producer & Consumer

use danube_client::{DanubeClient, SubType};

#[tokio::main]
async fn main() -> Result> {
    let client = DanubeClient::builder()
        .service_url("http://127.0.0.1:6650")
        .build()
        .await?;

    // Producer
    let mut producer = client
        .new_producer()
        .with_topic("/default/test_topic")
        .with_name("test_producer")
        .build()?;
    producer.create().await?;

    // Send messages
    for i in 0..5 {
        let msg = format!("Message {}", i);
        let id = producer.send(msg.as_bytes().to_vec(), None).await?;
        println!("Sent: {} (ID: {})", msg, id);
    }

    // Consumer
    let mut consumer = client
        .new_consumer()
        .with_topic("/default/test_topic")
        .with_consumer_name("test_consumer")
        .with_subscription("test_sub")
        .with_subscription_type(SubType::Exclusive)
        .build()?;
    consumer.subscribe().await?;

    let mut stream = consumer.receive().await?;
    while let Some(message) = stream.recv().await {
        println!("Received: {}", String::from_utf8_lossy(&message.payload));
        consumer.ack(&message).await?;
    }

    Ok(())
}

Reference Examples

Working examples in the danube repository:

• simple_producer_consumer.rs — basic produce/consume
• json_producer.rs / json_consumer.rs — JSON schema
• avro_producer.rs / avro_consumer.rs — Avro schema
• key_shared_producer.rs / key_shared_consumer.rs — Key-Shared subscription
• key_shared_filtered_consumer.rs — Key filtering
• partitions_producer.rs / partitions_consumer.rs — Partitioned topics
• reliable_dispatch_producer.rs / reliable_dispatch_consumer.rs — Reliable delivery
• schema_evolution.rs — Schema versioning

Troubleshooting

"no partitions found" on consumer.subscribe()

The topic must exist before calling consumer.subscribe(). If the topic doesn't exist, the consumer will fail with "no partitions found". Create the topic first:

danube-admin topics create /default/my-topic
# or with reliable delivery:
danube-admin topics create /default/my-topic --dispatch-strategy reliable

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: danube-messaging
• Source: danube-messaging/danube-agent-skills
• License: Apache-2.0

Install and usage instructions live in the source repository linked above.