risingwave_common/transaction/
transaction_message.rs

1// Copyright 2023 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use enum_as_inner::EnumAsInner;
16use tokio::sync::oneshot;
17
18use crate::array::StreamChunk;
19use crate::transaction::transaction_id::TxnId;
20use crate::transaction::transaction_message::TxnMsg::{Begin, Data, End, Rollback};
21
22#[derive(Debug, EnumAsInner)]
23pub enum TxnMsg {
24    Begin(TxnId),
25    Data(TxnId, StreamChunk),
26    /// The optional sender is fired after the data has been durably persisted.
27    End(TxnId, Option<oneshot::Sender<()>>),
28    Rollback(TxnId),
29}
30
31impl TxnMsg {
32    pub fn txn_id(&self) -> TxnId {
33        match self {
34            Begin(txn_id) => *txn_id,
35            Data(txn_id, _) => *txn_id,
36            End(txn_id, _) => *txn_id,
37            Rollback(txn_id) => *txn_id,
38        }
39    }
40
41    pub fn as_stream_chunk(&self) -> Option<&StreamChunk> {
42        match self {
43            Begin(_) | End(..) | Rollback(_) => None,
44            Data(_, chunk) => Some(chunk),
45        }
46    }
47}
risingwave_common/transaction/transaction_message.rs

risingwave_common/transaction/
transaction_message.rs
