#client-connection #actix #actor #nsq #async #queue

nightly nsq-client

Rust client for the NSQ realtime message processing system

12 releases

Uses old Rust 2015

0.1.12 Feb 21, 2019
0.1.11 Feb 19, 2019

#681 in Asynchronous

MIT license

78KB
1K SLoC

NSQ client written in rust Build Status Build status

Sponsored by

A Actix based client implementation for the NSQ realtime message processing system.

Nsq-client it's designed to support by default multiple Readers for Multiple Connections, readers are routed per single connection by a round robin algorithm.

Usage

To use nsq-client, add this to your Cargo.toml:

[dependencies]
actix = "0.7"
nsq-client = "0.1.12"

Create your first consumer

In order to use nsq-client you first need to create a Reader actor which implement Handler for the type of messages you want to receive from the connections and then subscribe it to the connections to be able to receive the type of messages you've selected.

Available messages are:

Simple Consumer (SUB)

extern crate actix;
extern crate nsq_client;

use std::sync::Arc;

use actix::prelude::*;

use nsq_client::{Connection, Msg, Fin, Subscribe, Config};

struct MyReader {
    pub conn: Arc<Addr<Connection>>,
}

impl Actor for MyReader {
    type Context = Context<Self>;
    fn started(&mut self, ctx: &mut Self::Context) {
        self.subscribe::<Msg>(ctx, self.conn.clone());
    }
}

impl Handler<Msg> for MyReader {
    fn handle(&mut self, msg: Msg, _: &mut Self::Context) {
        println!("MyReader received {:?}", msg);
        if let Ok(body) = String::from_utf8(msg.body) {
              println!("utf8 msg: {}", body);
        }
        self.conn.do_send(Fin(msg.id));
    }
}

fn main() {
    let sys = System::new("consumer");
    let config = Config::default().client_id("consumer");
    let c = Supervisor::start(|_| Connection::new(
        "test", // <- topic
        "test", // <- channel
        "0.0.0.0:4150", // <- nsqd tcp address
        Some(config), // <- config (Optional)
        None, // secret for Auth (Optional)
        Some(2) // <- RDY (Optional default: 1)
    ));
    let conn = Arc::new(c);
    let _ = MyReader{ conn: conn.clone() }.start(); // <- Same thread reader
    let _ = Arbiter::start(|_| MyReader{ conn: conn }); // <- start another reader in different thread
    sys.run();
}

launch nsqd

$ nsqd -verbose

launch the reader

$ RUST_LOG=nsq_client=debug cargo run

launch the producer

$ cargo run

asciicast

Examples

ToDo

  • Discovery
  • TLS
  • Snappy
  • First-ready-first-served readers routing algorithm.

License

Licensed under

Dependencies

~11MB
~191K SLoC