Struct ConnectedZmqQueue 

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

The connected form of ZmqQueue; sockets attach per subscription and per publisher.

Implementations

impl<Role> ConnectedZmqQueue<Role>

pub fn bound_address(&self) -> Option<String>

The address a local subscription resolved by binding (useful with an ephemeral tcp://...:0 endpoint); None until a subscription has bound.

pub fn publisher(&self) -> ZmqQueuePublisher

A publisher from the connected form.

Trait Implementations

impl<Role: EndpointRole> ConnectedBroker for ConnectedZmqQueue<Role>

type Error = ZmqError

The error type returned by connected-broker operations.

type Closed = ()

The terminal witness returned by shutdown.

fn shutdown(self) -> impl Future<Output = Result<(), Self::Error>>

Closes the broker connection, flushing in-flight publishes and stopping background tasks, consuming the connected form.

impl<Role> Debug for ConnectedZmqQueue<Role>

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

Formats the value using the given formatter.

impl<Role: EndpointRole> DefaultPublish for ConnectedZmqQueue<Role>

type Policy = ZmqQueuePublish

The broker’s plain publish policy, constructible with its defaults.

impl<Role: EndpointRole> PublishPolicy<ConnectedZmqQueue<Role>> for ZmqQueuePublish

fn pair( self, connected: &ConnectedZmqQueue<Role>, ) -> impl Future<Output = Result<Self::Live, PairError>>

On the bind side the publisher’s socket attaches here, so it listens once the service has started; a dialing publisher attaches on its first publish.

Errors

Returns PairError when the socket of a binding publisher cannot take the endpoint.

fn channel_bindings(&self, channel: &str) -> Bindings

The destination reaches the binding, because on this pattern it is also what frame 0 of every message on the channel holds.

type Live = ZmqQueuePublisher

The live form this policy pairs into: a Publisher for a leaf policy, or the live wiring form for a combinator stack (a typed publisher over a policy pairs into the same typed publisher over the live leaf).

fn operation_bindings(&self, channel: &str) -> Bindings

What a publish through this policy adds to its send operation in the document.

fn message_bindings(&self, channel: &str) -> Bindings

What a publish through this policy adds to the messages that leave through it.

fn reply_address_location(&self) -> Option<&'static str>

Where a client finds the address of a reply this policy publishes, as a runtime expression.

impl Subscribe for ConnectedZmqQueue<Bind>

type Copies = AddressedCopies

The subscribe name is the address: the subscription binds the endpoint, a PUSH socket of this service dials that listener, and a publish under the subscription’s name reaches it. The runtime publishes a retry copy there, and nothing has to be written at the mount site.

This is the whole retry path on ZeroMQ. Nothing settles a delivery, so a handler asking for retry_after is served only by the deferred copy the runtime publishes here. Competing consumers still compete for that copy, so the worker that retries is not necessarily the one that asked.

type Subscriber = ZmqSubscriber

The subscriber type opened by a by-name subscription.

async fn subscribe(&self, name: &str) -> Result<Self::Subscriber, Self::Error>

Opens a subscription to name, producing this broker’s Subscriber.

fn declare_retry( &self, name: &str, declaration: &RetryDeclaration, ) -> Result<(), DeclareRetryError>

Takes what a registration mounted by a bare name declared about its retries, before the subscription opens.

impl Subscribe for ConnectedZmqQueue<Connect>

type Copies = NamedCopies

A subscription that dials reads from the PUSH peer at the endpoint, a ventilator that sends and takes nothing, so a copy published back to it is refused by the handshake. The subscription therefore addresses no copy, and every registration names where its copies go: .out_retry(policy).to("name") over a broker that reaches a consumer, or a transform that names one per delivery. A registration that names neither is refused before the subscription opens, and a .build() chain that names neither does not compile.

type Subscriber = ZmqSubscriber

The subscriber type opened by a by-name subscription.

async fn subscribe(&self, name: &str) -> Result<Self::Subscriber, Self::Error>

Opens a subscription to name, producing this broker’s Subscriber.

fn declare_retry( &self, name: &str, declaration: &RetryDeclaration, ) -> Result<(), DeclareRetryError>

Takes what a registration mounted by a bare name declared about its retries, before the subscription opens.

impl<Role: EndpointRole> TestableBroker for ConnectedZmqQueue<Role>

The queue’s routing: a peer that pushes into an endpoint a subscription binds reaches that subscription, and one that subscriptions dial hands each message to one of them in turn, whatever its name frame says. A publish of this service reaches its own subscription only through the listener that subscription bound.

fn install_coordinator(&self, coordinator: Coordinator)

Installs the harness coordinator into this broker’s bus for a test run. Idempotent: a second install on the same broker is ignored.

fn inject(&self, message: OutgoingMessage<'_>)

Injects a message onto the bus as an external producer would, synchronously (no awaiting). Routes through the broker’s normal fanout, so it is recorded and counted like any publish.

fn published(&self, name: &str) -> Vec<RawMessage>

Returns every message published to name on this broker, in publish order, after the evicted ones. Backs the harness’s published::<T>(name) assertions and expect_published.

fn routes(&self, _destination: &str, subscriptions: &[&str]) -> Vec<usize>

Which subscriptions of this broker a message published to destination is delivered to, as the broker’s own routing picks them: the positions in subscriptions (the names the app subscribed under on this broker) of the ones it delivers to.

fn evicted(&self, name: &str) -> usize

How many of the oldest messages published to name this broker no longer holds, so published starts after them.

fn backlog(&self) -> Backlog

What a subscription opened by name receives of the messages published to that name before it opened, as the real broker answers it: Backlog::Missed for publish/subscribe, Backlog::Delivered for a queue or a log that keeps them.