Struct RedisPublisher 

pub struct RedisPublisher { /* private fields */ }

The live stream publisher: RedisPublish paired with a connection. Cheap to clone.

Publisher::publish appends the message to the stream named by OutgoingMessage::name with XADD <name> * .... The payload and headers are encoded as entry fields (see crate::RedisStream for the consuming side).

A publisher may outlive the connected broker it came from (it is a handle aliasing the connection), so every operation after shutdown reports RedisError::ShutDown rather than running against a closed pool.

Transactions

Both framework transaction kinds are available on standalone and sentinel topologies, and both commit the same way: the buffer is held client-side while the transaction is open and flushed as one MULTI / EXEC block, in publish order, so subscribers see the whole batch or none of it. They differ only in where that buffer lives.

• Borrowed (TransactionalPublisher): the handle carries one transaction. begin_transaction claims it and starts buffering published messages, commit flushes them, and abort discards them. Clones of a handle share the same open transaction, and a second begin_transaction while one is open is rejected.
• Owned (OwnedTransactions): every transaction call returns a RedisTransaction owning its own buffer, so any number can be open on one handle concurrently and the handle keeps publishing directly meanwhile.

Two Redis properties apply to both kinds. Cluster supports neither, because a MULTI block cannot span hash slots, so opening a transaction there returns RedisError::InvalidOptions. And Redis has no rollback: a command that fails at runtime inside EXEC does not undo the commands before it. For a block of XADDs against stream keys that is practically limited to out-of-memory and wrong-type keys; a command the server refuses to queue discards the whole block.

Examples

use ruststream::OutgoingMessage;
use ruststream_fred::stream::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Deserialize, Serialize, Outgoing)]
struct Order {
    id: u64,
}

#[subscriber("orders", reply("processed"))]
async fn process(orders: &[Order]) -> Result<Vec<Order>, HandlerOutcome> {
    Ok(orders.iter().map(|order| Order { id: order.id }).collect())
}

#[ruststream::app]
fn app() -> impl App {
    let broker = RedisBroker::standalone("redis://localhost:6379").default_group("workers");
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        // The borrowed kind: one batch's replies, buffered and committed as one MULTI / EXEC.
        b.include(process.batch(nonzero!(32)))
            .out_reply(TransactionalPublish)
            .transactional();

        // The owned kind, in the hook where a service makes its first publishes.
        b.after_startup(TransactionalPublish, async move |publisher| {
            let mut seed = publisher.transaction().await?;
            seed.publish(
                OutgoingMessage::new("processed", br#"{"id":0}"#.as_slice()),
                None,
            )
            .await?;
            seed.commit().await
        });
    })
}

Trait Implementations

impl Clone for RedisPublisher

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 RedisPublisher

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

Formats the value using the given formatter.

impl OwnedTransactions for RedisPublisher

Owned transactions: every transaction call opens an independent buffer-owning RedisTransaction, so any number can be open concurrently on one handle, next to (and unaffected by) the handle-level TransactionalPublisher transaction.

fn transaction( &self, ) -> impl Future<Output = Result<RedisTransaction, RedisError>>

Errors

Returns RedisError::InvalidOptions on a cluster topology, which cannot offer multi-key transactions.

type Transaction = RedisTransaction

The buffer-owning transaction opened by transaction.

impl Publisher for RedisPublisher

type Payload = Take

XADD hands the body to the client as the value of a field, and the client keeps it, so the buffer the framework wrote becomes that value.

type Options = RedisPublishOptions

XADD takes the entry id and the trim threshold per command, and this publisher fixes both (* and no trim); the stream key is the message’s name, not a setting. What is left to a call site is the partition key, which leaves as an entry field like any other header.

type Error = RedisError

The error type returned by publish.

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

Publishes a message to the broker, with the per-message settings the call site adjusted.

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

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

impl TransactionalPublisher for RedisPublisher

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

Starts buffering published messages on this handle.

Errors

Returns RedisError::InvalidOptions on a cluster topology, which cannot offer multi-key transactions, or RedisError::TransactionBusy when a transaction is already open on this handle (the open one is left untouched).

async fn commit(&self) -> Result<(), Self::Error>

Flushes the buffered XADDs as one MULTI / EXEC block, in publish order, then clears the transaction.

Errors

Returns RedisError::NoTransaction when no transaction is open on this handle, RedisError::ShutDown when the connection is gone, or RedisError::Publish if the block is rejected. On failure the transaction is already closed: the buffer is lost, and recovery is redelivery of the inputs rather than resubmission of the buffer.

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

Discards the buffered messages.

Errors

Returns RedisError::NoTransaction when no transaction is open on this handle.