Trait Transaction 

pub trait Transaction: Send {
    type Payload: PayloadForm;
    type Error: Error + Send + Sync + 'static;
    type Options: Clone + Send + Sync + 'static;

    // Required methods
    fn publish(
        &mut self,
        msg: OutgoingMessage<'_, <Self::Payload as PayloadForm>::Form<'_>>,
        options: Option<&Self::Options>,
    ) -> impl Future<Output = Result<(), Self::Error>> + Send;
    fn commit(self) -> impl Future<Output = Result<(), Self::Error>> + Send;
    fn abort(self) -> impl Future<Output = Result<(), Self::Error>> + Send;

    // Provided method
    fn base_headers(&self) -> Option<&HeaderMap> { ... }
}

A client-buffered transaction that owns its buffer, opened by OwnedTransactions::transaction.

publish appends to the buffer; nothing is visible to subscribers until commit flushes the whole buffer atomically, and abort discards it. Both settle the transaction by consuming self, so a double commit, a commit after an abort, or a publish after settling are compile errors, not runtime checks. Dropping an unsettled transaction discards the buffer like an abort (destructors cannot run async work); implementations log a warning.

This is the transaction value of the owned kind; see OwnedTransactions for the contrast with the borrowed TransactionalPublisher kind.

Examples

use ruststream::{OutgoingMessage, Transaction};

async fn settle_pair<T: Transaction>(mut txn: T) -> Result<(), T::Error> {
    txn.publish(OutgoingMessage::new("orders", b"{}".as_slice()), None).await?;
    txn.publish(OutgoingMessage::new("audit", b"{}".as_slice()), None).await?;
    txn.commit().await
}

Required Associated Types

type Payload: PayloadForm

How this transaction consumes the payload: Lend where it reads the bytes into a frame of its own, Take where the buffer it holds until the commit keeps them.

A type of its own, for the reason Options is one: a transaction is a publish surface of its own, and a client buffer keeps what the direct publish may only read. See Publisher::Payload.

type Error: Error + Send + Sync + 'static

The error type returned by transaction operations.

type Options: Clone + Send + Sync + 'static

The broker’s per-message settings inside this transaction: a type of its own, because a transaction is a publish surface of its own and may honour a different set of settings than the publisher it was opened from. Most brokers name the publisher’s type here.

See Publisher::Options, including why the type is Clone + 'static. A transaction with no per-message setting writes type Options = ();.

Required Methods

fn publish( &mut self, msg: OutgoingMessage<'_, <Self::Payload as PayloadForm>::Form<'_>>, options: Option<&Self::Options>, ) -> impl Future<Output = Result<(), Self::Error>> + Send

Publishes msg into the transaction: buffered, not visible before commit.

A failed publish does not settle the transaction; the caller decides between retrying and abort. options is what the call site adjusted, and None where nothing did.

Errors

Returns Self::Error when the message cannot be buffered. For a pure client buffer this is infallible in practice; implementations that stage broker-side state may reject here.

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

Commits the transaction: the whole buffer becomes visible atomically, in publish order.

Errors

Returns Self::Error when the flush fails. A failed commit has still consumed the transaction and its buffer is lost; redelivery of the inputs, not resubmission of the buffer, is the recovery path (the same rule as TransactionalPublisher::commit).

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

Aborts the transaction, discarding the buffer.

Errors

Returns Self::Error when the implementation fails to discard staged broker-side state; for a pure client buffer this is infallible in practice.

Provided Methods

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

The headers this transaction contributes to every message published through the publish builder, underneath whatever the call site names.

The transaction-side twin of Publisher::base_headers, with the same default (None) and the same precedence: the call site wins over the transaction, the transaction wins over nothing. A transaction opened from a handle that carries an argument passes that argument on here, so a publish behaves the same inside a transaction as outside one.

Examples

use ruststream::{HeaderMap, OutgoingFor, Transaction};

struct Tagged<T>(T, HeaderMap);

impl<T: Transaction> Transaction for Tagged<T> {
    type Payload = T::Payload;
    type Error = T::Error;
    type Options = T::Options;

    async fn publish(
        &mut self,
        msg: OutgoingFor<'_, Self::Payload>,
        options: Option<&Self::Options>,
    ) -> Result<(), Self::Error> {
        self.0.publish(msg, options).await
    }

    async fn commit(self) -> Result<(), Self::Error> {
        self.0.commit().await
    }

    async fn abort(self) -> Result<(), Self::Error> {
        self.0.abort().await
    }

    fn base_headers(&self) -> Option<&HeaderMap> {
        Some(&self.1)
    }
}

async fn settle<T: Transaction>(txn: T) -> Result<(), T::Error> {
    let base = [("tenant", "acme")].into_iter().collect();
    Tagged(txn, base).commit().await
}

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors

impl Transaction for RedisTransaction

type Payload = Take

type Error = RedisError

type Options = RedisPublishOptions