TMQ - 为 Tokio 提供的 Rust ZeroMQ 绑定

此包将 Tokio 和 ZeroMQ 连接起来，以便在异步世界中使用 ZeroMQ。

变更日志

0.3.1 - Multipart 迭代 Mut

为 Multipart 添加了 iter_mut() 方法

0.3.0 - 支持 Tokio 1.0

0.3.0 添加了对 tokio 1.0 的支持，感谢 YushiOMOTE！

已实现套接字

• 请求/回复
• 发布/订阅
• 经销商/路由器
• 推送/拉取

示例

请参阅 examples 目录中的完整示例。它们根据套接字类型成对出现。

用法

用法简单，但具有明确的观点。请参阅示例中的有效代码，但通常您需要导入 tokio 和 tmq::*

发布

要向所有连接的订阅者发布消息，可以使用 publish 函数

use tmq::{publish, Context, Result};

use futures::SinkExt;
use log::info;
use std::env;
use std::time::Duration;
use tokio::time::delay_for;

#[tokio::main]
async fn main() -> Result<()> {

    let mut socket = publish(&Context::new()).bind("tcp://127.0.0.1:7899")?;

    let mut i = 0;

    loop {
        i += 1;

        socket
            .send(vec!["topic", &format!("Broadcast #{}", i)])
            .await?;

        delay_for(Duration::from_secs(1)).await;
    }
}

订阅

订阅套接字是从发布套接字读取值的 Stream。您可以使用 subscribe 方法指定过滤器前缀，使用 "" 用于所有消息。

use futures::StreamExt;

use tmq::{subscribe, Context, Result};

use std::env;

#[tokio::main]
async fn main() -> Result<()> {

    let mut socket = subscribe(&Context::new())
        .connect("tcp://127.0.0.1:7899")?
        .subscribe(b"topic")?;

    while let Some(msg) = socket.next().await {
        println!(
            "Subscribe: {:?}",
            msg?.iter()
                .map(|item| item.as_str().unwrap_or("invalid text"))
                .collect::<Vec<&str>>()
        );
    }
    Ok(())
}