stan

基于 NATS.rs 构建的 NATS 流式传输客户端包装器

示例

use nats;
use std::{io, str::from_utf8, time};

fn main() -> io::Result<()> {
    let nats_url = "nats://127.0.0.1:4222";
    let nc = nats::connect(nats_url)?;
    let sc = stan::connect(nc, "test-cluster", "rust-client-1")?;

    sc.publish("foo", "hello from rust 1")?;

    sc.subscribe("foo", Default::default())?
        .with_handler(|msg| {
            println!("sub 1 got {:?}", from_utf8(&msg.data));
            Ok(())
        });

    let sub = sc
        .subscribe(
            "foo",
            stan::SubscriptionConfig {
                queue_group: Some("queue-group-name"),
                durable_name: Some("my-durable-queue"),
                start: stan::SubscriptionStart::AllAvailable,
                ..Default::default()
            },
        )?
        .with_handler(|msg| {
            println!("sub 2 got {:?}", from_utf8(&msg.data));
            msg.ack()?;
            println!("manually acked!");
            Ok(())
        });

    for msg in sc.subscribe("foo", Default::default())?.messages() {
        println!("sub 3 got {:?}", from_utf8(&msg.data));
        msg.ack()?;
        break; // just break for the example to run
    }

    for msg in sc
        .subscribe("foo", Default::default())?
        .timeout_iter(time::Duration::from_secs(1))
    {
        println!("sub 4 got {:?}", from_utf8(&msg.data));
        msg.ack()?;
        break; // just break for the example to run
    }

    sc.publish("foo", "hello from rust 2")?;
    sc.publish("foo", "hello from rust 3")?;

    sub.unsubscribe()?;

    sc.publish("foo", "hello from rust 4")?;
    Ok(())
}

理由

我们对 NATS 的至少一次交付很感兴趣，目前的选项有 NATS 流式传输、Lightbridge 或 Jetstream。

Jetstream 是 NATS 上至少一次交付的未来，但目前仍在开发中，而 NATS 流式传输已在生产中得到过测试。

同时，NATS 团队提供了一款出色的 Rust 客户端，它也支持 Jetstream，但他们没有计划支持 NATS 流式传输（这是合理的，因为 Jetstream 是他们的主要重点）。

由于 NATS 流式传输只是在 NATS 上加了一层，因此编写了这个库来仅封装 nats.rs 客户端以处理 NATS 流式传输协议，以供那些像我们一样仍然使用 NATS 流式传输的人使用。

安装

[dependencies]
nats = "0.9.7"
stan = "0.0.16"

开发

要启动本地 nats 流式传输服务器

docker run -p 4222:4222 -p 8222:8222 nats-streaming

运行测试需要 docker。要运行测试

cargo test