#nsq #tokio #client

tokio-nsq

A Rust NSQ client built on Tokio. Tokio NSQ aims to be a feature complete NSQ client implementation.

27 releases (10 breaking)

0.11.0 Aug 26, 2020
0.9.2 Aug 24, 2020
0.5.1 Jul 30, 2020

#240 in Asynchronous

Download history 2/week @ 2020-10-31 87/week @ 2020-11-07 21/week @ 2020-11-14 67/week @ 2020-11-21 30/week @ 2020-11-28 96/week @ 2020-12-05 45/week @ 2020-12-12 23/week @ 2020-12-19 52/week @ 2021-01-02 16/week @ 2021-01-09 28/week @ 2021-01-16 1/week @ 2021-01-23 2/week @ 2021-01-30 27/week @ 2021-02-06 203/week @ 2021-02-13

209 downloads per month

BSD-3-Clause

74KB
2K SLoC

Tokio NSQ

GitHub Actions crates.io

A Rust NSQ client built on Tokio. Tokio NSQ aims to be a feature complete NSQ client implementation.

Tokio NSQ is available as a cargo package, and API documentation is available on docs.rs.

Versioning

This project follows strict semantic versioning. While pre 1.0.0 breaking changes have only a minor version bump.

Basic consumer example

let topic   = NSQTopic::new("names").unwrap();
let channel = NSQChannel::new("first").unwrap();

let mut addresses = HashSet::new();
addresses.insert("http://127.0.0.1:4161".to_string());

let mut consumer = NSQConsumerConfig::new(topic, channel)
    .set_max_in_flight(15)
    .set_sources(
        NSQConsumerConfigSources::Lookup(
            NSQConsumerLookupConfig::new().set_addresses(addresses)
        )
    )
    .build();

let mut message = consumer.consume_filtered().await.unwrap();

let message_body_str = std::str::from_utf8(&message.body).unwrap();
println!("message body = {}", message_body_str);

message.finish();

Basic producer example

let topic = NSQTopic::new("names").unwrap();

let mut producer = NSQProducerConfig::new("127.0.0.1:4150").build();

// Wait until a connection is initialized
assert_matches!(producer.consume().await.unwrap(), NSQEvent::Healthy());
// Publish a single message
producer.publish(&topic, b"alice1".to_vec()).unwrap();
// Wait until the message is acknowledged by NSQ
assert_matches!(producer.consume().await.unwrap(), NSQEvent::Ok());

Features

  • Subscriptions
  • Publication
  • NSQLookupd based discovery.
  • Message requeue backoff
  • NSQD TLS negotiation
    • Unverified server certificates
    • Custom certificate authority
    • Client certificates
  • Deflate NSQD compression
  • Snappy NSQD compression
  • Sampling
  • Auth

Dependencies

~18MB
~405K SLoC