Cargo.toml
src/lib.rs


src/err.rs

src/types.rs
src/types/telegram.rs
src/types/params.rs
src/types/kvlines.rs
src/types/validators.rs
src/codec.rs
tests/*.rs

[package]
name = "blather"
version = "0.9.0"
authors = ["Jan Danielsson <jan.danielsson@qrnch.com>"]
edition = "2021"
license = "0BSD"
keywords = [ "line-based", "protocol", "tokio", "codec" ]
repository = "https://repos.qrnch.tech/pub/blather"
description = "A talkative line-based protocol"
exclude = [
  ".fossil-settings",
  ".efiles",
  ".fslckout",
  "rustfmt.toml",
  "tests",
  "www"
]

[dependencies]
bytes = { version = "1.1.0" }
futures = { version = "0.3.21" }

tokio = { version = "1.17.0" }
tokio-util = { version= "0.7.0", features = ["codec"] }

[dev-dependencies]
tokio = { version = "1.17.0", features = ["macros", "net"] }
tokio-stream = { version = "0.1.8" }
tokio-test = { version = "0.4.2" }




# vim: set ft=toml et sw=2 ts=2 sts=2 cinoptions=2 tw=79 :

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

use std::fmt;
use std::fs::File;
use std::io::Write;
use std::path::PathBuf;
use std::{cmp, collections::HashMap, mem};


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

use tokio::io;

use tokio_util::codec::Decoder;
use tokio_util::codec::Encoder;

use crate::err::Error;

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



/// Current state of decoder.
///
/// Controls what, if anything, will be returned to the application.
#[derive(Clone, Debug, PartialEq)]
enum CodecState {

  /// arrive.
  Chunks,

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

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

  /// Read a specified amount of raw bytes and store them in chunks as they
  /// arrive in a file.
  File,

  /// Read a specified amount of raw bytes and write them in chunks as they
  /// arrive to a writer object.
  Writer,

  /// 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 [`BytesMut`].  If
  /// the `usize` parameter is 0 it means this is the final chunk.
  Chunk(BytesMut, usize),

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

  /// A complete raw mutable buffer has been received.
  BytesMut(BytesMut),

  /// A complete buffer has been received and stored to the file specified in
  /// `PathBuf`.
  File(PathBuf),

  /// A complete buffer has been written to the writer.
  WriteDone,

  /// 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,
  bin_remain: usize,
  pathname: Option<PathBuf>,
  writer: Option<Box<dyn Write + Send + Sync>>,
  buf: BytesMut
}

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

/// 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_bytesmut(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.
  pub fn new() -> Codec {
    Codec {
      next_line_index: 0,
      max_line_length: usize::MAX,
      tg: Telegram::new(),
      params: Params::new(),
      kvlines: KVLines::new(),
      state: CodecState::Telegram,
      bin_remain: 0,
      pathname: None,
      writer: None,
      buf: BytesMut::new()
    }
  }

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

  /// 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_none() {


      self.tg.set_topic(line)?;

    } 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)?;
      }

      }
    }
  }
  */

  /// 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::BadFormat(
        "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.

        // 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).
          return Ok(Some(mem::take(&mut self.tg)));
        } else {
          self.decode_telegram_line(&line)?;
        }
      } else {
        // Returning Ok(None) instructs the FramedRead that more data is
        // needed.
        return Ok(None);
      }
    }

  /// 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`].
  pub fn expect_chunks(&mut self, size: usize) {




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


  }


  /// 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`].
  pub fn expect_bytes(&mut self, size: usize) -> Result<(), Error> {
    if size == 0 {
      return Err(Error::InvalidSize("The size must not be zero".to_string()));
    }
    self.state = CodecState::Bytes;
    self.bin_remain = size;
    self.buf = BytesMut::with_capacity(size);
    Ok(())
  }


  /// Expect a mutable 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::BytesMut(b)`](Input::BytesMut) where `b` is a
  /// [`bytes::BytesMut`] containing the entire buffer.
  ///
  /// Once the entire buffer has been received by the `Decoder` it will revert
  /// to expect an [`Input::Telegram`].
  pub fn expect_bytesmut(&mut self, size: usize) -> Result<(), Error> {
    if size == 0 {
      return Err(Error::InvalidSize("The size must not be zero".to_string()));
    }
    self.state = CodecState::BytesMut;
    self.bin_remain = size;
    self.buf = BytesMut::with_capacity(size);
    Ok(())
  }


  /// Expects a certain amount of bytes of data to arrive from the peer, and
  /// that data should be stored to a file.
  ///
  /// # Decoder behavior
  /// On successful completion the Decoder will return an
  /// [`Input::File(pathname)`](Input::File) once the entire file length has
  /// successfully been received, where the pathname is a PathBuf which
  /// matches the pathname parameter passed to this function.
  ///
  /// Once the entire buffer has been received by the `Decoder` it will revert
  /// to expect an [`Input::Telegram`].
  pub fn expect_file<P: Into<PathBuf>>(
    &mut self,
    pathname: P,
    size: usize
  ) -> Result<(), Error> {
    if size == 0 {
      return Err(Error::InvalidSize("The size must not be zero".to_string()));
    }
    self.state = CodecState::File;
    let pathname = pathname.into();
    self.writer = Some(Box::new(File::create(&pathname)?));
    self.pathname = Some(pathname);

    self.bin_remain = size;

    Ok(())
  }

  /// Called from an application to request that data should be written to a
  /// supplied writer.
  ///
  /// The writer's ownership will be transferred to the `Decoder` and will
  /// automatically be dropped once the entire buffer has been written.
  ///
  /// # Decoder behavior
  /// On successful completion the Decoder will return an Input::WriteDone to
  /// signal that the entire buffer has been received and written to the
  /// `Writer`.
  ///
  /// Once the entire buffer has been received by the `Decoder` it will revert
  /// to expect an [`Input::Telegram`].
  pub fn expect_writer<W: 'static + Write + Send + Sync>(
    &mut self,
    writer: W,
    size: usize
  ) -> Result<(), Error> {
    if size == 0 {
      return Err(Error::InvalidSize("The size must not be zero".to_string()));
    }
    self.state = CodecState::Writer;
    self.writer = Some(Box::new(writer));
    self.bin_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.

  /// 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`].
  ///
  /// Once the entire buffer has been skipped by the `Decoder` it will revert
  /// to expect an [`Input::Telegram`].
  pub fn skip(&mut self, size: usize) -> Result<(), Error> {
    if size == 0 {
      return Err(Error::InvalidSize("The size must not be zero".to_string()));
    }
    self.state = CodecState::Skip;
    self.bin_remain = size;
    Ok(())
  }
}

fn utf8(buf: &[u8]) -> Result<&str, io::Error> {
  std::str::from_utf8(buf).map_err(|_| {
    io::Error::new(

      }
      CodecState::Chunks => {
        if buf.is_empty() {
          // Need more data
          return Ok(None);
        }

        let read_to = cmp::min(self.bin_remain, buf.len());
        self.bin_remain -= read_to;

        if self.bin_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.



        Ok(Some(Input::Chunk(buf.split_to(read_to), self.bin_remain)))



      }
      CodecState::Bytes => {
        if buf.is_empty() {
          // Need more data
          return Ok(None);
        }
        let read_to = cmp::min(self.bin_remain, buf.len());

        // Transfer data from input to output buffer
        self.buf.put(buf.split_to(read_to));

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

        // 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.
        let bytesmut = mem::take(&mut self.buf);

        Ok(Some(Input::Bytes(Bytes::from(bytesmut))))
      }
      CodecState::BytesMut => {
        if buf.is_empty() {
          // Need more data
          return Ok(None);
        }
        let read_to = cmp::min(self.bin_remain, buf.len());

        // Transfer data from input to output buffer
        self.buf.put(buf.split_to(read_to));

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

        // 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.
        Ok(Some(Input::BytesMut(mem::take(&mut self.buf))))
      }
      CodecState::File | CodecState::Writer => {
        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.bin_remain, buf.len());
        if let Some(ref mut f) = self.writer {
          f.write_all(&buf.split_to(read_to))?;
        }

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

        // At this point the entire expected buffer has been received

        // Close file
        self.writer = None;

        // 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.
        let ret = if self.state == CodecState::File {
          let pathname = if let Some(ref fname) = self.pathname {
            fname.clone()
          } else {
            return Err(Error::BadState("Missing pathname".to_string()));
          };

          // Reset the pathname
          self.pathname = None;

          Input::File(pathname)
        } else {
          Input::WriteDone
        };

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

        Ok(Some(ret))
      } // CodecState::{File|Writer}
      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.bin_remain, buf.len());
        let _ = buf.split_to(read_to);

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

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

        Ok(Some(Input::SkipDone))

//! Error types and error management functions.

use std::fmt;

use tokio::io;

/// Error that `blather` can emit.
#[derive(Debug, PartialEq)]
pub enum Error {
  /// The requiested key was not found.
  KeyNotFound(String),

  /// The input format of a buffer was incorrect.
  BadFormat(String),

  /// Unable to serialize a buffer.
  SerializeError(String),

  /// A `std::io` or `tokio::io` error has occurred.
  IO(String),

  /// Something occurred which was unexpected in the current state.
  BadState(String),

  /// The specified size is invalid, or invalid in a specific context.
  InvalidSize(String)













}

impl std::error::Error for Error {}

impl fmt::Display for Error {
  fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
    match &*self {
      Error::KeyNotFound(s) => write!(f, "Parameter '{}' not found", s),
      Error::BadFormat(s) => write!(f, "Bad format; {}", s),
      Error::SerializeError(s) => write!(f, "Unable to serialize; {}", s),
      Error::IO(s) => write!(f, "I/O error; {}", s),
      Error::BadState(s) => {
        write!(f, "Encountred an unexpected/bad state: {}", s)
      }
      Error::InvalidSize(s) => write!(f, "Invalid size; {}", s)




    }
  }
}

impl From<io::Error> for Error {
  fn from(err: io::Error) -> Self {
    Error::IO(err.to_string())

//! A set of "parameters", represented by the Params struct, is a set of
//! key/value pairs.  They look similar to `Telegrams` because the `Telegram`'s
//! implement their key/value paris using a `Params` buffer.
//!
//! # Communication
//! blather handles transmission using tokio-util's
//! [`Framed`](tokio_util::codec::Framed) framework, by
//! implementing its own [`Codec`](codec::Codec).  It can be used to send and
//! receive the various communication buffers supported by the crate.

#![deny(missing_docs)]
#![deny(rustdoc::missing_crate_level_docs)]
#![deny(rustdoc::missing_doc_code_examples)]

pub mod codec;
mod err;
pub mod types;

pub use codec::Codec;
pub use err::Error;

//! A key/value pair list with stable ordering and non-unique keys.

use std::convert::From;
use std::fmt;

use bytes::{BufMut, BytesMut};

use crate::err::Error;

/// Representation of a key/value pair in `KVLines`.

//! The `Params` buffer is a set of unorderded key/value pairs, with unique
//! keys.  It's similar to a `HashMap`, but has constraints on key names and
//! offers conventions for value layouts, such as comma-separated values for
//! lists.


use std::collections::{HashMap, HashSet};
use std::convert::From;
use std::fmt;
use std::str::FromStr;


use bytes::{BufMut, BytesMut};

use super::validators::validate_param_key;

use crate::err::Error;

  /// Return the number of key/value pairs in the parameter buffer.
  pub fn len(&self) -> usize {
    self.hm.len()
  }









  /// Return reference to inner HashMap.
  pub fn get_inner(&self) -> &HashMap<String, String> {
    &self.hm
  }


  /// Add a parameter to the parameter.
  ///
  /// The `key` and `value` parameters are generic over the trait `ToString`,
  /// allowing a polymorphic behavior.
  ///
  /// # Examples
  /// ```
  /// use blather::Params;
  /// fn main() {
  ///   let mut params = Params::new();
  ///   params.add_param("integer", 42).unwrap();
  ///   params.add_param("string", "hello").unwrap();
  /// }
  /// ```
  pub fn add_param<T: ToString, U: ToString>(
    &mut self,
    key: T,
    value: U
  ) -> Result<(), Error> {
    let key = key.to_string();

  /// Add parameter where the value is generated from an iterator over
  /// strings, where entries are comma-separated.
  ///
  /// # Examples
  /// ```
  /// use std::collections::HashSet;
  /// use blather::Params;
  /// fn main() {
  ///   let mut params = Params::new();
  ///
  ///   params.add_strit("Cat", &["meow", "paws", "tail"]).unwrap();
  ///   assert_eq!(params.get_str("Cat"), Some("meow,paws,tail"));
  ///
  ///   let v = vec!["meow", "paws", "tail"];
  ///   params.add_strit("CatToo", v.into_iter()).unwrap();
  ///   assert_eq!(params.get_str("CatToo"), Some("meow,paws,tail"));
  ///
  ///   let mut hs = HashSet::new();
  ///   hs.insert("Elena");
  ///   hs.insert("Drake");
  ///   params.add_strit("Uncharted", hs.into_iter()).unwrap();
  /// }
  /// ```
  pub fn add_strit<I, S>(&mut self, key: &str, c: I) -> Result<(), Error>
  where
    I: IntoIterator<Item = S>,
    S: AsRef<str>
  {
    let mut sv = Vec::new();

  /// Add a boolean parameter.
  ///
  /// # Examples
  /// ```
  /// use blather::Params;
  /// fn main() {
  ///   let mut params = Params::new();
  ///   params.add_bool("should_be_true", true).unwrap();
  ///   params.add_bool("should_be_false", false).unwrap();
  ///   assert_eq!(params.get_bool("should_be_true"), Ok(true));
  ///   assert_eq!(params.get_bool("should_be_false"), Ok(false));
  /// }
  /// ```
  ///
  /// # Notes
  /// - Applications should not make assumptions about the specific string
  ///   value added by this function.  Do not treat boolean values as strings;
  ///   use the [`get_bool()`](Self::get_bool) method instead.
  pub fn add_bool<K: ToString>(

  /// Get a parameter and convert it to a requested type, fail if key isn't
  /// found.
  ///
  /// # Examples
  /// ```
  /// use blather::{Params, Error};
  /// fn main() {
  ///   let mut params = Params::new();
  ///   params.add_param("arthur", 42);
  ///   let fourtytwo = params.get_param::<u32>("arthur").unwrap();
  ///   assert_eq!(fourtytwo, 42);
  ///   let nonexist = params.get_param::<u32>("ford");
  ///   assert_eq!(nonexist, Err(Error::KeyNotFound("ford".to_string())));
  /// }
  /// ```
  pub fn get_param<T: FromStr>(&self, key: &str) -> Result<T, Error> {
    if let Some(val) = self.get_str(key) {
      if let Ok(v) = T::from_str(val) {
        return Ok(v);
      }
      return Err(Error::BadFormat(format!(

  /// Get a parameter and convert it to a requested type, return a default
  /// value if key isn't found.
  ///
  /// # Examples
  /// ```
  /// use blather::Params;
  /// fn main() {
  ///   let mut params = Params::new();
  ///   let val = params.get_param_def::<u32>("nonexist", 11);
  ///   assert_eq!(val, Ok(11));
  /// }
  /// ```
  pub fn get_param_def<T: FromStr>(
    &self,
    key: &str,
    def: T
  ) -> Result<T, Error> {
    if let Some(val) = self.get_str(key) {

  /// Get string representation of a value for a requested key.  Returns a
  /// default value if key does not exist in parameter buffer.
  ///
  /// # Examples
  /// ```
  /// use blather::Params;
  /// fn main() {
  ///   let params = Params::new();
  ///   let e = params.get_str_def("nonexist", "elena");
  ///   assert_eq!(e, "elena");
  /// }
  /// ```
  // Lifetimes of self and def don't really go hand-in-hand, but we bound them
  // together for the sake of the return value's lifetime.
  pub fn get_str_def<'a>(&'a self, key: &str, def: &'a str) -> &'a str {
    let kv = self.hm.get_key_value(key);
    if let Some((_k, v)) = kv {
      v
    } else {
      def
    }
  }


  /// Get a parameter and convert it to an integer type.
  ///
  /// # Examples
  /// ```
  /// use blather::Params;
  /// fn main() {
  ///   let mut params = Params::new();
  ///   params.add_param("Num", 7);
  ///   assert_eq!(params.get_int::<usize>("Num").unwrap(), 7);
  /// }
  /// ```
  ///
  /// # Notes
  /// - This method exists primarily to achive some sort of parity with a
  ///   corresponding C++ library.  It is recommended that applications use
  ///   [`Params::get_param()`](Self::get_param) instead.
  // This method should really have some integer trait bound, but it doesn't

  /// Try to get the value of a key and interpret it as an integer.  If the key
  /// does not exist then return a default value supplied by the caller.
  ///
  /// # Examples
  /// ```
  /// use blather::Params;
  /// fn main() {
  ///   let mut params = Params::new();
  ///   params.add_param("num", 11);
  ///   assert_eq!(params.get_int_def::<u32>("num", 5).unwrap(), 11);
  ///   assert_eq!(params.get_int_def::<u32>("nonexistent", 17).unwrap(), 17);
  /// }
  /// ```
  ///
  /// # Notes
  /// - It is recommended that application use
  ///   [`Params::get_param_def()`](Self::get_param_def) instead.
  pub fn get_int_def<T: FromStr>(
    &self,

  /// Parse the value of a key as a comma-separated list of strings and return
  /// it.  Only non-empty entries are returned.
  ///
  /// # Examples
  /// ```
  /// use blather::Params;
  /// fn main() {
  ///   let mut params = Params::new();
  ///   params.add_param("csv", "elena,chloe,drake");
  ///   let sv = params.get_strvec("csv").unwrap();
  ///   assert_eq!(sv, vec!["elena", "chloe", "drake"]);
  /// }
  /// ```
  pub fn get_strvec(&self, key: &str) -> Result<Vec<String>, Error> {
    let mut ret = Vec::new();

    if let Some(v) = self.get_str(key) {
      let split = v.split(',');
      for s in split {
        if s.len() != 0 {
          ret.push(s.to_string());
        }
      }
    }

    Ok(ret)
  }


  /// Parse the value of a key as a comma-separated list of uniqie strings and
  /// return them in a HashSet.  Only non-empty entries are returned.
  ///
  /// # Examples
  /// ```
  /// use blather::Params;
  /// fn main() {
  ///   let mut params = Params::new();
  ///   params.add_param("set", "elena,chloe");
  ///   let set = params.get_hashset("set").unwrap();
  ///   assert_eq!(set.len(), 2);
  ///   assert_eq!(set.contains("elena"), true);
  ///   assert_eq!(set.contains("chloe"), true);
  ///   assert_eq!(set.contains("drake"), false);
  /// }
  /// ```
  pub fn get_hashset(&self, key: &str) -> Result<HashSet<String>, Error> {
    let mut ret = HashSet::new();

    if let Some(v) = self.get_str(key) {
      let split = v.split(',');
      for s in split {
        if s.len() != 0 {
          ret.insert(s.to_string());
        }
      }
    }

    Ok(ret)
  }

//! Telegrams are objects that contain a _topic_ and a set of zero or more
//! parameters.  They can be serialized into a line-based format for
//! transmission over a network link.


use std::collections::{HashMap, HashSet};
use std::fmt;
use std::str::FromStr;


use bytes::{BufMut, BytesMut};

use crate::err::Error;

use super::params::Params;
use super::validators::validate_topic;

/// Representation of a Telegram; a buffer which contains a _topic_ and a set
/// of key/value parameters.
///
/// Internally the key/value parameters are represented by a [`Params`]
/// structure.
#[derive(Debug, Clone, Default)]

  }


  /// Create a new telegram object with a topic.
  ///
  /// ```
  /// use blather::Telegram;
  /// fn main() {
  ///   let mut tg = Telegram::new_topic("Hello").unwrap();
  ///   assert_eq!(tg.get_topic(), Some("Hello"));
  /// }
  /// ```
  pub fn new_topic(topic: &str) -> Result<Self, Error> {
    validate_topic(topic)?;
    Ok(Telegram {
      topic: Some(topic.to_string()),
      ..Default::default()
    })
  }


  /// Clear topic and internal parameters buffer.
  ///
  /// ```
  /// use blather::Telegram;
  /// fn main() {
  ///   let mut tg = Telegram::new();
  ///   tg.add_param("cat", "meow");
  ///   assert_eq!(tg.num_params(), 1);
  ///   tg.clear();
  ///   assert_eq!(tg.num_params(), 0);
  /// }
  /// ```
  pub fn clear(&mut self) {
    self.topic = None;
    self.params.clear();
  }


  /// Return the number of key/value parameters in the Telegram object.
  ///
  /// # Examples
  /// ```
  /// use blather::Telegram;
  /// fn main() {
  ///   let mut tg = Telegram::new();
  ///   assert_eq!(tg.num_params(), 0);
  ///   tg.add_param("cat", "meow");
  ///   assert_eq!(tg.num_params(), 1);
  /// }
  /// ```
  ///
  /// # Notes
  /// This is a wrapper around [`Params::len()`](crate::Params::len).
  pub fn num_params(&self) -> usize {
    self.params.len()
  }


  /// Get a reference to the internal parameters object.
  pub fn get_params(&self) -> &Params {
    &self.params
  }


  /// Get a mutable reference to the inner [`Params`](crate::Params) object.
  ///
  /// ```
  /// use blather::Telegram;
  /// fn main() {
  ///   let mut tg = Telegram::new();
  ///   tg.add_param("cat", "meow");
  ///   assert_eq!(tg.num_params(), 1);
  ///   tg.get_params_mut().clear();
  ///   assert_eq!(tg.num_params(), 0);
  /// }
  /// ```
  pub fn get_params_mut(&mut self) -> &mut Params {
    &mut self.params
  }


  /// Get a reference the the parameter's internal HashMap.
  ///
  /// Note: The inner representation of the Params object may change in the
  /// future.
  pub fn get_params_inner(&self) -> &HashMap<String, String> {
    &self.params.get_inner()
  }


  /// Set topic for telegram.
  ///
  /// Overwrites current topic is one has already been set.
  ///
  /// # Examples
  /// ```
  /// use blather::{Telegram, Error};
  /// fn main() {
  ///   let mut tg = Telegram::new();
  ///   assert_eq!(tg.set_topic("Hello"), Ok(()));
  ///
  ///   let e = Error::BadFormat("Invalid topic character".to_string());
  ///   assert_eq!(tg.set_topic("Hell o"), Err(e));
  /// }
  /// ```
  pub fn set_topic(&mut self, topic: &str) -> Result<(), Error> {
    validate_topic(topic)?;
    self.topic = Some(topic.to_string());
    Ok(())
  }


  /// Get a reference to the topic string, or None if topic is not been set.
  ///
  /// # Examples
  /// ```
  /// use blather::{Telegram, Error};
  /// fn main() {
  ///   let tg = Telegram::new_topic("shoe0nhead").unwrap();
  ///   assert_eq!(tg.get_topic(), Some("shoe0nhead"));
  ///
  ///   let tg = Telegram::new();
  ///   assert_eq!(tg.get_topic(), None);
  /// }
  /// ```
  pub fn get_topic(&self) -> Option<&str> {
    if let Some(t) = &self.topic {
      Some(t)
    } else {
      None
    }
  }


  /// Add a parameter to the telegram.
  ///
  /// The `key` and `value` parameters are generic over the trait `ToString`,
  /// allowing a polymorphic behavior.
  ///
  /// # Examples
  /// ```
  /// use blather::Telegram;
  /// fn main() {
  ///   let mut tg = Telegram::new();
  ///   tg.add_param("integer", 42).unwrap();
  ///   tg.add_param("string", "hello").unwrap();
  /// }
  /// ```
  ///
  /// # Notes
  /// - This is a thin wrapper around
  ///   [`Params::add_param()`](crate::Params::add_param).
  pub fn add_param<T: ToString, U: ToString>(
    &mut self,

  /// Add parameter where the value is generated from an iterator over a
  /// string container, where entries will be comma-separated.
  ///
  /// ```
  /// use blather::Telegram;
  /// fn main() {
  ///   let mut tg = Telegram::new();
  ///   tg.add_strit("Cat", &["meow", "paws", "tail"]).unwrap();
  ///   assert_eq!(tg.get_str("Cat"), Some("meow,paws,tail"));
  /// }
  /// ```
  ///
  /// # Notes
  /// - This is a thin wrapper for
  ///   [`Params::add_strit()`](crate::Params::add_strit).
  pub fn add_strit<I, S>(&mut self, key: &str, c: I) -> Result<(), Error>
  where

  }


  /// Get an integer representation of a parameter.
  ///
  /// ```
  /// use blather::Telegram;
  /// fn main() {
  ///   let mut tg = Telegram::new();
  ///   tg.add_param("Num", 7);
  ///   assert_eq!(tg.get_int::<usize>("Num").unwrap(), 7);
  /// }
  /// ```
  ///
  /// # Notes
  /// - This function uses the `FromStr` trait on the return-type so it
  ///   technically isn't limited to integers.
  /// - The method exists to mimic a C++ library.  It is recommeded that
  ///   applications use [`Telegram::get_param()`](Self::get_param) instead.
  pub fn get_int<T: FromStr>(&self, key: &str) -> Result<T, Error> {
    self.params.get_int(key)
  }


  /// Try to get the parameter value of a key and interpret it as an integer.
  /// If the key does not exist then return a default value supplied by the
  /// caller.
  ///
  /// ```
  /// use blather::Telegram;
  /// fn main() {
  ///   let mut tg = Telegram::new();
  ///   tg.add_param("num", 11);
  ///   assert_eq!(tg.get_int_def::<u32>("num", 5).unwrap(), 11);
  ///   assert_eq!(tg.get_int_def::<u32>("nonexistent", 17).unwrap(), 17);
  /// }
  /// ```
  pub fn get_int_def<T: FromStr>(
    &self,
    key: &str,
    def: T
  ) -> Result<T, Error> {
    self.params.get_int_def(key, def)

  }
}

impl fmt::Display for Telegram {
  fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
    let topic: &str = match &self.topic {
      Some(s) => s.as_ref(),
      None => &"<None>"
    };

    write!(f, "{}:{}", topic, self.params)
  }
}

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

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

  let mut frm = Framed::new(mock.build(), Codec::new());
  if let Some(e) = frm.next().await {
    if let Err(e) = e {
      match e {
        Error::BadFormat(s) => {
          assert_eq!(s, "Invalid topic character");
        }
        _ => {
          panic!("Wrong error");
        }
      }
    } else {
      panic!("Unexpected success");

#[test]
fn exists() {
  let mut params = Params::new();

  params.add_str("foo", "bar").unwrap();
  assert_eq!(params.have("foo"), true);

  assert_eq!(params.have("nonexistent"), false);
}


#[test]
fn integer() {
  let mut msg = Params::new();

  msg.add_str("Num", "64").unwrap();
  assert_eq!(msg.get_int::<u16>("Num").unwrap(), 64);
}


#[test]
fn size() {
  let mut msg = Params::new();

  msg.add_param("Num", 7 as usize).unwrap();
  assert_eq!(msg.get_int::<usize>("Num").unwrap(), 7);
}


#[test]
fn intoparams() {
  let mut msg = Params::new();

}


#[test]
fn strvec_single_add() {
  let mut params = Params::new();

  let mut sv = Vec::new();
  sv.push("foo");
  params.add_strit("hello", &sv).unwrap();


  //let v = params.get_str("hello").unwrap();
  assert_eq!(params.get_str("hello"), Some("foo"));


  let sv = params.get_strvec("hello").unwrap();
  assert_eq!(sv.len(), 1);
  assert_eq!(sv[0], "foo");
}


#[test]
fn strvec_two_add() {
  let mut params = Params::new();

  let mut sv = Vec::new();
  sv.push("foo");
  sv.push("bar");
  params.add_strit("hello", &sv).unwrap();

  assert_eq!(params.get_str("hello"), Some("foo,bar"));

  let sv = params.get_strvec("hello").unwrap();
  assert_eq!(sv.len(), 2);
  assert_eq!(sv[0], "foo");

#[test]
fn exist() {
  let mut tg = Telegram::new();

  tg.add_str("foo", "bar").unwrap();
  assert_eq!(tg.have_param("foo"), true);

  assert_eq!(tg.have_param("nonexistent"), false);
}


#[test]
fn integer() {
  let mut msg = Telegram::new();

  msg.set_topic("SomeTopic").unwrap();
  assert_eq!(msg.get_topic().unwrap(), "SomeTopic");

  msg.add_str("Num", "64").unwrap();
  assert_eq!(msg.get_int::<u16>("Num").unwrap(), 64);
}


#[test]
fn size() {
  let mut msg = Telegram::new();

  msg.add_param("Num", 7 as usize).unwrap();
  assert_eq!(msg.get_int::<usize>("Num").unwrap(), 7);
}


#[test]
fn intoparams() {
  let mut msg = Telegram::new();

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());

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


    let o = o.unwrap();

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

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

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