src/queue.rs

   1 // Copyright 2021 Ian Jackson and contributors to Hippotat
   2 // SPDX-License-Identifier: GPL-3.0-or-later
   3 // There is NO WARRANTY.
   4
   5 use crate::prelude::*;
   6
   7 // xxx are we using this at all ?
   8 #[derive(Default,Clone)]
   9 pub struct PacketQueue<D> {
  10   queue: VecDeque<D>,
  11   content: usize,
  12 }
  13
  14 impl<D> PacketQueue<D> where D: AsRef<[u8]> {
  15   pub fn push_back(&mut self, data: D) {
  16     self.content += data.as_ref().len();
  17     self.queue.push_back(data);
  18   }
  19
  20   pub fn pop_front(&mut self) -> Option<D> {
  21     let data = self.queue.pop_front()?;
  22     self.content -= data.as_ref().len();
  23     Some(data)
  24   }
  25
  26   pub fn is_empty(&self) -> bool { self.queue.is_empty() }
  27   pub fn peek_front(&self) -> Option<&D> { self.queue.front() }
  28 }
  29
  30 #[derive(Default,Clone)]
  31 pub struct QueueBuf<E> {
  32   content: usize,
  33   eaten1: usize, // 0 <= eaten1 < queue.front()...len()
  34   queue: VecDeque<E>,
  35 }
  36
  37 #[derive(Default,Debug,Clone)]
  38 pub struct FrameQueueBuf {
  39   queue: QueueBuf<Cervine<'static, Box<[u8]>, [u8]>>,
  40 }
  41
  42 impl<E> Debug for QueueBuf<E> where E: AsRef<[u8]> {
  43   #[throws(fmt::Error)]
  44   fn fmt(&self, f: &mut fmt::Formatter) {
  45     write!(f, "Queue{{content={},eaten1={},queue=[",
  46            self.content, self.eaten1)?;
  47     for q in &self.queue { write!(f, "{},", q.as_ref().len())?; }
  48     write!(f, "]}}")?;
  49   }
  50 }
  51
  52 impl<E> QueueBuf<E> where E: AsRef<[u8]> {
  53   pub fn push<B: Into<E>>(&mut self, b: B) {
  54     self.push_(b.into());
  55   }
  56   pub fn push_(&mut self, b: E) {
  57     let l = b.as_ref().len();
  58     self.queue.push_back(b);
  59     self.content += l;
  60   }
  61   pub fn is_empty(&self) -> bool { self.content == 0 }
  62 }
  63
  64 impl FrameQueueBuf {
  65   pub fn push<B: Into<Box<[u8]>>>(&mut self, b: B) {
  66     self.push_(b.into());
  67   }
  68   pub fn push_(&mut self, b: Box<[u8]>) {
  69     self.queue.push_(Cervine::Owned(b));
  70     self.queue.push_(Cervine::Borrowed(&SLIP_END_SLICE));
  71   }
  72   pub fn is_empty(&self) -> bool { self.queue.is_empty() }
  73 }
  74
  75 impl<E> Extend<E> for FrameQueueBuf where E: Into<Box<[u8]>> {
  76   fn extend<I>(&mut self, it: I)
  77   where I: IntoIterator<Item=E>
  78   {
  79     for b in it { self.push(b) }
  80   }
  81 }
  82
  83 impl<E> hyper::body::Buf for QueueBuf<E> where E: AsRef<[u8]> {
  84   fn remaining(&self) -> usize { self.content }
  85   fn chunk(&self) -> &[u8] {
  86     let front = if let Some(f) = self.queue.front() { f } else { return &[] };
  87     &front.as_ref()[ self.eaten1.. ]
  88   }
  89   fn advance(&mut self, cnt: usize) {
  90     self.content -= cnt;
  91     self.eaten1 += cnt;
  92     loop {
  93       if self.eaten1 == 0 { break }
  94       let front = self.queue.front().unwrap();
  95       if self.eaten1 < front.as_ref().len() { break; }
  96       self.eaten1 -= front.as_ref().len();
  97       self.queue.pop_front().unwrap();
  98     }
  99   }
 100 }
 101
 102 impl hyper::body::Buf for FrameQueueBuf {
 103   fn remaining(&self) -> usize { self.queue.remaining() }
 104   fn chunk(&self) -> &[u8] { self.queue.chunk() }
 105   fn advance(&mut self, cnt: usize) { self.queue.advance(cnt) }
 106 }