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