Streamline Rust Client

Native Rust client library for Streamline streaming platform.

Features

• Async/await with Tokio
• Admin client (topic management, consumer groups via HTTP REST API)
• SQL query support
• Connection pooling
• Automatic reconnection with exponential backoff
• Producer and consumer APIs
• Compression support (LZ4, Zstd, Snappy — feature-gated)
• TLS/mTLS support (feature-gated via tls)
• SASL authentication (PLAIN, SCRAM)
• OpenTelemetry tracing (feature-gated via telemetry)
• Builder pattern configuration

Installation

Add to your Cargo.toml:

[dependencies]
streamline-client = "0.1"

With optional features:

[dependencies]
streamline-client = { version = "0.1", features = ["compression-lz4", "tls"] }

With OpenTelemetry Tracing

[dependencies]
streamline-client = { version = "0.1", features = ["telemetry"] }

OpenTelemetry Tracing

The SDK supports optional OpenTelemetry tracing gated behind the telemetry Cargo feature. When enabled, produce and consume operations can be wrapped in OTel-compatible spans. When disabled, all tracing functions compile to zero-overhead no-ops.

Usage

use streamline_client::telemetry;
use streamline_client::Headers;

// Trace a produce operation
let metadata = telemetry::trace_produce("orders", || async {
    producer.send("orders", "key".to_string(), "value".to_string(), Headers::new()).await
}).await?;

// Trace a consume operation
let records = telemetry::trace_consume("events", || async {
    consumer.poll(Duration::from_millis(100)).await
}).await?;

// Trace individual record processing
for record in &records {
    telemetry::trace_process(
        &record.topic,
        record.partition,
        record.offset,
        || async { process(&record) }
    ).await;
}

// Inject context into headers for propagation
let mut headers = Headers::new();
telemetry::inject_context(&mut headers);

Span Conventions

Attribute	Value
Span name	{topic} {operation} (e.g., "orders produce")
messaging.system	streamline
messaging.destination.name	Topic name
messaging.operation	produce, consume, or process
Span kind	PRODUCER for produce, CONSUMER for consume

Trace context is propagated through message headers using W3C TraceContext format.

Quick Start

Producer

use streamline_client::{Streamline, StreamlineConfig};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Create client
    let client = Streamline::builder()
        .bootstrap_servers("localhost:9092")
        .build()
        .await?;

    // Produce a message
    let metadata = client.produce("my-topic", "key", "Hello, Streamline!").await?;
    println!("Produced to partition {} at offset {}", metadata.partition, metadata.offset);

    Ok(())
}

Transactions

let mut producer = client.producer::<String, String>();
producer.begin_transaction()?;

producer.send_transactional("orders", record1).await?;
producer.send_transactional("orders", record2).await?;

let results = producer.commit_transaction().await?;
// Or: producer.abort_transaction()?;

Note: Transactions use client-side buffering. Messages are collected locally and sent as a batch on commit, providing all-or-nothing delivery at the client level.

Consumer

use streamline_client::{Streamline, ConsumerConfig};
use std::time::Duration;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = Streamline::builder()
        .bootstrap_servers("localhost:9092")
        .build()
        .await?;

    let consumer = client.consumer::<String, String>("my-topic")
        .group_id("my-group")
        .auto_offset_reset("earliest")
        .build()
        .await?;

    consumer.subscribe().await?;

    loop {
        let records = consumer.poll(Duration::from_millis(100)).await?;
        for record in records {
            println!("Received: {} at offset {}", record.value, record.offset);
        }
    }
}

Producer with Headers

use streamline_client::Headers;

let headers = Headers::builder()
    .add("source", "my-app")
    .add("version", "1.0")
    .build();

client.produce_with_headers("my-topic", "key", "value", headers).await?;

Batch Production

use streamline_client::ProducerRecord;

let records = vec![
    ProducerRecord::new("key1", "value1"),
    ProducerRecord::new("key2", "value2"),
    ProducerRecord::new("key3", "value3"),
];

let results = client.produce_batch("my-topic", records).await?;

Configuration

Client Configuration

let client = Streamline::builder()
    .bootstrap_servers("localhost:9092")
    .connection_pool_size(4)
    .connect_timeout(Duration::from_secs(30))
    .request_timeout(Duration::from_secs(30))
    .build()
    .await?;

Producer Configuration

let producer = client.producer::<String, String>()
    .batch_size(16384)
    .linger_ms(1)
    .compression("lz4")
    .retries(3)
    .build();

Consumer Configuration

let consumer = client.consumer::<String, String>("my-topic")
    .group_id("my-group")
    .auto_offset_reset("earliest")
    .enable_auto_commit(true)
    .max_poll_records(500)
    .session_timeout(Duration::from_secs(30))
    .build()
    .await?;

Error Handling

use streamline_client::{StreamlineError, ErrorKind};

match client.produce("my-topic", "key", "value").await {
    Ok(metadata) => println!("Success: {:?}", metadata),
    Err(StreamlineError { kind: ErrorKind::TopicNotFound, hint, .. }) => {
        eprintln!("Topic not found. {}", hint.unwrap_or_default());
    }
    Err(e) => eprintln!("Error: {}", e),
}

TLS Configuration

let client = Streamline::builder()
    .bootstrap_servers("localhost:9092")
    .tls_config(TlsConfig::builder()
        .ca_cert("ca.crt")
        .client_cert("client.crt")
        .client_key("client.key")
        .build()?)
    .build()
    .await?;

Testing

With Testcontainers (recommended)

The streamline-testcontainers crate provides a Testcontainers module that spins up a disposable Streamline server inside your tests -- no manual Docker setup required.

[dev-dependencies]
streamline-testcontainers = "0.2"
testcontainers = "0.23"
tokio = { version = "1", features = ["full"] }

use streamline_testcontainers::{StreamlineImage, bootstrap_servers};
use testcontainers::runners::AsyncRunner;

#[tokio::test]
async fn test_produce() {
    let container = StreamlineImage::default().start().await.unwrap();
    let bs = bootstrap_servers(&container).await.unwrap();
    // Use `bs` with any Kafka client ...
}

See the testcontainers README for full documentation.

With Docker Compose

Alternatively, start a local Streamline server manually:

docker compose -f docker-compose.test.yml up -d

Run tests:

cargo test

API Reference

Client

Method	Description
Streamline::builder()	Create a new client builder
client.produce(topic, key, value).await	Send a message
client.produce_with_headers(topic, key, value, headers).await	Send with headers
client.producer()	Create a producer instance
client.consumer(topic)	Create a consumer builder
client.is_healthy().await	Check health status
client.config()	Get client configuration

Producer

Method	Description
producer.send(topic, key, value, headers).await	Send a message
producer.send_batch(topic, records).await	Send a batch of messages
producer.flush().await	Flush buffered messages
producer.config()	Get producer configuration

Consumer

Method	Description
consumer.subscribe().await	Subscribe to topic
consumer.poll(timeout).await	Poll for messages
consumer.commit().await	Commit offsets synchronously
consumer.commit_async().await	Commit offsets asynchronously
consumer.seek_to_beginning().await	Seek to start
consumer.seek_to_end().await	Seek to end
consumer.seek(partition, offset).await	Seek to specific offset
consumer.position(partition).await	Get current position
consumer.pause()	Pause consuming
consumer.resume()	Resume consuming

Admin (Kafka Protocol)

Method	Description
admin.create_topic(config).await	Create a topic
admin.delete_topic(name).await	Delete a topic
admin.list_topics().await	List all topics
admin.describe_topic(name).await	Get topic details
admin.list_brokers().await	List cluster brokers
admin.list_consumer_groups().await	List consumer groups
admin.describe_consumer_group(id).await	Describe a group
admin.delete_consumer_group(id).await	Delete a group

HTTP Admin (Cluster, Lag, Inspect, Metrics)

The HttpAdmin communicates with the Streamline HTTP REST API (port 9094) for operations not available via Kafka protocol.

use streamline_client::HttpAdmin;

let admin = HttpAdmin::new("http://localhost:9094");

// Cluster overview
let cluster = admin.cluster_info().await?;
println!("Cluster: {}, Brokers: {}", cluster.cluster_id, cluster.brokers.len());

// Consumer group lag monitoring
let lag = admin.consumer_group_lag("my-group").await?;
println!("Total lag: {}", lag.total_lag);
for p in &lag.partitions {
    println!("  {}:{} lag={}", p.topic, p.partition, p.lag);
}

// Message inspection
let messages = admin.inspect_messages("events", 0, None, 10).await?;
for m in &messages {
    println!("offset={} value={}", m.offset, m.value);
}

// Latest messages
let latest = admin.latest_messages("events", 5).await?;

// Server metrics
let metrics = admin.metrics_history().await?;

Requirements

• Rust 1.80 or later
• Streamline server 0.2.0 or later

Configuration Reference

Client

Parameter	Default	Description
bootstrap_servers	"localhost:9092"	Comma-separated broker addresses
connection_pool_size	4	Number of connections per broker
connect_timeout	10s	Connection timeout (Duration)
request_timeout	30s	Request timeout (Duration)

Producer

Parameter	Default	Description
batch_size	16384	Maximum batch size in bytes
linger_ms	0	Batch linger time in milliseconds
compression	None	Compression: None, Gzip, Snappy, Lz4, Zstd
retries	3	Retries on transient failures
acks	One	Acknowledgment: None, One, All
idempotent	false	Enable exactly-once semantics

Consumer

Parameter	Default	Description
group_id	(required)	Consumer group identifier
auto_offset_reset	Latest	Start position: Earliest, Latest
enable_auto_commit	true	Automatically commit offsets
max_poll_records	500	Maximum records per poll
session_timeout	30s	Session timeout (Duration)

TLS (feature: tls)

Parameter	Default	Description
ca_cert_path	—	Path to CA certificate (PEM)
client_cert_path	—	Client certificate for mTLS (PEM)
client_key_path	—	Client private key for mTLS (PEM)

Circuit Breaker

Protect your application from cascading failures when the Streamline server is unresponsive:

use streamline_client::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig};
use std::time::Duration;

let cb = CircuitBreaker::new(CircuitBreakerConfig::default()
    .failure_threshold(5)        // Open after 5 consecutive failures
    .success_threshold(2)        // Close after 2 half-open successes
    .open_timeout(Duration::from_secs(30)));

if cb.check() {
    match client.produce("events", "key", "value").await {
        Ok(_) => cb.record_success(),
        Err(e) => {
            cb.record_failure();
            return Err(e);
        }
    }
}

When the circuit is open, check() returns false and operations should be skipped. See the Circuit Breaker guide for details.

Examples

The examples/ directory contains runnable examples:

Example	Description
producer.rs	Produce messages
consumer.rs	Consume messages
query_usage.rs	SQL analytics with the embedded query engine
schema_registry.rs	Schema registration and validation
circuit_breaker.rs	Resilient production with circuit breaker
security.rs	TLS and SASL authentication

Run any example:

cargo run --example producer
cargo run --example circuit_breaker

Moonshot Features

⚠️ Experimental — These features require Streamline server 0.3.0+ with moonshot feature flags enabled. Enable with: streamline-client = { version = "0.1", features = ["moonshot"] }

Semantic Search

Query topics by meaning instead of offset. Requires a topic created with semantic.embed=true.

let results = client.search("logs.app", "payment failure", 10).await?;
for hit in &results {
    println!("[p{}] offset={} score={:.2}", hit.partition, hit.offset, hit.score);
}

Attestation Verification

Verify cryptographic provenance attestations attached to records by data contracts.

use streamline_client::StreamlineVerifier;

let verifier = StreamlineVerifier::new(&public_key_bytes)?;
let result = verifier.verify(&record)?;
println!("Verified: {}, Producer: {}", result.verified, result.producer_id);

Agent Memory (MCP)

Use Streamline as persistent memory for AI agents via the MCP protocol.

use streamline_client::MemoryClient;

let memory = MemoryClient::new("http://localhost:9094/mcp/v1");
memory.remember("user prefers dark mode", &["preferences"]).await?;
let results = memory.recall("user preferences", 5).await?;

Branched Streams

Create topic branches for replay, A/B testing, or counterfactual analysis.

let branch = admin.create_branch("events", "experiment-v2").await?;
let consumer = client.consumer::<String, String>(&branch.topic)
    .group_id("branch-group")
    .build()
    .await?;

Contributing

Contributions are welcome! Please see the organization contributing guide for guidelines.

License

Apache 2.0

Security

To report a security vulnerability, please email security@streamline.dev. Do not open a public issue.

See the Security Policy for details.