owned_transaction

owned_transaction is the same buffer over values, encoding each one with the default codec.

It is the owned kind on a live publisher outside a handler, where no slot carries a codec: the publisher an after_startup hook is handed, for one. publisher.owned_transaction() returns a TypedTransaction that owns its buffer. publish(name, &value) encodes the value and buffers it; commit sends the buffer as one MULTI / EXEC block, and abort discards it. As with transaction, any number can be open on one publisher, and commit and abort consume the value.

after_startup runs its hook once the broker is connected and the subscriptions are open, with the policy it names paired into a live publisher. That makes it the place for a service's first publishes, such as seed data that must arrive whole. A hook that returns an error fails the startup. On a cluster TransactionalPublish refuses to pair, so the service does not start and seeds nothing.

Example

The service writes its opening prices as one block once it is connected; the catalog reader sees both or neither.

Macros
Manual

examples/redis_transactions_owned.rs
/// Writes the opening prices as one block: readers of `prices` see the whole list or none of it.
async fn seed_prices<P: OwnedTransactions>(
    publisher: P,
) -> Result<(), Box<dyn Error + Send + Sync>> {
    let mut seed = publisher.owned_transaction().await?;
    for (sku, cents) in [("tea", 450), ("cake", 600)] {
        let price = Price {
            sku: sku.to_owned(),
            cents,
        };
        seed.publish("prices", &price).await?;
    }
    seed.commit().await?;
    Ok(())
}

/// The service on `broker`: a standalone or a sentinel deployment, where `MULTI` / `EXEC` runs.
fn app_on(broker: RedisBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("catalog", "0.1.0")).with_broker(broker, |b| {
        b.include(list_price);
        // Runs once the broker is connected and the subscription is open.
        b.after_startup(TransactionalPublish, async move |publisher| {
            seed_prices(publisher).await.map_err(io::Error::other)
        });
    })
}

examples/manual/redis_transactions_owned.rs
/// Writes the opening prices as one block: readers of `prices` see the whole list or none of it.
async fn seed_prices<P: OwnedTransactions>(
    publisher: P,
) -> Result<(), Box<dyn Error + Send + Sync>> {
    let mut seed = publisher.owned_transaction().await?;
    for (sku, cents) in [("tea", 450), ("cake", 600)] {
        let price = Price {
            sku: sku.to_owned(),
            cents,
        };
        seed.publish("prices", &price).await?;
    }
    seed.commit().await?;
    Ok(())
}

/// The service on `broker`: a standalone or a sentinel deployment, where `MULTI` / `EXEC` runs.
fn app_on(broker: RedisBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("catalog", "0.1.0")).with_broker(broker, |b| {
        let list_price = subscriber(RedisStream::new("prices").group("catalog"), ListPrice);
        b.include(list_price.build());
        // Runs once the broker is connected and the subscription is open.
        b.after_startup(TransactionalPublish, async move |publisher| {
            seed_prices(publisher).await.map_err(io::Error::other)
        });
    })
}

Example​

Example