RabbitMQ Streams Client for Rust

RabbitMQ Stream客户端

这是一个用于与RabbitMQ Streams交互的Rust客户端库。

安装

从crates.io安装

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

快速入门

主要访问点是Environment，用于连接到节点。

示例

构建环境

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

使用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()

使用负载均衡器构建环境

use rabbitmq_stream_client::Environment;


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

发布消息

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?;

消费消息

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?;

开发

编译

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

运行测试

要运行测试，您需要一个具有TLS配置的运行中的RabbitMQ Stream节点。必须使用make rabbitmq-server来创建与测试兼容的TLS配置。有关详细信息，请参阅Environment的TLS测试。

make rabbitmq-server
make test

运行基准测试

make rabbitmq-server
make run-benchmark