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 #[derive(Default,Clone)]
   8 pub struct PacketQueue<D> {
   9   queue: VecDeque<D>,
  10   content: usize,
  11 }
  12
  13 impl<D> PacketQueue<D> where D: AsRef<[u8]> {
  14   pub fn push_back(&mut self, data: D) {
  15     self.content += data.as_ref().len();
  16     self.queue.push_back(data);
  17   }
  18
  19   pub fn pop_front(&mut self) -> Option<D> {
  20     let data = self.queue.pop_front()?;
  21     self.content -= data.as_ref().len();
  22     Some(data)
  23   }
  24
  25   pub fn content_count(&self) -> usize { self.queue.len() }
  26   pub fn content_len(&self) -> usize { self.content }
  27   pub fn total_len(&self) -> usize {
  28     self.content_count() + self.content_len()
  29   }
  30
  31   pub fn is_empty(&self) -> bool { self.queue.is_empty() }
  32   pub fn peek_front(&self) -> Option<&D> { self.queue.front() }
  33 }
  34
  35 #[derive(Default,Clone)]
  36 pub struct QueueBuf<E> {
  37   content: usize,
  38   eaten1: usize, // 0 <= eaten1 < queue.front()...len()
  39   queue: VecDeque<E>,
  40 }
  41
  42 #[derive(Default,Debug,Clone)]
  43 pub struct FrameQueueBuf {
  44   queue: QueueBuf<Cervine<'static, Box<[u8]>, [u8]>>,
  45 }
  46
  47 impl<E> Debug for QueueBuf<E> where E: AsRef<[u8]> {
  48   #[throws(fmt::Error)]
  49   fn fmt(&self, f: &mut fmt::Formatter) {
  50     write!(f, "Queue{{content={},eaten1={},queue=[",
  51            self.content, self.eaten1)?;
  52     for q in &self.queue { write!(f, "{},", q.as_ref().len())?; }
  53     write!(f, "]}}")?;
  54   }
  55 }
  56
  57 impl<E> QueueBuf<E> where E: AsRef<[u8]> {
  58   pub fn push<B: Into<E>>(&mut self, b: B) {
  59     self.push_(b.into());
  60   }
  61   fn push_(&mut self, b: E) {
  62     let l = b.as_ref().len();
  63     self.queue.push_back(b);
  64     self.content += l;
  65   }
  66   pub fn is_empty(&self) -> bool { self.content == 0 }
  67   pub fn len(&self) -> usize { self.content }
  68 }
  69
  70 impl FrameQueueBuf {
  71   pub fn push_esc<B: Into<Box<[u8]>>>(&mut self, b: B) {
  72     self.push_esc_(b.into());
  73   }
  74   fn push_esc_(&mut self, b: Box<[u8]>) {
  75     self.queue.push_(Cervine::Owned(b));
  76     self.queue.push_(Cervine::Borrowed(&SLIP_END_SLICE));
  77   }
  78   pub fn esc_push(&mut self, b: Box<[u8]>) {
  79     self.queue.push_(Cervine::Borrowed(&SLIP_END_SLICE));
  80     self.queue.push_(Cervine::Owned(b));
  81   }
  82   pub fn push_raw(&mut self, b: Box<[u8]>) {
  83     self.queue.push_(Cervine::Owned(b));
  84   }
  85   pub fn is_empty(&self) -> bool { self.queue.is_empty() }
  86   pub fn len(&self) -> usize { self.queue.len() }
  87 }
  88
  89 impl<E> hyper::body::Buf for QueueBuf<E> where E: AsRef<[u8]> {
  90   fn remaining(&self) -> usize { self.content }
  91   fn chunk(&self) -> &[u8] {
  92     let front = if let Some(f) = self.queue.front() { f } else { return &[] };
  93     &front.as_ref()[ self.eaten1.. ]
  94   }
  95   fn advance(&mut self, cnt: usize) {
  96     self.content -= cnt;
  97     self.eaten1 += cnt;
  98     loop {
  99       if self.eaten1 == 0 { break }
 100       let front = self.queue.front().unwrap();
 101       if self.eaten1 < front.as_ref().len() { break; }
 102       self.eaten1 -= front.as_ref().len();
 103       self.queue.pop_front().unwrap();
 104     }
 105   }
 106 }
 107
 108 impl hyper::body::Buf for FrameQueueBuf {
 109   fn remaining(&self) -> usize { self.queue.remaining() }
 110   fn chunk(&self) -> &[u8] { self.queue.chunk() }
 111   fn advance(&mut self, cnt: usize) { self.queue.advance(cnt) }
 112 }
 113
 114 pin_project!{
 115   pub struct BufBody<B:Buf> {
 116     body: Option<B>,
 117   }
 118 }
 119 impl<B:Buf> BufBody<B> {
 120   pub fn new(body: B) -> Self { Self { body: Some(body ) } }
 121 }
 122 impl BufBody<FrameQueueBuf> {
 123   pub fn display<S:Display>(s: S) -> Self {
 124     let s = s.to_string().into_bytes();
 125     let mut buf: FrameQueueBuf = default();
 126     buf.push_raw(s.into());
 127     Self::new(buf)
 128   }
 129 }
 130
 131 impl<B:Buf> HttpBody for BufBody<B> {
 132   type Error = Void;
 133   type Data = B;
 134   fn poll_data(self: Pin<&mut Self>, _: &mut std::task::Context<'_>)
 135                -> Poll<Option<Result<B, Void>>> {
 136     Poll::Ready(Ok(self.project().body.take()).transpose())
 137   }
 138   fn poll_trailers(self: Pin<&mut Self>, _: &mut std::task::Context<'_>)
 139  -> Poll<Result<Option<hyper::HeaderMap<hyper::header::HeaderValue>>, Void>> {
 140     Poll::Ready(Ok(None))
 141   }
 142 }