1use byteorder::{BigEndian, ByteOrder, ReadBytesExt};
3use bytes::{BufMut, BytesMut};
4use fallible_iterator::FallibleIterator;
5use std::boxed::Box as StdBox;
6use std::error::Error;
7use std::io::Read;
8use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
9use std::str;
10
11use crate::{write_nullable, FromUsize, IsNull, Lsn, Oid};
12
13#[cfg(test)]
14mod test;
15
16const RANGE_UPPER_UNBOUNDED: u8 = 0b0001_0000;
17const RANGE_LOWER_UNBOUNDED: u8 = 0b0000_1000;
18const RANGE_UPPER_INCLUSIVE: u8 = 0b0000_0100;
19const RANGE_LOWER_INCLUSIVE: u8 = 0b0000_0010;
20const RANGE_EMPTY: u8 = 0b0000_0001;
21
22const PGSQL_AF_INET: u8 = 2;
23const PGSQL_AF_INET6: u8 = 3;
24
25#[inline]
27pub fn bool_to_sql(v: bool, buf: &mut BytesMut) {
28 buf.put_u8(v as u8);
29}
30
31#[inline]
33pub fn bool_from_sql(buf: &[u8]) -> Result<bool, StdBox<dyn Error + Sync + Send>> {
34 if buf.len() != 1 {
35 return Err("invalid buffer size".into());
36 }
37
38 Ok(buf[0] != 0)
39}
40
41#[inline]
43pub fn bytea_to_sql(v: &[u8], buf: &mut BytesMut) {
44 buf.put_slice(v);
45}
46
47#[inline]
49pub fn bytea_from_sql(buf: &[u8]) -> &[u8] {
50 buf
51}
52
53#[inline]
55pub fn text_to_sql(v: &str, buf: &mut BytesMut) {
56 buf.put_slice(v.as_bytes());
57}
58
59#[inline]
61pub fn text_from_sql(buf: &[u8]) -> Result<&str, StdBox<dyn Error + Sync + Send>> {
62 Ok(str::from_utf8(buf)?)
63}
64
65#[inline]
67pub fn char_to_sql(v: i8, buf: &mut BytesMut) {
68 buf.put_i8(v);
69}
70
71#[inline]
73pub fn char_from_sql(mut buf: &[u8]) -> Result<i8, StdBox<dyn Error + Sync + Send>> {
74 let v = buf.read_i8()?;
75 if !buf.is_empty() {
76 return Err("invalid buffer size".into());
77 }
78 Ok(v)
79}
80
81#[inline]
83pub fn int2_to_sql(v: i16, buf: &mut BytesMut) {
84 buf.put_i16(v);
85}
86
87#[inline]
89pub fn int2_from_sql(mut buf: &[u8]) -> Result<i16, StdBox<dyn Error + Sync + Send>> {
90 let v = buf.read_i16::<BigEndian>()?;
91 if !buf.is_empty() {
92 return Err("invalid buffer size".into());
93 }
94 Ok(v)
95}
96
97#[inline]
99pub fn int4_to_sql(v: i32, buf: &mut BytesMut) {
100 buf.put_i32(v);
101}
102
103#[inline]
105pub fn int4_from_sql(mut buf: &[u8]) -> Result<i32, StdBox<dyn Error + Sync + Send>> {
106 let v = buf.read_i32::<BigEndian>()?;
107 if !buf.is_empty() {
108 return Err("invalid buffer size".into());
109 }
110 Ok(v)
111}
112
113#[inline]
115pub fn oid_to_sql(v: Oid, buf: &mut BytesMut) {
116 buf.put_u32(v);
117}
118
119#[inline]
121pub fn oid_from_sql(mut buf: &[u8]) -> Result<Oid, StdBox<dyn Error + Sync + Send>> {
122 let v = buf.read_u32::<BigEndian>()?;
123 if !buf.is_empty() {
124 return Err("invalid buffer size".into());
125 }
126 Ok(v)
127}
128
129#[inline]
131pub fn int8_to_sql(v: i64, buf: &mut BytesMut) {
132 buf.put_i64(v);
133}
134
135#[inline]
137pub fn int8_from_sql(mut buf: &[u8]) -> Result<i64, StdBox<dyn Error + Sync + Send>> {
138 let v = buf.read_i64::<BigEndian>()?;
139 if !buf.is_empty() {
140 return Err("invalid buffer size".into());
141 }
142 Ok(v)
143}
144
145#[inline]
147pub fn lsn_to_sql(v: Lsn, buf: &mut BytesMut) {
148 buf.put_u64(v);
149}
150
151#[inline]
153pub fn lsn_from_sql(mut buf: &[u8]) -> Result<Lsn, StdBox<dyn Error + Sync + Send>> {
154 let v = buf.read_u64::<BigEndian>()?;
155 if !buf.is_empty() {
156 return Err("invalid buffer size".into());
157 }
158 Ok(v)
159}
160
161#[inline]
163pub fn float4_to_sql(v: f32, buf: &mut BytesMut) {
164 buf.put_f32(v);
165}
166
167#[inline]
169pub fn float4_from_sql(mut buf: &[u8]) -> Result<f32, StdBox<dyn Error + Sync + Send>> {
170 let v = buf.read_f32::<BigEndian>()?;
171 if !buf.is_empty() {
172 return Err("invalid buffer size".into());
173 }
174 Ok(v)
175}
176
177#[inline]
179pub fn float8_to_sql(v: f64, buf: &mut BytesMut) {
180 buf.put_f64(v);
181}
182
183#[inline]
185pub fn float8_from_sql(mut buf: &[u8]) -> Result<f64, StdBox<dyn Error + Sync + Send>> {
186 let v = buf.read_f64::<BigEndian>()?;
187 if !buf.is_empty() {
188 return Err("invalid buffer size".into());
189 }
190 Ok(v)
191}
192
193#[inline]
195pub fn hstore_to_sql<'a, I>(
196 values: I,
197 buf: &mut BytesMut,
198) -> Result<(), StdBox<dyn Error + Sync + Send>>
199where
200 I: IntoIterator<Item = (&'a str, Option<&'a str>)>,
201{
202 let base = buf.len();
203 buf.put_i32(0);
204
205 let mut count = 0;
206 for (key, value) in values {
207 count += 1;
208
209 write_pascal_string(key, buf)?;
210
211 match value {
212 Some(value) => {
213 write_pascal_string(value, buf)?;
214 }
215 None => buf.put_i32(-1),
216 }
217 }
218
219 let count = i32::from_usize(count)?;
220 BigEndian::write_i32(&mut buf[base..], count);
221
222 Ok(())
223}
224
225fn write_pascal_string(s: &str, buf: &mut BytesMut) -> Result<(), StdBox<dyn Error + Sync + Send>> {
226 let size = i32::from_usize(s.len())?;
227 buf.put_i32(size);
228 buf.put_slice(s.as_bytes());
229 Ok(())
230}
231
232#[inline]
234pub fn hstore_from_sql(
235 mut buf: &[u8],
236) -> Result<HstoreEntries<'_>, StdBox<dyn Error + Sync + Send>> {
237 let count = buf.read_i32::<BigEndian>()?;
238 if count < 0 {
239 return Err("invalid entry count".into());
240 }
241
242 Ok(HstoreEntries {
243 remaining: count,
244 buf,
245 })
246}
247
248pub struct HstoreEntries<'a> {
250 remaining: i32,
251 buf: &'a [u8],
252}
253
254impl<'a> FallibleIterator for HstoreEntries<'a> {
255 type Item = (&'a str, Option<&'a str>);
256 type Error = StdBox<dyn Error + Sync + Send>;
257
258 #[inline]
259 #[allow(clippy::type_complexity)]
260 fn next(
261 &mut self,
262 ) -> Result<Option<(&'a str, Option<&'a str>)>, StdBox<dyn Error + Sync + Send>> {
263 if self.remaining == 0 {
264 if !self.buf.is_empty() {
265 return Err("invalid buffer size".into());
266 }
267 return Ok(None);
268 }
269
270 self.remaining -= 1;
271
272 let key_len = self.buf.read_i32::<BigEndian>()?;
273 if key_len < 0 {
274 return Err("invalid key length".into());
275 }
276 let (key, buf) = self.buf.split_at(key_len as usize);
277 let key = str::from_utf8(key)?;
278 self.buf = buf;
279
280 let value_len = self.buf.read_i32::<BigEndian>()?;
281 let value = if value_len < 0 {
282 None
283 } else {
284 let (value, buf) = self.buf.split_at(value_len as usize);
285 let value = str::from_utf8(value)?;
286 self.buf = buf;
287 Some(value)
288 };
289
290 Ok(Some((key, value)))
291 }
292
293 #[inline]
294 fn size_hint(&self) -> (usize, Option<usize>) {
295 let len = self.remaining as usize;
296 (len, Some(len))
297 }
298}
299
300#[inline]
302pub fn varbit_to_sql<I>(
303 len: usize,
304 v: I,
305 buf: &mut BytesMut,
306) -> Result<(), StdBox<dyn Error + Sync + Send>>
307where
308 I: Iterator<Item = u8>,
309{
310 let len = i32::from_usize(len)?;
311 buf.put_i32(len);
312
313 for byte in v {
314 buf.put_u8(byte);
315 }
316
317 Ok(())
318}
319
320#[inline]
322pub fn varbit_from_sql(mut buf: &[u8]) -> Result<Varbit<'_>, StdBox<dyn Error + Sync + Send>> {
323 let len = buf.read_i32::<BigEndian>()?;
324 if len < 0 {
325 return Err("invalid varbit length: varbit < 0".into());
326 }
327 let bytes = (len as usize).div_ceil(8);
328 if buf.len() != bytes {
329 return Err("invalid message length: varbit mismatch".into());
330 }
331
332 Ok(Varbit {
333 len: len as usize,
334 bytes: buf,
335 })
336}
337
338pub struct Varbit<'a> {
340 len: usize,
341 bytes: &'a [u8],
342}
343
344impl<'a> Varbit<'a> {
345 #[inline]
347 pub fn len(&self) -> usize {
348 self.len
349 }
350
351 #[inline]
353 pub fn is_empty(&self) -> bool {
354 self.len == 0
355 }
356
357 #[inline]
359 pub fn bytes(&self) -> &'a [u8] {
360 self.bytes
361 }
362}
363
364#[inline]
368pub fn timestamp_to_sql(v: i64, buf: &mut BytesMut) {
369 buf.put_i64(v);
370}
371
372#[inline]
376pub fn timestamp_from_sql(mut buf: &[u8]) -> Result<i64, StdBox<dyn Error + Sync + Send>> {
377 let v = buf.read_i64::<BigEndian>()?;
378 if !buf.is_empty() {
379 return Err("invalid message length: timestamp not drained".into());
380 }
381 Ok(v)
382}
383
384#[inline]
388pub fn date_to_sql(v: i32, buf: &mut BytesMut) {
389 buf.put_i32(v);
390}
391
392#[inline]
396pub fn date_from_sql(mut buf: &[u8]) -> Result<i32, StdBox<dyn Error + Sync + Send>> {
397 let v = buf.read_i32::<BigEndian>()?;
398 if !buf.is_empty() {
399 return Err("invalid message length: date not drained".into());
400 }
401 Ok(v)
402}
403
404#[inline]
408pub fn time_to_sql(v: i64, buf: &mut BytesMut) {
409 buf.put_i64(v);
410}
411
412#[inline]
416pub fn time_from_sql(mut buf: &[u8]) -> Result<i64, StdBox<dyn Error + Sync + Send>> {
417 let v = buf.read_i64::<BigEndian>()?;
418 if !buf.is_empty() {
419 return Err("invalid message length: time not drained".into());
420 }
421 Ok(v)
422}
423
424#[inline]
426pub fn macaddr_to_sql(v: [u8; 6], buf: &mut BytesMut) {
427 buf.put_slice(&v);
428}
429
430#[inline]
432pub fn macaddr_from_sql(buf: &[u8]) -> Result<[u8; 6], StdBox<dyn Error + Sync + Send>> {
433 if buf.len() != 6 {
434 return Err("invalid message length: macaddr length mismatch".into());
435 }
436 let mut out = [0; 6];
437 out.copy_from_slice(buf);
438 Ok(out)
439}
440
441#[inline]
443pub fn uuid_to_sql(v: [u8; 16], buf: &mut BytesMut) {
444 buf.put_slice(&v);
445}
446
447#[inline]
449pub fn uuid_from_sql(buf: &[u8]) -> Result<[u8; 16], StdBox<dyn Error + Sync + Send>> {
450 if buf.len() != 16 {
451 return Err("invalid message length: uuid size mismatch".into());
452 }
453 let mut out = [0; 16];
454 out.copy_from_slice(buf);
455 Ok(out)
456}
457
458#[inline]
460pub fn array_to_sql<T, I, J, F>(
461 dimensions: I,
462 element_type: Oid,
463 elements: J,
464 mut serializer: F,
465 buf: &mut BytesMut,
466) -> Result<(), StdBox<dyn Error + Sync + Send>>
467where
468 I: IntoIterator<Item = ArrayDimension>,
469 J: IntoIterator<Item = T>,
470 F: FnMut(T, &mut BytesMut) -> Result<IsNull, StdBox<dyn Error + Sync + Send>>,
471{
472 let dimensions_idx = buf.len();
473 buf.put_i32(0);
474 let flags_idx = buf.len();
475 buf.put_i32(0);
476 buf.put_u32(element_type);
477
478 let mut num_dimensions = 0;
479 for dimension in dimensions {
480 num_dimensions += 1;
481 buf.put_i32(dimension.len);
482 buf.put_i32(dimension.lower_bound);
483 }
484
485 let num_dimensions = i32::from_usize(num_dimensions)?;
486 BigEndian::write_i32(&mut buf[dimensions_idx..], num_dimensions);
487
488 let mut has_nulls = false;
489 for element in elements {
490 write_nullable(
491 |buf| {
492 let r = serializer(element, buf);
493 if let Ok(IsNull::Yes) = r {
494 has_nulls = true;
495 }
496 r
497 },
498 buf,
499 )?;
500 }
501
502 BigEndian::write_i32(&mut buf[flags_idx..], has_nulls as i32);
503
504 Ok(())
505}
506
507#[inline]
509pub fn array_from_sql(mut buf: &[u8]) -> Result<Array<'_>, StdBox<dyn Error + Sync + Send>> {
510 let dimensions = buf.read_i32::<BigEndian>()?;
511 if dimensions < 0 {
512 return Err("invalid dimension count".into());
513 }
514 let has_nulls = buf.read_i32::<BigEndian>()? != 0;
515 let element_type = buf.read_u32::<BigEndian>()?;
516
517 let mut r = buf;
518 let mut elements = 1i32;
519 for _ in 0..dimensions {
520 let len = r.read_i32::<BigEndian>()?;
521 if len < 0 {
522 return Err("invalid dimension size".into());
523 }
524 let _lower_bound = r.read_i32::<BigEndian>()?;
525 elements = match elements.checked_mul(len) {
526 Some(elements) => elements,
527 None => return Err("too many array elements".into()),
528 };
529 }
530
531 if dimensions == 0 {
532 elements = 0;
533 }
534
535 Ok(Array {
536 dimensions,
537 has_nulls,
538 element_type,
539 elements,
540 buf,
541 })
542}
543
544pub struct Array<'a> {
546 dimensions: i32,
547 has_nulls: bool,
548 element_type: Oid,
549 elements: i32,
550 buf: &'a [u8],
551}
552
553impl<'a> Array<'a> {
554 #[inline]
556 pub fn has_nulls(&self) -> bool {
557 self.has_nulls
558 }
559
560 #[inline]
562 pub fn element_type(&self) -> Oid {
563 self.element_type
564 }
565
566 #[inline]
568 pub fn dimensions(&self) -> ArrayDimensions<'a> {
569 ArrayDimensions(&self.buf[..self.dimensions as usize * 8])
570 }
571
572 #[inline]
574 pub fn values(&self) -> ArrayValues<'a> {
575 ArrayValues {
576 remaining: self.elements,
577 buf: &self.buf[self.dimensions as usize * 8..],
578 }
579 }
580}
581
582pub struct ArrayDimensions<'a>(&'a [u8]);
584
585impl FallibleIterator for ArrayDimensions<'_> {
586 type Item = ArrayDimension;
587 type Error = StdBox<dyn Error + Sync + Send>;
588
589 #[inline]
590 fn next(&mut self) -> Result<Option<ArrayDimension>, StdBox<dyn Error + Sync + Send>> {
591 if self.0.is_empty() {
592 return Ok(None);
593 }
594
595 let len = self.0.read_i32::<BigEndian>()?;
596 let lower_bound = self.0.read_i32::<BigEndian>()?;
597
598 Ok(Some(ArrayDimension { len, lower_bound }))
599 }
600
601 #[inline]
602 fn size_hint(&self) -> (usize, Option<usize>) {
603 let len = self.0.len() / 8;
604 (len, Some(len))
605 }
606}
607
608#[derive(Debug, Copy, Clone, PartialEq, Eq)]
610pub struct ArrayDimension {
611 pub len: i32,
613
614 pub lower_bound: i32,
616}
617
618pub struct ArrayValues<'a> {
620 remaining: i32,
621 buf: &'a [u8],
622}
623
624impl<'a> FallibleIterator for ArrayValues<'a> {
625 type Item = Option<&'a [u8]>;
626 type Error = StdBox<dyn Error + Sync + Send>;
627
628 #[inline]
629 fn next(&mut self) -> Result<Option<Option<&'a [u8]>>, StdBox<dyn Error + Sync + Send>> {
630 if self.remaining == 0 {
631 if !self.buf.is_empty() {
632 return Err("invalid message length: arrayvalue not drained".into());
633 }
634 return Ok(None);
635 }
636 self.remaining -= 1;
637
638 let len = self.buf.read_i32::<BigEndian>()?;
639 let val = if len < 0 {
640 None
641 } else {
642 let (val, buf) = self
643 .buf
644 .split_at_checked(len as usize)
645 .ok_or("invalid value length")?;
646 self.buf = buf;
647 Some(val)
648 };
649
650 Ok(Some(val))
651 }
652
653 fn size_hint(&self) -> (usize, Option<usize>) {
654 let len = self.remaining as usize;
655 (len, Some(len))
656 }
657}
658
659#[inline]
661pub fn empty_range_to_sql(buf: &mut BytesMut) {
662 buf.put_u8(RANGE_EMPTY);
663}
664
665pub fn range_to_sql<F, G>(
667 lower: F,
668 upper: G,
669 buf: &mut BytesMut,
670) -> Result<(), StdBox<dyn Error + Sync + Send>>
671where
672 F: FnOnce(&mut BytesMut) -> Result<RangeBound<IsNull>, StdBox<dyn Error + Sync + Send>>,
673 G: FnOnce(&mut BytesMut) -> Result<RangeBound<IsNull>, StdBox<dyn Error + Sync + Send>>,
674{
675 let tag_idx = buf.len();
676 buf.put_u8(0);
677 let mut tag = 0;
678
679 match write_bound(lower, buf)? {
680 RangeBound::Inclusive(()) => tag |= RANGE_LOWER_INCLUSIVE,
681 RangeBound::Exclusive(()) => {}
682 RangeBound::Unbounded => tag |= RANGE_LOWER_UNBOUNDED,
683 }
684
685 match write_bound(upper, buf)? {
686 RangeBound::Inclusive(()) => tag |= RANGE_UPPER_INCLUSIVE,
687 RangeBound::Exclusive(()) => {}
688 RangeBound::Unbounded => tag |= RANGE_UPPER_UNBOUNDED,
689 }
690
691 buf[tag_idx] = tag;
692
693 Ok(())
694}
695
696fn write_bound<F>(
697 bound: F,
698 buf: &mut BytesMut,
699) -> Result<RangeBound<()>, StdBox<dyn Error + Sync + Send>>
700where
701 F: FnOnce(&mut BytesMut) -> Result<RangeBound<IsNull>, StdBox<dyn Error + Sync + Send>>,
702{
703 let base = buf.len();
704 buf.put_i32(0);
705
706 let (null, ret) = match bound(buf)? {
707 RangeBound::Inclusive(null) => (Some(null), RangeBound::Inclusive(())),
708 RangeBound::Exclusive(null) => (Some(null), RangeBound::Exclusive(())),
709 RangeBound::Unbounded => (None, RangeBound::Unbounded),
710 };
711
712 match null {
713 Some(null) => {
714 let len = match null {
715 IsNull::No => i32::from_usize(buf.len() - base - 4)?,
716 IsNull::Yes => -1,
717 };
718 BigEndian::write_i32(&mut buf[base..], len);
719 }
720 None => buf.truncate(base),
721 }
722
723 Ok(ret)
724}
725
726pub enum RangeBound<T> {
728 Inclusive(T),
730 Exclusive(T),
732 Unbounded,
734}
735
736#[inline]
738pub fn range_from_sql(mut buf: &[u8]) -> Result<Range<'_>, StdBox<dyn Error + Sync + Send>> {
739 let tag = buf.read_u8()?;
740
741 if tag == RANGE_EMPTY {
742 if !buf.is_empty() {
743 return Err("invalid message size".into());
744 }
745 return Ok(Range::Empty);
746 }
747
748 let lower = read_bound(&mut buf, tag, RANGE_LOWER_UNBOUNDED, RANGE_LOWER_INCLUSIVE)?;
749 let upper = read_bound(&mut buf, tag, RANGE_UPPER_UNBOUNDED, RANGE_UPPER_INCLUSIVE)?;
750
751 if !buf.is_empty() {
752 return Err("invalid message size".into());
753 }
754
755 Ok(Range::Nonempty(lower, upper))
756}
757
758#[inline]
759fn read_bound<'a>(
760 buf: &mut &'a [u8],
761 tag: u8,
762 unbounded: u8,
763 inclusive: u8,
764) -> Result<RangeBound<Option<&'a [u8]>>, StdBox<dyn Error + Sync + Send>> {
765 if tag & unbounded != 0 {
766 Ok(RangeBound::Unbounded)
767 } else {
768 let len = buf.read_i32::<BigEndian>()?;
769 let value = if len < 0 {
770 None
771 } else {
772 let len = len as usize;
773 let (value, tail) = buf.split_at_checked(len).ok_or("invalid message size")?;
774 *buf = tail;
775 Some(value)
776 };
777
778 if tag & inclusive != 0 {
779 Ok(RangeBound::Inclusive(value))
780 } else {
781 Ok(RangeBound::Exclusive(value))
782 }
783 }
784}
785
786pub enum Range<'a> {
788 Empty,
790 Nonempty(RangeBound<Option<&'a [u8]>>, RangeBound<Option<&'a [u8]>>),
792}
793
794#[inline]
796pub fn point_to_sql(x: f64, y: f64, buf: &mut BytesMut) {
797 buf.put_f64(x);
798 buf.put_f64(y);
799}
800
801#[inline]
803pub fn point_from_sql(mut buf: &[u8]) -> Result<Point, StdBox<dyn Error + Sync + Send>> {
804 let x = buf.read_f64::<BigEndian>()?;
805 let y = buf.read_f64::<BigEndian>()?;
806 if !buf.is_empty() {
807 return Err("invalid buffer size".into());
808 }
809 Ok(Point { x, y })
810}
811
812#[derive(Copy, Clone)]
814pub struct Point {
815 x: f64,
816 y: f64,
817}
818
819impl Point {
820 #[inline]
822 pub fn x(&self) -> f64 {
823 self.x
824 }
825
826 #[inline]
828 pub fn y(&self) -> f64 {
829 self.y
830 }
831}
832
833#[inline]
835pub fn box_to_sql(x1: f64, y1: f64, x2: f64, y2: f64, buf: &mut BytesMut) {
836 buf.put_f64(x1);
837 buf.put_f64(y1);
838 buf.put_f64(x2);
839 buf.put_f64(y2);
840}
841
842#[inline]
844pub fn box_from_sql(mut buf: &[u8]) -> Result<Box, StdBox<dyn Error + Sync + Send>> {
845 let x1 = buf.read_f64::<BigEndian>()?;
846 let y1 = buf.read_f64::<BigEndian>()?;
847 let x2 = buf.read_f64::<BigEndian>()?;
848 let y2 = buf.read_f64::<BigEndian>()?;
849 if !buf.is_empty() {
850 return Err("invalid buffer size".into());
851 }
852 Ok(Box {
853 upper_right: Point { x: x1, y: y1 },
854 lower_left: Point { x: x2, y: y2 },
855 })
856}
857
858#[derive(Copy, Clone)]
860pub struct Box {
861 upper_right: Point,
862 lower_left: Point,
863}
864
865impl Box {
866 #[inline]
868 pub fn upper_right(&self) -> Point {
869 self.upper_right
870 }
871
872 #[inline]
874 pub fn lower_left(&self) -> Point {
875 self.lower_left
876 }
877}
878
879#[inline]
881pub fn path_to_sql<I>(
882 closed: bool,
883 points: I,
884 buf: &mut BytesMut,
885) -> Result<(), StdBox<dyn Error + Sync + Send>>
886where
887 I: IntoIterator<Item = (f64, f64)>,
888{
889 buf.put_u8(closed as u8);
890 let points_idx = buf.len();
891 buf.put_i32(0);
892
893 let mut num_points = 0;
894 for (x, y) in points {
895 num_points += 1;
896 buf.put_f64(x);
897 buf.put_f64(y);
898 }
899
900 let num_points = i32::from_usize(num_points)?;
901 BigEndian::write_i32(&mut buf[points_idx..], num_points);
902
903 Ok(())
904}
905
906#[inline]
908pub fn path_from_sql(mut buf: &[u8]) -> Result<Path<'_>, StdBox<dyn Error + Sync + Send>> {
909 let closed = buf.read_u8()? != 0;
910 let points = buf.read_i32::<BigEndian>()?;
911
912 Ok(Path {
913 closed,
914 points,
915 buf,
916 })
917}
918
919pub struct Path<'a> {
921 closed: bool,
922 points: i32,
923 buf: &'a [u8],
924}
925
926impl<'a> Path<'a> {
927 #[inline]
929 pub fn closed(&self) -> bool {
930 self.closed
931 }
932
933 #[inline]
935 pub fn points(&self) -> PathPoints<'a> {
936 PathPoints {
937 remaining: self.points,
938 buf: self.buf,
939 }
940 }
941}
942
943pub struct PathPoints<'a> {
945 remaining: i32,
946 buf: &'a [u8],
947}
948
949impl FallibleIterator for PathPoints<'_> {
950 type Item = Point;
951 type Error = StdBox<dyn Error + Sync + Send>;
952
953 #[inline]
954 fn next(&mut self) -> Result<Option<Point>, StdBox<dyn Error + Sync + Send>> {
955 if self.remaining == 0 {
956 if !self.buf.is_empty() {
957 return Err("invalid message length: path points not drained".into());
958 }
959 return Ok(None);
960 }
961 self.remaining -= 1;
962
963 let x = self.buf.read_f64::<BigEndian>()?;
964 let y = self.buf.read_f64::<BigEndian>()?;
965
966 Ok(Some(Point { x, y }))
967 }
968
969 #[inline]
970 fn size_hint(&self) -> (usize, Option<usize>) {
971 let len = self.remaining as usize;
972 (len, Some(len))
973 }
974}
975
976#[inline]
978pub fn inet_to_sql(addr: IpAddr, netmask: u8, buf: &mut BytesMut) {
979 let family = match addr {
980 IpAddr::V4(_) => PGSQL_AF_INET,
981 IpAddr::V6(_) => PGSQL_AF_INET6,
982 };
983 buf.put_u8(family);
984 buf.put_u8(netmask);
985 buf.put_u8(0); match addr {
987 IpAddr::V4(addr) => {
988 buf.put_u8(4);
989 buf.put_slice(&addr.octets());
990 }
991 IpAddr::V6(addr) => {
992 buf.put_u8(16);
993 buf.put_slice(&addr.octets());
994 }
995 }
996}
997
998#[inline]
1000pub fn inet_from_sql(mut buf: &[u8]) -> Result<Inet, StdBox<dyn Error + Sync + Send>> {
1001 let family = buf.read_u8()?;
1002 let netmask = buf.read_u8()?;
1003 buf.read_u8()?; let len = buf.read_u8()?;
1005
1006 let addr = match family {
1007 PGSQL_AF_INET => {
1008 if netmask > 32 {
1009 return Err("invalid IPv4 netmask".into());
1010 }
1011 if len != 4 {
1012 return Err("invalid IPv4 address length".into());
1013 }
1014 let mut addr = [0; 4];
1015 buf.read_exact(&mut addr)?;
1016 IpAddr::V4(Ipv4Addr::from(addr))
1017 }
1018 PGSQL_AF_INET6 => {
1019 if netmask > 128 {
1020 return Err("invalid IPv6 netmask".into());
1021 }
1022 if len != 16 {
1023 return Err("invalid IPv6 address length".into());
1024 }
1025 let mut addr = [0; 16];
1026 buf.read_exact(&mut addr)?;
1027 IpAddr::V6(Ipv6Addr::from(addr))
1028 }
1029 _ => return Err("invalid IP family".into()),
1030 };
1031
1032 if !buf.is_empty() {
1033 return Err("invalid buffer size".into());
1034 }
1035
1036 Ok(Inet { addr, netmask })
1037}
1038
1039pub struct Inet {
1041 addr: IpAddr,
1042 netmask: u8,
1043}
1044
1045impl Inet {
1046 #[inline]
1048 pub fn addr(&self) -> IpAddr {
1049 self.addr
1050 }
1051
1052 #[inline]
1054 pub fn netmask(&self) -> u8 {
1055 self.netmask
1056 }
1057}
1058
1059#[inline]
1061pub fn ltree_to_sql(v: &str, buf: &mut BytesMut) {
1062 buf.put_u8(1);
1064 buf.put_slice(v.as_bytes());
1066}
1067
1068#[inline]
1070pub fn ltree_from_sql(buf: &[u8]) -> Result<&str, StdBox<dyn Error + Sync + Send>> {
1071 match buf {
1072 [1u8, rest @ ..] => Ok(str::from_utf8(rest)?),
1074 _ => Err("ltree version 1 only supported".into()),
1075 }
1076}
1077
1078#[inline]
1080pub fn lquery_to_sql(v: &str, buf: &mut BytesMut) {
1081 buf.put_u8(1);
1083 buf.put_slice(v.as_bytes());
1085}
1086
1087#[inline]
1089pub fn lquery_from_sql(buf: &[u8]) -> Result<&str, StdBox<dyn Error + Sync + Send>> {
1090 match buf {
1091 [1u8, rest @ ..] => Ok(str::from_utf8(rest)?),
1093 _ => Err("lquery version 1 only supported".into()),
1094 }
1095}
1096
1097#[inline]
1099pub fn ltxtquery_to_sql(v: &str, buf: &mut BytesMut) {
1100 buf.put_u8(1);
1102 buf.put_slice(v.as_bytes());
1104}
1105
1106#[inline]
1108pub fn ltxtquery_from_sql(buf: &[u8]) -> Result<&str, StdBox<dyn Error + Sync + Send>> {
1109 match buf {
1110 [1u8, rest @ ..] => Ok(str::from_utf8(rest)?),
1112 _ => Err("ltxtquery version 1 only supported".into()),
1113 }
1114}