Struct ZmqQueuePublisher 

pub struct ZmqQueuePublisher { /* private fields */ }

Publishes into the queue over a PUSH socket of its own.

Paired by the runtime on the bind side, the socket listens from the moment the service starts, so a PULL peer that dials the endpoint connects before the first push. A publisher that dials, or one taken from a publisher() constructor, attaches on its first publish.

On an endpoint a subscription of this service bound, the socket dials that subscription, so a publish reaches it and nothing else: one under the subscription’s own name (a retry copy, a job the service feeds itself) is sent, and one under any other name returns ZmqError::Send rather than arriving at the subscription as its next delivery. On an endpoint a subscription of this service dials, the peer there only pushes, so every publish returns ZmqError::Send naming that subscription.

A publisher that sends before any subscription of its service binds the endpoint listens on the endpoint itself, for peers to dial. When a subscription of the same broker binds the endpoint later, the publisher gives the endpoint up first, on a fixed port and on port zero alike, and its next publish dials that subscription.

Examples

A worker’s results leave through a queue on an endpoint of their own, here one a sink binds. The jobs arrive on the socket the jobs subscription binds, and that socket takes every message pushed into it: a result published there would come back as the next job, so the worker’s own publisher refuses it, and the mount binds the reply to the second broker:

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 results = ZmqQueue::new(ZmqEndpoint::connect("tcp://sink:5556")).bindable();
    let to_results = results.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(to_results);
        })
        .with_broker(results, |_b| {})
}

Trait Implementations

impl Clone for ZmqQueuePublisher

fn clone(&self) -> Self

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for ZmqQueuePublisher

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

Formats the value using the given formatter.

impl Publisher for ZmqQueuePublisher

type Payload = Take

A frame owns its bytes: the payload becomes the message’s third frame and the socket keeps it until the send completes.

type Options = ()

ZMTP carries no per-message setting: a send takes the frames and nothing else, so there is nothing for a call site to adjust. The Publishing page shows what a publish carries.

async fn publish( &self, msg: OutgoingMessage<'_, BytesMut>, _options: Option<&Self::Options>, ) -> Result<(), Self::Error>

Cancel safety

Cancel-safe. A publish dropped before the socket takes its message sends nothing. One dropped mid-send, while the peer applies back-pressure, leaves its message with the publisher: the next publish through it or a clone completes that send before its own, and a failure of it is logged against its name.

type Error = ZmqError

The error type returned by publish.

fn base_headers(&self) -> Option<&HeaderMap>

The headers this publisher contributes to every message it sends, underneath whatever the message itself names.