Struct ZmqEndpoint 

pub struct ZmqEndpoint<Role = Bind> { /* private fields */ }

An address (tcp://... or ipc://...) and the side of it this process takes, fixed by the constructor and carried in the type.

Examples

A worker that listens for jobs from a producer on the same host and pushes its results to a collector it dials over the network:

use ruststream_zeromq::queue::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Deserialize)]
struct Job {
    id: u64,
}

#[derive(Serialize, Outgoing)]
#[outgoing(name = "results")]
struct Done {
    id: u64,
}

#[subscriber("jobs", reply)]
async fn work(job: &Job) -> Done {
    Done { id: job.id }
}

#[ruststream::app]
fn app() -> impl App {
    let collector = ZmqQueue::new(ZmqEndpoint::connect("tcp://collector:5560")).bindable();
    let results = collector.bind(Publish);
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker(ZmqQueue::new(ZmqEndpoint::bind("ipc:///tmp/jobs")), |b| {
            b.include(work).out_reply(results);
        })
        .register_broker(collector)
}

Implementations

impl ZmqEndpoint<Bind>

pub fn bind(address: impl Into<String>) -> Self

This process listens on address.

Examples

A worker that listens on a local socket for the producers on its host:

use ruststream_zeromq::queue::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Job {
    id: u64,
}

#[subscriber("jobs")]
async fn work(job: &Job) -> HandlerOutcome {
    tracing::info!(job.id, "working on the job");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("worker", "0.1.0")).with_broker(
        ZmqQueue::new(ZmqEndpoint::bind("ipc:///tmp/jobs")),
        |b| {
            b.include(work);
        },
    )
}

impl ZmqEndpoint<Connect>

pub fn connect(address: impl Into<String>) -> Self

This process dials out to address.

Examples

A worker that pushes its results to the collector it dials:

use ruststream_zeromq::queue::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Deserialize)]
struct Job {
    id: u64,
}

#[derive(Serialize, Outgoing)]
#[outgoing(name = "results")]
struct Done {
    id: u64,
}

#[subscriber("jobs", reply)]
async fn work(job: &Job) -> Done {
    Done { id: job.id }
}

#[ruststream::app]
fn app() -> impl App {
    let collector = ZmqQueue::new(ZmqEndpoint::connect("tcp://collector:5560")).bindable();
    let results = collector.bind(Publish);
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker(ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5555")), |b| {
            b.include(work).out_reply(results);
        })
        .register_broker(collector)
}

impl<Role> ZmqEndpoint<Role>

pub fn address(&self) -> &str

The address string.

Examples

A worker that logs where it listens before it starts:

use ruststream_zeromq::queue::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Job {
    id: u64,
}

#[subscriber("jobs")]
async fn work(job: &Job) -> HandlerOutcome {
    tracing::info!(job.id, "working on the job");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    let endpoint = ZmqEndpoint::bind("tcp://0.0.0.0:5555");
    tracing::info!(address = endpoint.address(), "worker listening");
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker(ZmqQueue::new(endpoint), |b| {
            b.include(work);
        })
}

Trait Implementations

impl<Role: Clone> Clone for ZmqEndpoint<Role>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<Role: Debug> Debug for ZmqEndpoint<Role>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<Role: Eq> Eq for ZmqEndpoint<Role>

impl<Role: PartialEq> PartialEq for ZmqEndpoint<Role>

fn eq(&self, other: &Self) -> bool

Equality operator ==.

fn ne(&self, other: &Rhs) -> bool

Inequality operator !=.

impl<Role: PartialEq> StructuralPartialEq for ZmqEndpoint<Role>