postgres_protocol/message/
backend.rs

1#![allow(missing_docs)]
2
3use std::io::{self, Read};
4use std::ops::Range;
5use std::str;
6
7use crate::message::shared;
8use byteorder::{BigEndian, ReadBytesExt};
9use bytes::{Bytes, BytesMut};
10use fallible_iterator::FallibleIterator;
11use memchr::memchr;
12
13use crate::message::shared::Buffer;
14use crate::Oid;
15
16pub const PARSE_COMPLETE_TAG: u8 = b'1';
17pub const BIND_COMPLETE_TAG: u8 = b'2';
18pub const CLOSE_COMPLETE_TAG: u8 = b'3';
19pub const NOTIFICATION_RESPONSE_TAG: u8 = b'A';
20pub const COPY_DONE_TAG: u8 = b'c';
21pub const COMMAND_COMPLETE_TAG: u8 = b'C';
22pub const COPY_DATA_TAG: u8 = b'd';
23pub const DATA_ROW_TAG: u8 = b'D';
24pub const ERROR_RESPONSE_TAG: u8 = b'E';
25pub const COPY_IN_RESPONSE_TAG: u8 = b'G';
26pub const COPY_OUT_RESPONSE_TAG: u8 = b'H';
27pub const EMPTY_QUERY_RESPONSE_TAG: u8 = b'I';
28pub const BACKEND_KEY_DATA_TAG: u8 = b'K';
29pub const NO_DATA_TAG: u8 = b'n';
30pub const NOTICE_RESPONSE_TAG: u8 = b'N';
31pub const AUTHENTICATION_TAG: u8 = b'R';
32pub const PORTAL_SUSPENDED_TAG: u8 = b's';
33pub const PARAMETER_STATUS_TAG: u8 = b'S';
34pub const PARAMETER_DESCRIPTION_TAG: u8 = b't';
35pub const ROW_DESCRIPTION_TAG: u8 = b'T';
36pub const READY_FOR_QUERY_TAG: u8 = b'Z';
37
38pub use shared::Header;
39
40/// An enum representing Postgres backend messages.
41#[non_exhaustive]
42pub enum Message {
43    AuthenticationCleartextPassword,
44    AuthenticationGss,
45    AuthenticationKerberosV5,
46    AuthenticationMd5Password(AuthenticationMd5PasswordBody),
47    AuthenticationOk,
48    AuthenticationScmCredential,
49    AuthenticationSspi,
50    AuthenticationGssContinue(AuthenticationGssContinueBody),
51    AuthenticationSasl(AuthenticationSaslBody),
52    AuthenticationSaslContinue(AuthenticationSaslContinueBody),
53    AuthenticationSaslFinal(AuthenticationSaslFinalBody),
54    BackendKeyData(BackendKeyDataBody),
55    BindComplete,
56    CloseComplete,
57    CommandComplete(CommandCompleteBody),
58    CopyData(CopyDataBody),
59    CopyDone,
60    CopyInResponse(CopyInResponseBody),
61    CopyOutResponse(CopyOutResponseBody),
62    DataRow(DataRowBody),
63    EmptyQueryResponse,
64    ErrorResponse(ErrorResponseBody),
65    NoData,
66    NoticeResponse(NoticeResponseBody),
67    NotificationResponse(NotificationResponseBody),
68    ParameterDescription(ParameterDescriptionBody),
69    ParameterStatus(ParameterStatusBody),
70    ParseComplete,
71    PortalSuspended,
72    ReadyForQuery(ReadyForQueryBody),
73    RowDescription(RowDescriptionBody),
74}
75
76impl Message {
77    #[inline]
78    pub fn parse(buf: &mut BytesMut) -> io::Result<Option<Message>> {
79        if buf.len() < 5 {
80            let to_read = 5 - buf.len();
81            buf.reserve(to_read);
82            return Ok(None);
83        }
84
85        let tag = buf[0];
86        let len = (&buf[1..5]).read_u32::<BigEndian>().unwrap();
87
88        if len < 4 {
89            return Err(io::Error::new(
90                io::ErrorKind::InvalidInput,
91                "invalid message length: parsing u32",
92            ));
93        }
94
95        let total_len = len as usize + 1;
96        if buf.len() < total_len {
97            let to_read = total_len - buf.len();
98            buf.reserve(to_read);
99            return Ok(None);
100        }
101
102        let mut buf = Buffer {
103            bytes: buf.split_to(total_len).freeze(),
104            idx: 5,
105        };
106
107        let message = match tag {
108            PARSE_COMPLETE_TAG => Message::ParseComplete,
109            BIND_COMPLETE_TAG => Message::BindComplete,
110            CLOSE_COMPLETE_TAG => Message::CloseComplete,
111            NOTIFICATION_RESPONSE_TAG => {
112                let process_id = buf.read_i32::<BigEndian>()?;
113                let channel = buf.read_cstr()?;
114                let message = buf.read_cstr()?;
115                Message::NotificationResponse(NotificationResponseBody {
116                    process_id,
117                    channel,
118                    message,
119                })
120            }
121            COPY_DONE_TAG => Message::CopyDone,
122            COMMAND_COMPLETE_TAG => {
123                let tag = buf.read_cstr()?;
124                Message::CommandComplete(CommandCompleteBody { tag })
125            }
126            COPY_DATA_TAG => {
127                let storage = buf.read_all();
128                Message::CopyData(CopyDataBody { storage })
129            }
130            DATA_ROW_TAG => {
131                let len = buf.read_u16::<BigEndian>()?;
132                let storage = buf.read_all();
133                Message::DataRow(DataRowBody { storage, len })
134            }
135            ERROR_RESPONSE_TAG => {
136                let storage = buf.read_all();
137                Message::ErrorResponse(ErrorResponseBody { storage })
138            }
139            COPY_IN_RESPONSE_TAG => {
140                let format = buf.read_u8()?;
141                let len = buf.read_u16::<BigEndian>()?;
142                let storage = buf.read_all();
143                Message::CopyInResponse(CopyInResponseBody {
144                    format,
145                    len,
146                    storage,
147                })
148            }
149            COPY_OUT_RESPONSE_TAG => {
150                let format = buf.read_u8()?;
151                let len = buf.read_u16::<BigEndian>()?;
152                let storage = buf.read_all();
153                Message::CopyOutResponse(CopyOutResponseBody {
154                    format,
155                    len,
156                    storage,
157                })
158            }
159            EMPTY_QUERY_RESPONSE_TAG => Message::EmptyQueryResponse,
160            BACKEND_KEY_DATA_TAG => {
161                let process_id = buf.read_i32::<BigEndian>()?;
162                let secret_key = buf.read_i32::<BigEndian>()?;
163                Message::BackendKeyData(BackendKeyDataBody {
164                    process_id,
165                    secret_key,
166                })
167            }
168            NO_DATA_TAG => Message::NoData,
169            NOTICE_RESPONSE_TAG => {
170                let storage = buf.read_all();
171                Message::NoticeResponse(NoticeResponseBody { storage })
172            }
173            AUTHENTICATION_TAG => match buf.read_i32::<BigEndian>()? {
174                0 => Message::AuthenticationOk,
175                2 => Message::AuthenticationKerberosV5,
176                3 => Message::AuthenticationCleartextPassword,
177                5 => {
178                    let mut salt = [0; 4];
179                    buf.read_exact(&mut salt)?;
180                    Message::AuthenticationMd5Password(AuthenticationMd5PasswordBody { salt })
181                }
182                6 => Message::AuthenticationScmCredential,
183                7 => Message::AuthenticationGss,
184                8 => {
185                    let storage = buf.read_all();
186                    Message::AuthenticationGssContinue(AuthenticationGssContinueBody(storage))
187                }
188                9 => Message::AuthenticationSspi,
189                10 => {
190                    let storage = buf.read_all();
191                    Message::AuthenticationSasl(AuthenticationSaslBody(storage))
192                }
193                11 => {
194                    let storage = buf.read_all();
195                    Message::AuthenticationSaslContinue(AuthenticationSaslContinueBody(storage))
196                }
197                12 => {
198                    let storage = buf.read_all();
199                    Message::AuthenticationSaslFinal(AuthenticationSaslFinalBody(storage))
200                }
201                tag => {
202                    return Err(io::Error::new(
203                        io::ErrorKind::InvalidInput,
204                        format!("unknown authentication tag `{}`", tag),
205                    ));
206                }
207            },
208            PORTAL_SUSPENDED_TAG => Message::PortalSuspended,
209            PARAMETER_STATUS_TAG => {
210                let name = buf.read_cstr()?;
211                let value = buf.read_cstr()?;
212                Message::ParameterStatus(ParameterStatusBody { name, value })
213            }
214            PARAMETER_DESCRIPTION_TAG => {
215                let len = buf.read_u16::<BigEndian>()?;
216                let storage = buf.read_all();
217                Message::ParameterDescription(ParameterDescriptionBody { storage, len })
218            }
219            ROW_DESCRIPTION_TAG => {
220                let len = buf.read_u16::<BigEndian>()?;
221                let storage = buf.read_all();
222                Message::RowDescription(RowDescriptionBody { storage, len })
223            }
224            READY_FOR_QUERY_TAG => {
225                let status = buf.read_u8()?;
226                Message::ReadyForQuery(ReadyForQueryBody { status })
227            }
228            tag => {
229                return Err(io::Error::new(
230                    io::ErrorKind::InvalidInput,
231                    format!("unknown message tag `{}`", tag),
232                ));
233            }
234        };
235
236        if !buf.is_empty() {
237            return Err(io::Error::new(
238                io::ErrorKind::InvalidInput,
239                "invalid message length: expected buffer to be empty",
240            ));
241        }
242
243        Ok(Some(message))
244    }
245}
246
247pub struct AuthenticationMd5PasswordBody {
248    salt: [u8; 4],
249}
250
251impl AuthenticationMd5PasswordBody {
252    #[inline]
253    pub fn salt(&self) -> [u8; 4] {
254        self.salt
255    }
256}
257
258pub struct AuthenticationGssContinueBody(Bytes);
259
260impl AuthenticationGssContinueBody {
261    #[inline]
262    pub fn data(&self) -> &[u8] {
263        &self.0
264    }
265}
266
267pub struct AuthenticationSaslBody(Bytes);
268
269impl AuthenticationSaslBody {
270    #[inline]
271    pub fn mechanisms(&self) -> SaslMechanisms<'_> {
272        SaslMechanisms(&self.0)
273    }
274}
275
276pub struct SaslMechanisms<'a>(&'a [u8]);
277
278impl<'a> FallibleIterator for SaslMechanisms<'a> {
279    type Item = &'a str;
280    type Error = io::Error;
281
282    #[inline]
283    fn next(&mut self) -> io::Result<Option<&'a str>> {
284        let value_end = find_null(self.0, 0)?;
285        if value_end == 0 {
286            if self.0.len() != 1 {
287                return Err(io::Error::new(
288                    io::ErrorKind::InvalidData,
289                    "invalid message length: expected to be at end of iterator for sasl",
290                ));
291            }
292            Ok(None)
293        } else {
294            let value = get_str(&self.0[..value_end])?;
295            self.0 = &self.0[value_end + 1..];
296            Ok(Some(value))
297        }
298    }
299}
300
301pub struct AuthenticationSaslContinueBody(Bytes);
302
303impl AuthenticationSaslContinueBody {
304    #[inline]
305    pub fn data(&self) -> &[u8] {
306        &self.0
307    }
308}
309
310pub struct AuthenticationSaslFinalBody(Bytes);
311
312impl AuthenticationSaslFinalBody {
313    #[inline]
314    pub fn data(&self) -> &[u8] {
315        &self.0
316    }
317}
318
319pub struct BackendKeyDataBody {
320    process_id: i32,
321    secret_key: i32,
322}
323
324impl BackendKeyDataBody {
325    #[inline]
326    pub fn process_id(&self) -> i32 {
327        self.process_id
328    }
329
330    #[inline]
331    pub fn secret_key(&self) -> i32 {
332        self.secret_key
333    }
334}
335
336pub struct CommandCompleteBody {
337    tag: Bytes,
338}
339
340impl CommandCompleteBody {
341    #[inline]
342    pub fn tag(&self) -> io::Result<&str> {
343        get_str(&self.tag)
344    }
345}
346
347pub struct CopyDataBody {
348    storage: Bytes,
349}
350
351impl CopyDataBody {
352    #[inline]
353    pub fn data(&self) -> &[u8] {
354        &self.storage
355    }
356
357    #[inline]
358    pub fn into_bytes(self) -> Bytes {
359        self.storage
360    }
361}
362
363pub struct CopyInResponseBody {
364    format: u8,
365    len: u16,
366    storage: Bytes,
367}
368
369impl CopyInResponseBody {
370    #[inline]
371    pub fn format(&self) -> u8 {
372        self.format
373    }
374
375    #[inline]
376    pub fn column_formats(&self) -> ColumnFormats<'_> {
377        ColumnFormats {
378            remaining: self.len,
379            buf: &self.storage,
380        }
381    }
382}
383
384pub struct ColumnFormats<'a> {
385    buf: &'a [u8],
386    remaining: u16,
387}
388
389impl FallibleIterator for ColumnFormats<'_> {
390    type Item = u16;
391    type Error = io::Error;
392
393    #[inline]
394    fn next(&mut self) -> io::Result<Option<u16>> {
395        if self.remaining == 0 {
396            if self.buf.is_empty() {
397                return Ok(None);
398            } else {
399                return Err(io::Error::new(
400                    io::ErrorKind::InvalidInput,
401                    "invalid message length: wrong column formats",
402                ));
403            }
404        }
405
406        self.remaining -= 1;
407        self.buf.read_u16::<BigEndian>().map(Some)
408    }
409
410    #[inline]
411    fn size_hint(&self) -> (usize, Option<usize>) {
412        let len = self.remaining as usize;
413        (len, Some(len))
414    }
415}
416
417pub struct CopyOutResponseBody {
418    format: u8,
419    len: u16,
420    storage: Bytes,
421}
422
423impl CopyOutResponseBody {
424    #[inline]
425    pub fn format(&self) -> u8 {
426        self.format
427    }
428
429    #[inline]
430    pub fn column_formats(&self) -> ColumnFormats<'_> {
431        ColumnFormats {
432            remaining: self.len,
433            buf: &self.storage,
434        }
435    }
436}
437
438#[derive(Debug, Clone)]
439pub struct DataRowBody {
440    storage: Bytes,
441    len: u16,
442}
443
444impl DataRowBody {
445    #[inline]
446    pub fn ranges(&self) -> DataRowRanges<'_> {
447        DataRowRanges {
448            buf: &self.storage,
449            len: self.storage.len(),
450            remaining: self.len,
451        }
452    }
453
454    #[inline]
455    pub fn buffer(&self) -> &[u8] {
456        &self.storage
457    }
458
459    #[inline]
460    pub fn buffer_bytes(&self) -> &Bytes {
461        &self.storage
462    }
463}
464
465pub struct DataRowRanges<'a> {
466    buf: &'a [u8],
467    len: usize,
468    remaining: u16,
469}
470
471impl FallibleIterator for DataRowRanges<'_> {
472    type Item = Option<Range<usize>>;
473    type Error = io::Error;
474
475    #[inline]
476    fn next(&mut self) -> io::Result<Option<Option<Range<usize>>>> {
477        if self.remaining == 0 {
478            if self.buf.is_empty() {
479                return Ok(None);
480            } else {
481                return Err(io::Error::new(
482                    io::ErrorKind::InvalidInput,
483                    "invalid message length: datarowrange is not empty",
484                ));
485            }
486        }
487
488        self.remaining -= 1;
489        let len = self.buf.read_i32::<BigEndian>()?;
490        if len < 0 {
491            Ok(Some(None))
492        } else {
493            let len = len as usize;
494            if self.buf.len() < len {
495                return Err(io::Error::new(
496                    io::ErrorKind::UnexpectedEof,
497                    "unexpected EOF",
498                ));
499            }
500            let base = self.len - self.buf.len();
501            self.buf = &self.buf[len..];
502            Ok(Some(Some(base..base + len)))
503        }
504    }
505
506    #[inline]
507    fn size_hint(&self) -> (usize, Option<usize>) {
508        let len = self.remaining as usize;
509        (len, Some(len))
510    }
511}
512
513pub struct ErrorResponseBody {
514    storage: Bytes,
515}
516
517impl ErrorResponseBody {
518    #[inline]
519    pub fn fields(&self) -> ErrorFields<'_> {
520        ErrorFields { buf: &self.storage }
521    }
522}
523
524pub struct ErrorFields<'a> {
525    buf: &'a [u8],
526}
527
528impl<'a> FallibleIterator for ErrorFields<'a> {
529    type Item = ErrorField<'a>;
530    type Error = io::Error;
531
532    #[inline]
533    fn next(&mut self) -> io::Result<Option<ErrorField<'a>>> {
534        let type_ = self.buf.read_u8()?;
535        if type_ == 0 {
536            if self.buf.is_empty() {
537                return Ok(None);
538            } else {
539                return Err(io::Error::new(
540                    io::ErrorKind::InvalidInput,
541                    "invalid message length: error fields is not drained",
542                ));
543            }
544        }
545
546        let value_end = find_null(self.buf, 0)?;
547        let value = &self.buf[..value_end];
548        self.buf = &self.buf[value_end + 1..];
549
550        Ok(Some(ErrorField { type_, value }))
551    }
552}
553
554pub struct ErrorField<'a> {
555    type_: u8,
556    value: &'a [u8],
557}
558
559impl ErrorField<'_> {
560    #[inline]
561    pub fn type_(&self) -> u8 {
562        self.type_
563    }
564
565    #[inline]
566    #[deprecated(note = "use value_bytes instead", since = "0.6.7")]
567    pub fn value(&self) -> &str {
568        str::from_utf8(self.value).expect("error field value contained non-UTF8 bytes")
569    }
570
571    #[inline]
572    pub fn value_bytes(&self) -> &[u8] {
573        self.value
574    }
575}
576
577pub struct NoticeResponseBody {
578    storage: Bytes,
579}
580
581impl NoticeResponseBody {
582    #[inline]
583    pub fn fields(&self) -> ErrorFields<'_> {
584        ErrorFields { buf: &self.storage }
585    }
586}
587
588pub struct NotificationResponseBody {
589    process_id: i32,
590    channel: Bytes,
591    message: Bytes,
592}
593
594impl NotificationResponseBody {
595    #[inline]
596    pub fn process_id(&self) -> i32 {
597        self.process_id
598    }
599
600    #[inline]
601    pub fn channel(&self) -> io::Result<&str> {
602        get_str(&self.channel)
603    }
604
605    #[inline]
606    pub fn message(&self) -> io::Result<&str> {
607        get_str(&self.message)
608    }
609}
610
611pub struct ParameterDescriptionBody {
612    storage: Bytes,
613    len: u16,
614}
615
616impl ParameterDescriptionBody {
617    #[inline]
618    pub fn parameters(&self) -> Parameters<'_> {
619        Parameters {
620            buf: &self.storage,
621            remaining: self.len,
622        }
623    }
624}
625
626pub struct Parameters<'a> {
627    buf: &'a [u8],
628    remaining: u16,
629}
630
631impl FallibleIterator for Parameters<'_> {
632    type Item = Oid;
633    type Error = io::Error;
634
635    #[inline]
636    fn next(&mut self) -> io::Result<Option<Oid>> {
637        if self.remaining == 0 {
638            if self.buf.is_empty() {
639                return Ok(None);
640            } else {
641                return Err(io::Error::new(
642                    io::ErrorKind::InvalidInput,
643                    "invalid message length: parameters is not drained",
644                ));
645            }
646        }
647
648        self.remaining -= 1;
649        self.buf.read_u32::<BigEndian>().map(Some)
650    }
651
652    #[inline]
653    fn size_hint(&self) -> (usize, Option<usize>) {
654        let len = self.remaining as usize;
655        (len, Some(len))
656    }
657}
658
659pub struct ParameterStatusBody {
660    name: Bytes,
661    value: Bytes,
662}
663
664impl ParameterStatusBody {
665    #[inline]
666    pub fn name(&self) -> io::Result<&str> {
667        get_str(&self.name)
668    }
669
670    #[inline]
671    pub fn value(&self) -> io::Result<&str> {
672        get_str(&self.value)
673    }
674}
675
676pub struct ReadyForQueryBody {
677    status: u8,
678}
679
680impl ReadyForQueryBody {
681    #[inline]
682    pub fn status(&self) -> u8 {
683        self.status
684    }
685}
686
687pub struct RowDescriptionBody {
688    storage: Bytes,
689    len: u16,
690}
691
692impl RowDescriptionBody {
693    #[inline]
694    pub fn fields(&self) -> Fields<'_> {
695        Fields {
696            buf: &self.storage,
697            remaining: self.len,
698        }
699    }
700}
701
702pub struct Fields<'a> {
703    buf: &'a [u8],
704    remaining: u16,
705}
706
707impl<'a> FallibleIterator for Fields<'a> {
708    type Item = Field<'a>;
709    type Error = io::Error;
710
711    #[inline]
712    fn next(&mut self) -> io::Result<Option<Field<'a>>> {
713        if self.remaining == 0 {
714            if self.buf.is_empty() {
715                return Ok(None);
716            } else {
717                return Err(io::Error::new(
718                    io::ErrorKind::InvalidInput,
719                    "invalid message length: field is not drained",
720                ));
721            }
722        }
723
724        self.remaining -= 1;
725        let name_end = find_null(self.buf, 0)?;
726        let name = get_str(&self.buf[..name_end])?;
727        self.buf = &self.buf[name_end + 1..];
728        let table_oid = self.buf.read_u32::<BigEndian>()?;
729        let column_id = self.buf.read_i16::<BigEndian>()?;
730        let type_oid = self.buf.read_u32::<BigEndian>()?;
731        let type_size = self.buf.read_i16::<BigEndian>()?;
732        let type_modifier = self.buf.read_i32::<BigEndian>()?;
733        let format = self.buf.read_i16::<BigEndian>()?;
734
735        Ok(Some(Field {
736            name,
737            table_oid,
738            column_id,
739            type_oid,
740            type_size,
741            type_modifier,
742            format,
743        }))
744    }
745}
746
747pub struct Field<'a> {
748    name: &'a str,
749    table_oid: Oid,
750    column_id: i16,
751    type_oid: Oid,
752    type_size: i16,
753    type_modifier: i32,
754    format: i16,
755}
756
757impl<'a> Field<'a> {
758    #[inline]
759    pub fn name(&self) -> &'a str {
760        self.name
761    }
762
763    #[inline]
764    pub fn table_oid(&self) -> Oid {
765        self.table_oid
766    }
767
768    #[inline]
769    pub fn column_id(&self) -> i16 {
770        self.column_id
771    }
772
773    #[inline]
774    pub fn type_oid(&self) -> Oid {
775        self.type_oid
776    }
777
778    #[inline]
779    pub fn type_size(&self) -> i16 {
780        self.type_size
781    }
782
783    #[inline]
784    pub fn type_modifier(&self) -> i32 {
785        self.type_modifier
786    }
787
788    #[inline]
789    pub fn format(&self) -> i16 {
790        self.format
791    }
792}
793
794#[inline]
795fn find_null(buf: &[u8], start: usize) -> io::Result<usize> {
796    match memchr(0, &buf[start..]) {
797        Some(pos) => Ok(pos + start),
798        None => Err(io::Error::new(
799            io::ErrorKind::UnexpectedEof,
800            "unexpected EOF",
801        )),
802    }
803}
804
805#[inline]
806fn get_str(buf: &[u8]) -> io::Result<&str> {
807    str::from_utf8(buf).map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))
808}
postgres_protocol/message/backend.rs

postgres_protocol/message/
backend.rs
