Struct ServerTxPublish 

pub struct ServerTxPublish(/* private fields */);

The server-transactional publish policy: same options as LapinPublish, pairing into ServerTxPublisher.

Reached with LapinPublish::server_tx.

Examples

use std::error::Error;

use ruststream_lapin::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Deserialize)]
struct Transfer {
    from: u64,
    to: u64,
    cents: i64,
}

#[derive(Serialize, Outgoing)]
#[outgoing(name = "ledger.entry")]
struct Entry {
    account: u64,
    cents: i64,
}

// Both legs of a transfer become visible together, or neither does.
async fn post(
    ledger: &impl TransactionalPublisher,
    transfer: &Transfer,
) -> Result<(), Box<dyn Error + Send + Sync>> {
    let debit = Entry {
        account: transfer.from,
        cents: -transfer.cents,
    };
    let credit = Entry {
        account: transfer.to,
        cents: transfer.cents,
    };
    let mut scope = ledger.begin().await?;
    for entry in [debit, credit] {
        if let Err(err) = scope.message(&entry).publish().await {
            scope.abort().await?;
            return Err(err.into());
        }
    }
    scope.commit().await?;
    Ok(())
}

#[subscriber("transfers")]
async fn book(
    transfer: &Transfer,
    Out(ledger): Out<impl TransactionalPublisher>,
) -> HandlerOutcome {
    if post(ledger, transfer).await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

fn app() -> RustStream {
    let broker = LapinBroker::new("amqp://localhost:5672");
    let ledger = LapinPublish::default().exchange("ledger").server_tx();
    RustStream::new(AppInfo::new("ledger", "0.1.0")).with_broker(broker, |b| {
        b.include(book).out(DefaultSlot, ledger).build();
    })
}

Implementations

impl ServerTxPublish

pub fn exchange(self, exchange: impl Into<String>) -> Self

Publishes to exchange instead of the default exchange.

pub fn persistent(self, persistent: bool) -> Self

Whether messages are marked persistent (delivery mode 2). Defaults to true.

One message departs from this with the publish builder’s persistent step.

pub fn priority(self, priority: u8) -> Self

The AMQP priority property messages carry. Unset by default, so the broker sees no priority at all.

It only orders deliveries on a queue declared with x-max-priority. One message departs from this with the publish builder’s priority step.

pub fn expiration(self, ttl: Duration) -> Self

The AMQP per-message expiration (TTL) messages carry: the broker drops one once ttl has passed without it being consumed. Unset by default, so messages do not expire.

One message departs from this with the publish builder’s expiration step.

Trait Implementations

impl Clone for ServerTxPublish

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 ServerTxPublish

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

Formats the value using the given formatter.

impl Default for ServerTxPublish

fn default() -> Self

Returns the “default value” for a type.

impl Eq for ServerTxPublish

impl LapinPublishPolicy for ServerTxPublish

fn bind(self, connected: &ConnectedLapinBroker) -> Self::Live

Pairs the policy with the connected broker, producing the live publisher.

impl PartialEq for ServerTxPublish

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

Equality operator ==.

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

Inequality operator !=.

impl PublishPolicy<ConnectedLapinBroker> for ServerTxPublish

type Live = ServerTxPublisher

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 pair( self, connected: &ConnectedLapinBroker, ) -> impl Future<Output = Result<Self::Live, PairError>>

Pairs the policy with a connected broker, producing the live publisher.

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

What a publish through this policy adds to its channel in the generated AsyncAPI document.

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

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

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

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

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

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

impl StructuralPartialEq for ServerTxPublish