RabbitMQ Streams Client for Rust

A work-in-progress Rust client for RabbitMQ Streams

RabbitMQ Stream Client

This is a Rust client library for working with RabbitMQ Streams.

Installation

Install from crates.io

[dependencies]
rabbitmq-stream-client = "*"

Quick Start

The main access point is Environment, which is used to connect to a node.

Example

Building the environment

use rabbitmq_stream_client::Environment;
let environment = Environment::builder().build().await?;

Building the environment with TLS

use rabbitmq_stream_client::Environment;

let tls_configuration: TlsConfiguration = TlsConfiguration::builder()
.add_root_certificates(String::from(".ci/certs/ca_certificate.pem"))
.build();

// Use this configuration if you want to trust the certificates
// without providing the root certificate
let tls_configuration: TlsConfiguration = TlsConfiguration::builder()
     .trust_certificates(true)
     .build();

let environment = Environment::builder()
    .host("localhost")
    .port(5551) // specify the TLS port of the node
    .tls(tls_configuration)
    .build()

Building the environment with a load balancer

use rabbitmq_stream_client::Environment;


let environment = Environment::builder()
    .load_balancer_mode(true)
    .build()

Publishing messages

use rabbitmq_stream_client::{Environment, types::Message};
let environment = Environment::builder().build().await?;
let producer = environment.producer().name("myproducer").build("mystream").await?;
for i in 0..10 {
    producer
      .send_with_confirm(Message::builder().body(format!("message{}", i)).build())
      .await?;
}
producer.close().await?;

Consuming messages

use rabbitmq_stream_client::{Environment};
use futures::StreamExt;
use tokio::task;
use tokio::time::{sleep, Duration};
let environment = Environment::builder().build().await?;
let mut consumer = environment.consumer().build("mystream").await?;
let handle = consumer.handle();
task::spawn(async move {
        while let Some(delivery) = consumer.next().await {
            let d = delivery.unwrap();
            println!("Got message: {:#?} with offset: {}",
                     d.message().data().map(|data| String::from_utf8(data.to_vec()).unwrap()),
                     d.offset(),);
        }
    });
// wait 10 second and then close the consumer
sleep(Duration::from_secs(10)).await;
handle.close().await?;

Development

Compiling

git clone https://github.com/rabbitmq/rabbitmq-stream-rust-client .
make build

Running Tests

To run tests you need to have a running RabbitMQ Stream node with a TLS configuration. It is mandatory to use make rabbitmq-server to create a TLS configuration compatible with the tests. See the Environment TLS tests for more details.

make rabbitmq-server
make test

Running Benchmarks

make rabbitmq-server
make run-benchmark