elefant_client/postgres_client/replication/
messages.rs

1use crate::protocol::frame_reader::ByteSliceError;
2use crate::protocol::FieldDescription;
3use crate::types::{FromSqlBase, FromSqlText};
4use crate::ElefantClientError;
5use crate::PostgresType;
6use std::borrow::Cow;
7use std::error::Error;
8use std::fmt;
9use std::str::FromStr;
10
11#[derive(Debug)]
12pub enum ReplicationError {
13    TruncatedMessage,
14    UnknownReplicationMessageType(u8),
15    UnknownTupleColumnType(u8),
16    UnknownUpdateMarker(u8),
17}
18
19impl From<ByteSliceError> for ReplicationError {
20    fn from(_: ByteSliceError) -> Self {
21        ReplicationError::TruncatedMessage
22    }
23}
24
25impl From<ReplicationError> for ElefantClientError {
26    fn from(e: ReplicationError) -> Self {
27        ElefantClientError::PostgresError(format!("{e:?}"))
28    }
29}
30
31/// PostgreSQL Log Sequence Number.
32#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
33pub struct Lsn(pub u64);
34
35impl fmt::Display for Lsn {
36    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
37        write!(f, "{:X}/{:X}", self.0 >> 32, self.0 & 0xFFFF_FFFF)
38    }
39}
40
41impl FromStr for Lsn {
42    type Err = String;
43
44    fn from_str(s: &str) -> Result<Self, Self::Err> {
45        let (high_str, low_str) = s
46            .split_once('/')
47            .ok_or_else(|| format!("Invalid LSN format: {s}"))?;
48        let high =
49            u64::from_str_radix(high_str, 16).map_err(|e| format!("Invalid LSN high part: {e}"))?;
50        let low =
51            u64::from_str_radix(low_str, 16).map_err(|e| format!("Invalid LSN low part: {e}"))?;
52        Ok(Lsn((high << 32) | low))
53    }
54}
55
56impl<'a> FromSqlBase<'a> for Lsn {
57    fn accepts_postgres_type(oid: i32) -> bool {
58        oid == PostgresType::PG_LSN.oid
59    }
60}
61
62impl<'a> FromSqlText<'a> for Lsn {
63    fn from_sql_text(
64        raw: &'a str,
65        _field: &FieldDescription,
66    ) -> Result<Self, Box<dyn Error + Sync + Send>> {
67        Ok(raw.parse()?)
68    }
69}
70
71#[derive(Debug)]
72pub enum ReplicationMessage<'a> {
73    XLogData(XLogData<'a>),
74    PrimaryKeepalive(PrimaryKeepalive),
75}
76
77#[derive(Debug)]
78pub struct XLogData<'a> {
79    pub start_lsn: Lsn,
80    pub end_lsn: Lsn,
81    pub server_time: i64,
82    pub data: &'a [u8],
83}
84
85#[derive(Debug)]
86pub struct PrimaryKeepalive {
87    pub end_lsn: Lsn,
88    pub server_time: i64,
89    pub reply_requested: bool,
90}
91
92#[derive(Debug)]
93pub enum PgOutputMessage<'a> {
94    Begin(BeginMessage),
95    Commit(CommitMessage),
96    Relation(RelationMessage<'a>),
97    Insert(InsertMessage<'a>),
98    Update(UpdateMessage<'a>),
99    Delete(DeleteMessage<'a>),
100    Truncate(TruncateMessage),
101    Origin(OriginMessage<'a>),
102    Type(TypeMessage<'a>),
103    LogicalDecodingMessage(LogicalDecodingMessage<'a>),
104    StreamStart(StreamStartMessage),
105    StreamStop,
106    StreamCommit(StreamCommitMessage),
107    StreamAbort(StreamAbortMessage),
108    /// A pgoutput message type that is part of the protocol but not yet
109    /// fully supported (e.g., two-phase commit messages from proto_version 3+).
110    Unsupported {
111        msg_type: u8,
112        data: &'a [u8],
113    },
114}
115
116#[derive(Debug)]
117pub struct BeginMessage {
118    pub final_lsn: Lsn,
119    pub commit_timestamp: i64,
120    pub xid: u32,
121}
122
123#[derive(Debug)]
124pub struct CommitMessage {
125    pub flags: u8,
126    pub commit_lsn: Lsn,
127    pub end_lsn: Lsn,
128    pub commit_timestamp: i64,
129}
130
131#[derive(Debug)]
132pub struct RelationMessage<'a> {
133    pub relation_id: u32,
134    pub namespace: Cow<'a, str>,
135    pub name: Cow<'a, str>,
136    pub replica_identity: u8,
137    pub columns: Vec<RelationColumn<'a>>,
138}
139
140#[derive(Debug)]
141pub struct RelationColumn<'a> {
142    pub flags: u8,
143    pub name: Cow<'a, str>,
144    pub type_oid: u32,
145    pub type_modifier: i32,
146}
147
148#[derive(Debug)]
149pub struct InsertMessage<'a> {
150    pub relation_id: u32,
151    pub tuple: TupleData<'a>,
152}
153
154#[derive(Debug)]
155pub struct UpdateMessage<'a> {
156    pub relation_id: u32,
157    pub old_tuple: Option<TupleData<'a>>,
158    pub new_tuple: TupleData<'a>,
159}
160
161#[derive(Debug)]
162pub struct DeleteMessage<'a> {
163    pub relation_id: u32,
164    pub old_tuple: TupleData<'a>,
165}
166
167#[derive(Debug)]
168pub struct TruncateMessage {
169    pub option_bits: u8,
170    pub relation_ids: Vec<u32>,
171}
172
173#[derive(Debug)]
174pub struct OriginMessage<'a> {
175    pub origin_lsn: Lsn,
176    pub origin_name: Cow<'a, str>,
177}
178
179#[derive(Debug)]
180pub struct TypeMessage<'a> {
181    pub type_oid: u32,
182    pub namespace: Cow<'a, str>,
183    pub name: Cow<'a, str>,
184}
185
186#[derive(Debug)]
187pub struct LogicalDecodingMessage<'a> {
188    pub transactional: bool,
189    pub lsn: Lsn,
190    pub prefix: Cow<'a, str>,
191    pub content: &'a [u8],
192}
193
194#[derive(Debug)]
195pub struct StreamStartMessage {
196    pub xid: u32,
197    pub first_segment: bool,
198}
199
200#[derive(Debug)]
201pub struct StreamCommitMessage {
202    pub xid: u32,
203    pub flags: u8,
204    pub commit_lsn: Lsn,
205    pub end_lsn: Lsn,
206    pub commit_timestamp: i64,
207}
208
209#[derive(Debug)]
210pub struct StreamAbortMessage {
211    pub xid: u32,
212    pub sub_xid: u32,
213}
214
215#[derive(Debug)]
216pub struct TupleData<'a> {
217    pub columns: Vec<TupleColumn<'a>>,
218}
219
220#[derive(Debug)]
221pub enum TupleColumn<'a> {
222    Null,
223    Unchanged,
224    Text(Cow<'a, str>),
225    Binary(&'a [u8]),
226}
elefant_client/postgres_client/replication/messages.rs

elefant_client/postgres_client/replication/
messages.rs
