blather 0.12.0

//! A [`tokio_util::codec`] Codec that is used to encode and decode the
//! blather protocol.

use std::{
  fmt,
  {cmp, collections::HashMap, mem}
};

use bytes::{BufMut, Bytes, BytesMut};

use {
  tokio::{io, sync::mpsc::UnboundedSender},
  tokio_util::codec::{Decoder, Encoder}
};

use crate::{
  err::Error,
  {KVLines, Params, Telegram}
};


/// Current state of decoder.
///
/// Controls what, if anything, will be returned to the application.
#[derive(Clone, Debug, PartialEq, Eq)]
enum CodecState {
  /// Read and decode a [`Telegram`] buffer from the network.
  Telegram,

  /// Read and decode a [`Params`] buffer from the network.
  Params,

  /// Read and decode an vector of key/value pairs.
  KVLines,

  /// Read a specified amount of raw bytes, and return it in chunks as they
  /// arrive.
  Chunks,

  /// Read a specified amount of raw bytes, and return the entire immutable
  /// buffer when it has arrived.
  Bytes,

  /// Transmit received bytes buffers to a channel.
  BytesCh,

  /// Ignore a specified amount of raw bytes.
  Skip
}

/// Data returned to the application when the Codec's Decode iterator is
/// called and the decoder has a complete entity to return.
pub enum Input {
  /// A complete [`Telegram`] has been received.
  Telegram(Telegram),

  /// A complete key/value lines buffer ([`KVLines`]) has been received.
  KVLines(KVLines),

  /// A complete [`Params`] has been received.
  Params(Params),

  /// A chunk of raw data has arrived.  The second argument is the amount of
  /// data remains, which has been adjusted for the current [`Bytes`].  If
  /// the `u64` parameter is 0 it means this is the final chunk.
  Chunk(Bytes, u64),

  /// A complete raw immutable buffer has been received.
  Bytes(Bytes),

  /// Sentinel value used to signal that transfer of buffers to a channel is
  /// done.
  BytesChDone,

  /// The requested number of bytes have been ignored.
  SkipDone
}


/// The Codec is used to keep track of the state of the inbound and outbound
/// communication.
pub struct Codec {
  next_line_index: usize,
  max_line_length: usize,
  tg: Telegram,
  params: Params,
  kvlines: KVLines,
  state: CodecState,
  remain: u64,
  bytes_tx: Option<UnboundedSender<Bytes>>
}

#[allow(clippy::missing_fields_in_debug)]
impl fmt::Debug for Codec {
  fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
    f.debug_struct("Codec").field("state", &self.state).finish()
  }
}

impl Default for Codec {
  fn default() -> Self {
    Self::new()
  }
}

/// A Codec used to encode and decode the blather protocol.
///
/// # Notes
/// Normally the Codec object is hidden inside a
/// [`Framed`](tokio_util::codec::Framed) object. In order to
/// call methods in the codec it must be accessed through the Framed object:
///
/// ```no_run
/// use tokio::net::TcpStream;
/// use tokio_util::codec::Framed;
/// use blather::Codec;
///
/// async fn do_something() {
///   let socket = TcpStream::connect("127.0.0.1:8080").await.unwrap();
///   let mut conn = Framed::new(socket, Codec::new());
///
///   // .. do stuff ..
///
///   let len = 8192;
///   conn.codec_mut().expect_bytes(len);
/// }
/// ```
impl Codec {
  /// Create a new `Codec`.  It will default to having not practical limit to
  /// the maximum line length and it will expect a [`Telegram`] buffer to
  /// arrive as the first frame.
  #[must_use]
  pub fn new() -> Self {
    Self {
      next_line_index: 0,
      max_line_length: usize::MAX,
      tg: Telegram::new_uninit(),
      params: Params::new(),
      kvlines: KVLines::new(),
      state: CodecState::Telegram,
      remain: 0,
      bytes_tx: None
    }
  }

  /// Create a new `Codec` with a specific maximum line length.  The default
  /// state will be to expect a [`Telegram`].
  #[must_use]
  pub fn new_with_max_length(max_line_length: usize) -> Self {
    Self {
      max_line_length,
      ..Self::new()
    }
  }

  /// Get the current maximum line length.
  #[must_use]
  pub const fn max_line_length(&self) -> usize {
    self.max_line_length
  }

  /// Determine how far into the buffer we'll search for a newline. If
  /// there's no `max_length` set, we'll read to the end of the buffer.
  fn find_newline(&self, buf: &BytesMut) -> (usize, Option<usize>) {
    let read_to = cmp::min(self.max_line_length.saturating_add(1), buf.len());
    let newline_offset = buf[self.next_line_index..read_to]
      .iter()
      .position(|b| *b == b'\n');

    (read_to, newline_offset)
  }

  /// This is called when `decode_telegram_lines` has encountered an eol,
  /// determined that the string is longer than zero characters, and thus
  /// passed the line to this function to process it.
  ///
  /// The first line received is a telegram topic.  This is a required line.
  /// Following lines are parameter lines, which are a single space character
  /// separated key/value pairs.
  fn decode_telegram_line(&mut self, line: &str) -> Result<(), Error> {
    if self.tg.get_topic().is_empty() {
      self
        .tg
        .set_topic(line)
        .map_err(|e| Error::Protocol(e.to_string()))?;
    } else {
      let idx = line.find(' ');
      if let Some(idx) = idx {
        let (k, v) = line.split_at(idx);
        let v = &v[1..v.len()];
        self.tg.add_param(k, v)?;
      }
    }
    Ok(())
  }

  /*
  fn getline_owned(
    &mut self,
    buf: &mut BytesMut
  ) -> Result<Option<String>, Error> {
    let (read_to, newline_offset) = self.find_newline(&buf);
    match newline_offset {
      Some(offset) => {
        // Found an eol
        let newline_index = offset + self.next_line_index;
        self.next_line_index = 0;
        let line = buf.split_to(newline_index + 1);
        let line = &line[..line.len() - 1];
        let line = utf8(without_carriage_return(line))?;

        Ok(Some(line.to_owned()))
      }
      None if buf.len() > self.max_line_length => Err(Error::BadFormat(
        "Exceeded maximum line length.".to_string()
      )),
      None => {
        // We didn't find a line or reach the length limit, so the next
        // call will resume searching at the current offset.
        self.next_line_index = read_to;

        // Returning Ok(None) instructs the FramedRead that more data is
        // needed.
        Ok(None)
      }
    }
  }
  */

  /// Get index of the next end of line in `buf`.
  fn get_eol_idx(&mut self, buf: &BytesMut) -> Result<Option<usize>, Error> {
    let (read_to, newline_offset) = self.find_newline(buf);
    match newline_offset {
      Some(offset) => {
        // Found an eol
        let newline_index = offset + self.next_line_index;
        self.next_line_index = 0;
        Ok(Some(newline_index + 1))
      }
      None if buf.len() > self.max_line_length => {
        Err(Error::Protocol("Exceeded maximum line length.".to_string()))
      }
      None => {
        // Didn't find a line or reach the length limit, so the next
        // call will resume searching at the current offset.
        self.next_line_index = read_to;

        // Returning Ok(None) instructs the FramedRead that more data is
        // needed.
        Ok(None)
      }
    }
  }

  /// (New) data is available in the input buffer.
  ///
  /// Try to parse lines until an empty line as been encountered, at which
  /// point the buffer is parsed and returned in an [`Telegram`] buffer.
  ///
  /// If the buffer doesn't contain enough data to finalize a complete telegram
  /// buffer return `Ok(None)` to inform the calling `FramedRead` that more
  /// data is needed.
  fn decode_telegram_lines(
    &mut self,
    buf: &mut BytesMut
  ) -> Result<Option<Telegram>, Error> {
    loop {
      if let Some(idx) = self.get_eol_idx(buf)? {
        let line = buf.split_to(idx);
        let line = &line[..line.len() - 1];
        let line = utf8(without_carriage_return(line))?;

        // Empty line marks end of Telegram
        if line.is_empty() {
          // mem::take() can replace a member of a struct.
          // (This requires Default to be implemented for the object being
          // taken).
          let newtg = Telegram::new_uninit();

          //return Ok(Some(mem::take(&mut self.tg)));
          return Ok(Some(mem::replace(&mut self.tg, newtg)));
        }
        self.decode_telegram_line(line)?;
      } else {
        // Returning Ok(None) instructs the FramedRead that more data is
        // needed.
        return Ok(None);
      }
    }
  }

  /// Read buffer line-by-line, split each line at the first space character
  /// and store the left part as a key and the right part as a value in a
  /// Params structure.
  fn decode_params_lines(
    &mut self,
    buf: &mut BytesMut
  ) -> Result<Option<Params>, Error> {
    loop {
      if let Some(idx) = self.get_eol_idx(buf)? {
        // Found an eol
        let line = buf.split_to(idx);
        let line = &line[..line.len() - 1];
        let line = utf8(without_carriage_return(line))?;

        // Empty line marks end of Params
        if line.is_empty() {
          // Revert to expecting a telegram once a Params has been completed.
          // The application can override this when needed.
          self.state = CodecState::Telegram;

          // mem::take() can replace a member of a struct.
          // (This requires Default to be implemented for the object being
          // taken).
          return Ok(Some(mem::take(&mut self.params)));
        }
        let idx = line.find(' ');
        if let Some(idx) = idx {
          let (k, v) = line.split_at(idx);
          let v = &v[1..v.len()];
          self.params.add_param(k, v)?;
        }
      } else {
        // Need more data
        return Ok(None);
      }
    }
  }

  /// Rea buffer line-by-line, split each at the first space character and
  /// store the left and right part in a vector.  When an empty line is
  /// encountered, return the vector and return to expecting a [`Telegram`].
  fn decode_kvlines(
    &mut self,
    buf: &mut BytesMut
  ) -> Result<Option<KVLines>, Error> {
    loop {
      if let Some(idx) = self.get_eol_idx(buf)? {
        // Found an eol
        let line = buf.split_to(idx);
        let line = &line[..line.len() - 1];
        let line = utf8(without_carriage_return(line))?;

        // Empty line marks end of Params
        if line.is_empty() {
          // Revert to expecting a telegram once a KVLines  has been
          // completed.
          // The application can override this when needed.
          self.state = CodecState::Telegram;

          // mem::take() can replace a member of a struct.
          // (This requires Default to be implemented for the object being
          // taken).
          return Ok(Some(mem::take(&mut self.kvlines)));
        }
        let idx = line.find(' ');
        if let Some(idx) = idx {
          let (k, v) = line.split_at(idx);
          let v = &v[1..v.len()];

          self.kvlines.append(k, v)?;
        }
      } else {
        // Need more data
        return Ok(None);
      }
    }
  }

  /// Set the decoder to treat the next `size` bytes as raw bytes to be
  /// received in chunks as `Bytes`.
  ///
  /// # Decoder behavior
  /// The decoder will return an [`Input::Chunk(buf, remain)`](Input::Chunk) to
  /// the application each time a new chunk has been received.  In addition to
  /// the actual chunk the number of bytes remaining will be returned.  The
  /// remaining bytes value is adjusted to subtract the currently returned
  /// chunk, which means that the application can detect the end of the
  /// buffer by checking if the remaining value is zero.
  ///
  /// Once the entire buffer has been received by the `Decoder` it will revert
  /// to expect an [`Input::Telegram`].
  ///
  /// # Errors
  /// [`Error::InvalidSize`] means the `size` parameter was set to `0`.
  pub fn expect_chunks(&mut self, size: u64) -> Result<(), Error> {
    if size == 0 {
      return Err(Error::InvalidSize("zero size".to_string()));
    }

    //println!("Expecting bin {}", size);
    self.state = CodecState::Chunks;
    self.remain = size;

    Ok(())
  }

  /// Expect a immutable buffer of a certain size to be received.
  ///
  /// The returned buffer will be stored in process memory.
  ///
  /// # Decoder behavior
  /// Once a complete buffer has been successfully reaceived the `Decoder` will
  /// return an [`Input::Bytes(b)`](Input::Bytes) where `b` is a
  /// [`bytes::Bytes`] containing the entire buffer.
  ///
  /// Once the entire buffer has been received by the `Decoder` it will revert
  /// to expect an [`Input::Telegram`].
  ///
  /// # Errors
  /// [`Error::InvalidSize`] means the `size` parameter was set to `0`.
  #[allow(clippy::missing_panics_doc)]
  pub fn expect_bytes(&mut self, size: usize) -> Result<(), Error> {
    if size == 0 {
      return Err(Error::InvalidSize("zero size".to_string()));
    }
    self.state = CodecState::Bytes;

    // unwrap() should be safe, unless running on a platform where
    // size_of::<usize>() > size_of::<u64>() and the buffer is larger than
    // usize::MAX.
    self.remain = size.try_into().unwrap();
    Ok(())
  }

  /// Expect specified amount of data which will be received as `Bytes` buffers
  /// and sent over a channel using a [`UnboundedSender`] end-point.
  ///
  /// # Errors
  /// [`Error::InvalidSize`] means the `size` parameter was set to `0`.
  pub fn expect_bytes_channel(
    &mut self,
    size: u64,
    tx: UnboundedSender<Bytes>
  ) -> Result<(), Error> {
    if size == 0 {
      return Err(Error::InvalidSize("must not be zero".to_string()));
    }
    self.state = CodecState::BytesCh;
    self.bytes_tx = Some(tx);
    self.remain = size;
    Ok(())
  }

  /// Tell the Decoder to expect lines of key/value pairs.
  ///
  /// # Decoder behavior
  /// On successful completion the the decoder will next return an
  /// [`Input::Params(params)`](Input::Params) once a complete `Params` buffer
  /// has been received.
  ///
  /// Once the entire buffer has been received by the `Decoder` it will revert
  /// to expect an [`Input::Telegram`].
  pub const fn expect_params(&mut self) {
    self.state = CodecState::Params;
  }

  /// Tell the Decoder to expect lines ordered key/value pairs.
  ///
  /// # Decoder behavior
  /// On successful completion the Framed `StreamExt::next()` will return an
  /// [`Input::KVLines(kvlines)`](Input::KVLines) once a complete `KVLines`
  /// buffer has been received.
  ///
  /// Once the entire buffer has been received by the `Decoder` it will revert
  /// to expect an [`Input::Telegram`].
  pub const fn expect_kvlines(&mut self) {
    self.state = CodecState::KVLines;
  }

  /// Skip a requested number of bytes.
  ///
  /// # Decoder behavior
  /// On successful completion the decoder will have ignored the specified
  /// number of byes, reverts back to waiting for a [`Input::Telegram`] and
  /// returns [`Input::SkipDone`].
  ///
  /// # Errors
  /// [`Error::InvalidSize`] means the `size` parameter was set to `0`.
  pub fn skip(&mut self, size: u64) -> Result<(), Error> {
    if size == 0 {
      return Err(Error::InvalidSize("zero size".to_string()));
    }
    self.state = CodecState::Skip;
    self.remain = size;
    Ok(())
  }
}

fn utf8(buf: &[u8]) -> Result<&str, io::Error> {
  std::str::from_utf8(buf).map_err(|_| {
    io::Error::new(
      io::ErrorKind::InvalidData,
      "Unable to decode input as UTF8"
    )
  })
}

fn without_carriage_return(s: &[u8]) -> &[u8] {
  if s.last() == Some(&b'\r') {
    &s[..s.len() - 1]
  } else {
    s
  }
}


/// A Decoder implementation that is used to assist in decoding data arriving
/// over a DDM client interface.
///
/// The default behavior for the Decoder is to wait for a Telegram buffer.  It
/// will, on success, return an `Input::Telegram(tg)`, where `tg` is a
/// `blather::Telegram` object.
impl Decoder for Codec {
  type Item = Input;
  type Error = crate::err::Error;

  fn decode(&mut self, buf: &mut BytesMut) -> Result<Option<Input>, Error> {
    // The codec's internal decoder state denotes whether lines or binary data
    // is currently being expected.
    match self.state {
      CodecState::Telegram => {
        // If decode_telegram_lines returns Some(value) it means that a
        // complete buffer has been received.
        let tg = self.decode_telegram_lines(buf)?;
        if let Some(tg) = tg {
          // A complete Telegram was received
          return Ok(Some(Input::Telegram(tg)));
        }

        // Returning Ok(None) tells the caller that we need more data
        Ok(None)
      }
      CodecState::Params => {
        // If decode_telegram_lines returns Some(value) it means that a
        // complete buffer has been received.
        let params = self.decode_params_lines(buf)?;
        if let Some(params) = params {
          // A complete Params buffer was received
          return Ok(Some(Input::Params(params)));
        }

        // Returning Ok(None) tells the caller that we need more data
        Ok(None)
      }
      CodecState::KVLines => {
        // If decode_telegram_lines returns Some(value) it means that a
        // complete buffer has been received.
        let kvlines = self.decode_kvlines(buf)?;
        if let Some(kvlines) = kvlines {
          // A complete Params buffer was received
          return Ok(Some(Input::KVLines(kvlines)));
        }

        // Returning Ok(None) tells the caller that we need more data
        Ok(None)
      }
      CodecState::Chunks => {
        if buf.is_empty() {
          // Need more data
          return Ok(None);
        }

        let read_to = cmp::min(self.remain, buf.len() as u64);
        self.remain -= read_to;

        if self.remain == 0 {
          // When no more data is expected for this binary part, revert to
          // expecting Telegram lines
          self.state = CodecState::Telegram;
        }

        // Return a buffer and the amount of data remaining, this buffer
        // included.  The application can check if remain is 0 to determine
        // if it has received all the expected binary data.
        //
        // .unwrap() is safe here, because read_to is guaranteed to
        // be within the bounds of an usize due to the `cmp::min()` above.
        let len = usize::try_from(read_to).unwrap();
        Ok(Some(Input::Chunk(buf.split_to(len).freeze(), self.remain)))
      }
      CodecState::Bytes => {
        // This is guaranteed to work, because expect_bytes() takes in an
        // usize.
        let remain: usize = self.remain.try_into().unwrap();
        if buf.len() < remain {
          Ok(None)
        } else {
          // Revert to expecting Telegram lines
          self.state = CodecState::Telegram;

          Ok(Some(Input::Bytes(buf.split_to(remain).freeze())))
        }
      }
      CodecState::BytesCh => {
        // Calculate how much data to take off read buffer, capping at
        // `remain`.
        let read_to = cmp::min(self.remain, buf.len() as u64);
        self.remain -= read_to;

        // Return a buffer and the amount of data remaining, this buffer
        // included.  The application can check if remain is 0 to determine
        // if it has received all the expected binary data.
        //
        // .unwrap() is safe here, because read_to is guaranteed to
        // be within the bounds of an usize due to the `cmp::min()` above.
        let len = usize::try_from(read_to).unwrap();

        let buf = buf.split_to(len).freeze();
        if let Some(ref tx) = self.bytes_tx {
          // Ignore errors -- we'll assume this means the application dropped
          // the reader end-point to signal it is no longer interested in
          // receiving the data.
          let _ = tx.send(buf);
        }

        // If we have all the expected data, then report that we're done with a
        // senitnel value.  Otherwise report back to caller that more data is
        // needed.
        if self.remain == 0 {
          // When no more data is expected for this binary part, revert to
          // expecting a Telegram
          self.state = CodecState::Telegram;

          // Send an empty buffer to signal eof, then drop the end-point
          if let Some(tx) = self.bytes_tx.take() {
            let _ = tx.send(Bytes::new());
          }

          // Return sentinel value just to signal that we're done
          Ok(Some(Input::BytesChDone))
        } else {
          Ok(None)
        }
      }
      CodecState::Skip => {
        if buf.is_empty() {
          return Ok(None); // Need more data
        }

        // Read as much data as available or requested and write it to our
        // output.
        let read_to = cmp::min(self.remain, buf.len() as u64);

        // unwrap() is okay here, due to the cmd::min() above
        let len = usize::try_from(read_to).unwrap();
        let _ = buf.split_to(len);

        self.remain -= read_to;
        if self.remain != 0 {
          return Ok(None); // Need more data
        }

        // Revert to the default of expecting a telegram.
        self.state = CodecState::Telegram;

        Ok(Some(Input::SkipDone))
      } // CodecState::Skip
    } // match self.state
  }
}


impl Encoder<&Telegram> for Codec {
  type Error = crate::err::Error;

  fn encode(
    &mut self,
    tg: &Telegram,
    buf: &mut BytesMut
  ) -> Result<(), Error> {
    tg.encoder_write(buf)?;
    Ok(())
  }
}


impl Encoder<&Params> for Codec {
  type Error = crate::err::Error;

  fn encode(
    &mut self,
    params: &Params,
    buf: &mut BytesMut
  ) -> Result<(), Error> {
    params.encoder_write(buf);
    Ok(())
  }
}


impl Encoder<&HashMap<String, String>> for Codec {
  type Error = crate::err::Error;

  fn encode(
    &mut self,
    data: &HashMap<String, String>,
    buf: &mut BytesMut
  ) -> Result<(), Error> {
    // Calculate the amount of space required
    let mut sz = 0;
    for (k, v) in data {
      // key space + whitespace + value space + eol
      sz += k.len() + 1 + v.len() + 1;
    }

    // Terminating empty line
    sz += 1;

    //println!("Writing {} bin data", data.len());
    buf.reserve(sz);

    for (k, v) in data {
      buf.put(k.as_bytes());
      buf.put_u8(b' ');
      buf.put(v.as_bytes());
      buf.put_u8(b'\n');
    }
    buf.put_u8(b'\n');

    Ok(())
  }
}


impl Encoder<&KVLines> for Codec {
  type Error = crate::err::Error;

  fn encode(
    &mut self,
    kvlines: &KVLines,
    buf: &mut BytesMut
  ) -> Result<(), Error> {
    kvlines.encoder_write(buf);
    Ok(())
  }
}


impl Encoder<Bytes> for Codec {
  type Error = crate::err::Error;

  fn encode(
    &mut self,
    data: Bytes,
    buf: &mut BytesMut
  ) -> Result<(), crate::err::Error> {
    buf.reserve(data.len());
    buf.put(data);
    Ok(())
  }
}


impl Encoder<&[u8]> for Codec {
  type Error = crate::err::Error;

  fn encode(
    &mut self,
    data: &[u8],
    buf: &mut BytesMut
  ) -> Result<(), crate::err::Error> {
    buf.reserve(data.len());
    buf.put(data);
    Ok(())
  }
}


#[cfg(test)]
mod tests {
  use {
    futures::sink::SinkExt, tokio::sync::mpsc::unbounded_channel,
    tokio_stream::StreamExt, tokio_test::io::Builder,
    tokio_util::codec::Framed
  };

  use bytes::BytesMut;

  use super::{Bytes, Codec, Input, Telegram};

  #[tokio::test]
  async fn valid_no_params() {
    let mut mock = Builder::new();

    mock.read(b"hello\n\n");

    let mut frm = Framed::new(mock.build(), Codec::new());

    while let Some(o) = frm.next().await {
      let o = o.unwrap();
      if let Input::Telegram(tg) = o {
        assert_eq!(tg.get_topic(), "hello");
        let params = tg.into_params();
        let map = params.into_inner();
        assert_eq!(map.len(), 0);
      } else {
        panic!("Not a Telegram");
      }
    }
  }

  #[tokio::test]
  async fn valid_with_params() {
    let mut mock = Builder::new();

    mock.read(b"hello\nmurky_waters off\nwrong_impression cows\n\n");

    let mut frm = Framed::new(mock.build(), Codec::new());

    while let Some(o) = frm.next().await {
      let o = o.unwrap();

      match o {
        Input::Telegram(tg) => {
          assert_eq!(tg.get_topic(), "hello");
          let params = tg.into_params();
          let map = params.into_inner();
          assert_eq!(map.len(), 2);
          assert_eq!(map.get("murky_waters").unwrap(), "off");
          assert_eq!(map.get("wrong_impression").unwrap(), "cows");
        }
        _ => {
          panic!("Not a Telegram");
        }
      }
    }
  }

  #[tokio::test]
  #[should_panic(
    expected = "Protocol(\"Bad format; Invalid topic character\")"
  )]
  async fn bad_topic() {
    let mut mock = Builder::new();

    // space isn't allowed in topic
    mock.read(b"hel lo\n\n");

    let mut frm = Framed::new(mock.build(), Codec::new());
    let e = frm.next().await.unwrap();
    e.unwrap();
  }

  #[tokio::test]
  async fn multiple() {
    let mut mock = Builder::new();

    mock.read(b"hello\nfoo bar\n\nworld\nholy cows\n\nfinal\nthe thing\n\n");

    let mut frm = Framed::new(mock.build(), Codec::new());

    let o = frm.next().await.unwrap().unwrap();
    let Input::Telegram(tg) = o else {
      panic!("Unexpectely not Input::Telegram");
    };
    assert_eq!(tg.get_topic(), "hello");
    assert_eq!(tg.get_str("foo"), Some("bar"));

    let o = frm.next().await.unwrap().unwrap();
    let Input::Telegram(tg) = o else {
      panic!("Unexpectely not Input::Telegram");
    };
    assert_eq!(tg.get_topic(), "world");
    assert_eq!(tg.get_str("holy"), Some("cows"));

    let o = frm.next().await.unwrap().unwrap();
    let Input::Telegram(tg) = o else {
      panic!("Unexpectely not Input::Telegram");
    };
    assert_eq!(tg.get_topic(), "final");
    assert_eq!(tg.get_str("the"), Some("thing"));
  }

  #[tokio::test]
  async fn tg_followed_by_buf() {
    let mut mock = Builder::new();

    mock.read(b"hello\nlen 4\n\n1234");

    let mut frm = Framed::new(mock.build(), Codec::new());

    let Some(o) = frm.next().await else {
      panic!("No frame");
    };
    let o = o.unwrap();

    if let Input::Telegram(tg) = o {
      assert_eq!(tg.get_topic(), "hello");
      assert_eq!(tg.get_fromstr::<usize, _>("len").unwrap().unwrap(), 4);
      frm.codec_mut().expect_bytes(4).unwrap();
    } else {
      panic!("Not a Telegram");
    }

    while let Some(o) = frm.next().await {
      let o = o.unwrap();
      if let Input::Bytes(_bm) = o {
      } else {
        panic!("Not a Buf");
      }
    }
  }

  #[tokio::test]
  async fn tg_buf_tg() {
    let mut mock = Builder::new();

    mock.read(b"hello\nlen 4\n\n1234world\nfoo bar\n\n");

    let mut frm = Framed::new(mock.build(), Codec::new());

    // Expect Telegram, which sets up for getting Bytes
    let o = frm.next().await.unwrap().unwrap();
    let Input::Telegram(tg) = o else {
      panic!("Unexpectedly not Input::Telegram(_)");
    };
    assert_eq!(tg.get_topic(), "hello");
    let len = tg.get_fromstr::<usize, _>("len").unwrap().unwrap();
    assert_eq!(len, 4);
    frm.codec_mut().expect_bytes(len).unwrap();

    // Expect Bytes
    let o = frm.next().await.unwrap().unwrap();
    let Input::Bytes(buf) = o else {
      panic!("Unexpectedly not Input::Bytes(_)");
    };
    assert_eq!(buf, "1234");

    // Expect Telegram
    let o = frm.next().await.unwrap().unwrap();
    let Input::Telegram(tg) = o else {
      panic!("Unexpectedly not Input::Telegram(_)");
    };
    assert_eq!(tg.get_topic(), "world");
    assert_eq!(tg.get_str("foo"), Some("bar"));
  }


  #[tokio::test]
  async fn expect_bytes_ch() {
    let (client, server) = tokio::io::duplex(64);

    let mut frmin = Framed::new(server, Codec::new());
    let mut frmout = Framed::new(client, Codec::new());

    // Spawn server task
    let jh = tokio::task::spawn(async move {
      let o = frmin.next().await.unwrap().unwrap();
      let Input::Telegram(tg) = o else {
        panic!("Unexpectedly not Input::Telegram(_)");
      };
      assert_eq!(tg.as_ref(), "ReqToSend");
      let len = tg.get_fromstr::<u64, _>("Len").unwrap().unwrap();

      // Create a channel for receiving stream of Bytes
      let (tx, mut rx) = unbounded_channel();

      // Expect requested number of bytes
      frmin.codec_mut().expect_bytes_channel(len, tx).unwrap();

      // Spawn task for receiving expected data through channel
      let jh = tokio::task::spawn(async move {
        let mut inbuf = BytesMut::new();
        inbuf.reserve(16);

        // Receive the expected amount of bytes over channel
        loop {
          let buf = rx.recv().await.unwrap();
          if buf.is_empty() {
            break;
          }
          // ToDo: Verify buffer contents
          inbuf.extend_from_slice(&buf);
        }

        let buf = inbuf.freeze();
        assert_eq!(buf, "0123456789abcdef");
      });

      // Wait for codec to report that it is done feeding data to the channel
      let o = frmin.next().await.unwrap().unwrap();
      let Input::BytesChDone = o else {
        panic!("Unexpectedly not Input::BytesChDone");
      };

      jh.await.unwrap();
    });

    let len = 16;

    let mut tg = Telegram::new("ReqToSend");
    tg.add_param("Len", len).unwrap();
    frmout.send(&tg).await.unwrap();

    // Send `len` amount of binary data
    frmout.send(Bytes::from("0123")).await.unwrap();
    frmout.send(Bytes::from("4567")).await.unwrap();
    frmout.send(Bytes::from("89ab")).await.unwrap();
    frmout.send(Bytes::from("cdef")).await.unwrap();

    jh.await.unwrap();
  }
}

// vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 :