6 releases (3 breaking)

0.3.0 Feb 3, 2021
0.2.1 May 15, 2020
0.2.0 Apr 14, 2020
0.1.0 Feb 7, 2019
0.0.2 Oct 30, 2018

#21 in Asynchronous

Download history 1168/week @ 2020-11-07 954/week @ 2020-11-14 727/week @ 2020-11-21 816/week @ 2020-11-28 968/week @ 2020-12-05 947/week @ 2020-12-12 883/week @ 2020-12-19 813/week @ 2020-12-26 1219/week @ 2021-01-02 1124/week @ 2021-01-09 1156/week @ 2021-01-16 1234/week @ 2021-01-23 1023/week @ 2021-01-30 901/week @ 2021-02-06 655/week @ 2021-02-13 364/week @ 2021-02-20

4,220 downloads per month
Used in 2 crates

MIT/Apache

46KB
969 lines

TMQ - Rust ZeroMQ bindings for Tokio

This crate bridges Tokio and ZeroMQ to allow for ZeroMQ in the async world.

Changelog

0.3.0 - Tokio 1.0 Support

0.3.0 adds support for tokio 1.0 thanks to YushiOMOTE!

Currently Implemented Sockets

  • Request/Reply
  • Publish/Subscribe
  • Dealer/Router
  • Push/Pull

Examples

Please see the examples directory for a full set of examples. They are paired together based upon the socket types.

Usage

Usage is made to be simple, but opinionated. See the examples for working code, but in general, you need to import tokio and tmq::*

Publish

To publish messages to all connected subscribers, you can use the publish function:

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

Subscribe

a subscribe socket is a Stream that reads in values from a publish socket. You specify the filter prefix using the subscribe method, using "" for all messages.

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(())
}

Dependencies

~4MB
~77K SLoC