magnetar/
transaction.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Pulsar transactions (PIP-31).
4//!
5//! Mirrors Java's `org.apache.pulsar.client.api.transaction.Transaction`. A
6//! [`Transaction`] is a thin token over a [`magnetar_proto::TxnId`]. Stamp the
7//! id on an [`crate::OutgoingMessage`] via `.txn(id)` (producer side) or on a
8//! consumer ack via the runtime engine's `ack_with_txn` family; then commit
9//! or abort via [`PulsarClient::commit_transaction`] /
10//! [`PulsarClient::abort_transaction`].
11//!
12//! The five façade methods are generic over [`crate::Engine`] (D1 phase 4 of
13//! the lift train, ADR-0026 §D1). Both `PulsarClient<TokioEngine>` and
14//! `PulsarClient<MoonpoolEngine<P>>` carry the same Transaction surface by
15//! dispatching through the [`crate::TransactionApi`] extension trait
16//! implemented per engine on its `ClientState` type.
17
18/// Result of committing or aborting a [`Transaction`]. Re-exported from `magnetar-proto`.
19pub use magnetar_proto::TxnState;
20
21use crate::client::PulsarError;
22use crate::{Engine, PulsarClient, TransactionApi};
23
24/// A live Pulsar transaction token. Holds the broker-assigned [`magnetar_proto::TxnId`].
25///
26/// `Transaction` is `Copy` (`TxnId` is 128 bits of plain data) so it can be passed to
27/// multiple producers / consumers without juggling references.
28#[derive(Debug, Clone, Copy)]
29pub struct Transaction {
30    id: magnetar_proto::TxnId,
31}
32
33impl Transaction {
34    pub(crate) fn new(id: magnetar_proto::TxnId) -> Self {
35        Self { id }
36    }
37
38    /// The transaction id — stamp this on producer sends via
39    /// [`crate::OutgoingMessage::txn`] and on consumer acks via the runtime
40    /// engine's `ack_with_txn` family.
41    #[must_use]
42    pub fn id(&self) -> magnetar_proto::TxnId {
43        self.id
44    }
45}
46
47impl From<Transaction> for magnetar_proto::TxnId {
48    fn from(txn: Transaction) -> Self {
49        txn.id
50    }
51}
52
53impl<E: Engine> PulsarClient<E>
54where
55    E::ClientState: TransactionApi,
56{
57    /// Open a new Pulsar transaction at the broker-side transaction coordinator (PIP-31).
58    /// Mirrors Java `PulsarClient#newTransaction()`.
59    ///
60    /// # Errors
61    /// - [`PulsarError::Other`] (with the runtime's error stringified) on broker rejection or wire
62    ///   failure.
63    pub async fn new_transaction(
64        &self,
65        timeout: std::time::Duration,
66    ) -> Result<Transaction, PulsarError> {
67        let id = TransactionApi::new_txn(&self.inner, timeout)
68            .await
69            .map_err(|err| PulsarError::Other(format!("new_transaction: {err}")))?;
70        Ok(Transaction::new(id))
71    }
72
73    /// Register a partition that the given transaction will write to.
74    /// Mirrors Java `Transaction#registerProducedTopic`.
75    ///
76    /// # Errors
77    /// - [`PulsarError::Other`] on broker rejection or wire failure.
78    pub async fn register_partition_to_transaction(
79        &self,
80        txn: Transaction,
81        topic: impl Into<String>,
82    ) -> Result<(), PulsarError> {
83        TransactionApi::add_partition_to_txn(&self.inner, txn.id(), topic.into())
84            .await
85            .map_err(|err| PulsarError::Other(format!("register_partition_to_transaction: {err}")))
86    }
87
88    /// Register a subscription that the given transaction will acknowledge on.
89    /// Mirrors Java `Transaction#registerSubscriptionToTxn`.
90    ///
91    /// # Errors
92    /// - [`PulsarError::Other`] on broker rejection or wire failure.
93    pub async fn register_subscription_to_transaction(
94        &self,
95        txn: Transaction,
96        topic: impl Into<String>,
97        subscription: impl Into<String>,
98    ) -> Result<(), PulsarError> {
99        TransactionApi::add_subscription_to_txn(
100            &self.inner,
101            txn.id(),
102            topic.into(),
103            subscription.into(),
104        )
105        .await
106        .map_err(|err| PulsarError::Other(format!("register_subscription_to_transaction: {err}")))
107    }
108
109    /// Commit a transaction at the TC. Returns the final state reported by the TC.
110    /// Mirrors Java `Transaction#commit`.
111    ///
112    /// # Errors
113    /// - [`PulsarError::Other`] on broker rejection or wire failure.
114    pub async fn commit_transaction(&self, txn: Transaction) -> Result<TxnState, PulsarError> {
115        TransactionApi::end_txn(&self.inner, txn.id(), magnetar_proto::TxnAction::Commit)
116            .await
117            .map_err(|err| PulsarError::Other(format!("commit_transaction: {err}")))
118    }
119
120    /// Abort a transaction at the TC. Returns the final state reported by the TC. Mirrors
121    /// Java `Transaction#abort`.
122    ///
123    /// # Errors
124    /// - [`PulsarError::Other`] on broker rejection or wire failure.
125    pub async fn abort_transaction(&self, txn: Transaction) -> Result<TxnState, PulsarError> {
126        TransactionApi::end_txn(&self.inner, txn.id(), magnetar_proto::TxnAction::Abort)
127            .await
128            .map_err(|err| PulsarError::Other(format!("abort_transaction: {err}")))
129    }
130}
magnetar/transaction.rs

magnetar/
transaction.rs
