[package]
name = "protwrap"
version = "0.3.1"
edition = "2021"
license = "0BSD"
# https://crates.io/category_slugs
categories = [ "asynchronous", "network-programming" ]
keywords = [ "network", "wrapper" ]
repository = "https://repos.qrnch.tech/pub/protwrap"
description = "Thin protocol wrapper for network applications."
exclude = [
  ".fossil-settings",
  ".efiles",
  ".fslckout",
  "examples",
  "www",
  "bacon.toml",
  "rustfmt.toml"
]

# https://doc.rust-lang.org/cargo/reference/manifest.html#the-badges-section
[badges]
maintenance = { status = "experimental" }

[features]
tls = ["dep:tokio-rustls"]
tokio = ["dep:tokio", "dep:tokio-util", "dep:async-trait", "dep:killswitch"]

[dependencies]
async-trait = { version = "0.1.82", optional = true }
killswitch = { version = "0.4.2", optional = true }
tokio = { version = "1.37.0", optional = true, features = [
  "macros", "net", "rt"
] }
tokio-rustls = { version = "0.24.0", optional = true, features = [
  "dangerous_configuration"
] }
tokio-util = { version = "0.7.12", optional = true }

[target.'cfg(unix)'.dependencies]
tokio = { version = "1.40.0", optional = true, features = ["fs"] }

[dev-dependencies]
tokio = { version = "1.38.0", features = [
  "io-util", "rt-multi-thread", "time"
] }

[package.metadata.docs.rs]
all-features = true
rustdoc-args = ["--cfg", "docsrs", "--generate-link-to-definition"]

[lints.clippy]
all = { level = "deny", priority = -1 }
pedantic = { level = "warn", priority = -1 }
nursery = { level = "warn", priority = -1 }
cargo = { level = "warn", priority = -1 }

# This is a configuration file for the bacon tool
#
# Bacon repository: https://github.com/Canop/bacon
# Complete help on configuration: https://dystroy.org/bacon/config/
# You can also check bacon's own bacon.toml file
#  as an example: https://github.com/Canop/bacon/blob/main/bacon.toml

# For information about clippy lints, see:
# https://github.com/rust-lang/rust-clippy/blob/master/README.md

#default_job = "check"
default_job = "clippy-all"

[jobs.check]
command = ["cargo", "check", "--color", "always"]
need_stdout = false

[jobs.check-all]
command = ["cargo", "check", "--all-targets", "--color", "always"]
need_stdout = false

# Run clippy on the default target
[jobs.clippy]
command = [
    "cargo", "clippy",
    "--all-features",
    "--color", "always",
]
need_stdout = false

# Run clippy on all targets
# To disable some lints, you may change the job this way:
#    [jobs.clippy-all]
#    command = [
#        "cargo", "clippy",
#        "--all-targets",
#        "--color", "always",
#    	 "--",
#    	 "-A", "clippy::bool_to_int_with_if",
#    	 "-A", "clippy::collapsible_if",
#    	 "-A", "clippy::derive_partial_eq_without_eq",
#    ]
# need_stdout = false
[jobs.clippy-all]
command = [
    "cargo", "clippy",
    "--all-features",
    "--all-targets",
    "--color", "always",
]
need_stdout = false

# This job lets you run
# - all tests: bacon test
# - a specific test: bacon test -- config::test_default_files
# - the tests of a package: bacon test -- -- -p config
[jobs.test]
command = [
    "cargo", "test", "--color", "always",
    "--", "--color", "always", # see https://github.com/Canop/bacon/issues/124
]
need_stdout = true

[jobs.doc]
command = ["cargo", "doc", "--color", "always", "--no-deps"]
need_stdout = false

# If the doc compiles, then it opens in your browser and bacon switches
# to the previous job
[jobs.doc-open]
command = ["cargo", "doc", "--color", "always", "--no-deps", "--open"]
need_stdout = false
on_success = "back" # so that we don't open the browser at each change

# You can run your application and have the result displayed in bacon,
# *if* it makes sense for this crate.
# Don't forget the `--color always` part or the errors won't be
# properly parsed.
# If your program never stops (eg a server), you may set `background`
# to false to have the cargo run output immediately displayed instead
# of waiting for program's end.
[jobs.run]
command = [
    "cargo", "run",
    "--color", "always",
    # put launch parameters for your program behind a `--` separator
]
need_stdout = true
allow_warnings = true
background = true

# This parameterized job runs the example of your choice, as soon
# as the code compiles.
# Call it as
#    bacon ex -- my-example
[jobs.ex]
command = ["cargo", "run", "--color", "always", "--example"]
need_stdout = true
allow_warnings = true

# You may define here keybindings that would be specific to
# a project, for example a shortcut to launch a specific job.
# Shortcuts to internal functions (scrolling, toggling, etc.)
# should go in your personal global prefs.toml file instead.
[keybindings]
# alt-m = "job:my-job"
c = "job:clippy-all" # comment this to have 'c' run clippy on only the default target

#[cfg(feature = "tokio")]
mod tok {

  pub use protwrap::tokio::{
    client::connector,
    server::listener::{
      async_trait, Acceptor, KillSwitch, Listener, SockAddr
    },
    ServerStream
  };

  pub use tokio::{
    io::{AsyncReadExt, AsyncWriteExt},
    sync::oneshot
  };

  pub struct MyAcceptor {
    pub(super) tx_port: Option<oneshot::Sender<u16>>,
    pub(super) ks: KillSwitch
  }

  #[async_trait]
  impl Acceptor for MyAcceptor {
    async fn bound(&mut self, _listener: &Listener, sa: SockAddr) {
      //
      // The listener has been successfully bound to a socket address
      //
      // Retreive the system-allocated port number and send it to the client
      // ask using the one-shot channel.
      //
      let sa = sa.unwrap_std();
      println!("Bound to {sa:?}");
      let port = sa.port();
      let Some(tx) = self.tx_port.take() else {
        panic!("Channel end-point missing");
      };
      tx.send(port).unwrap();
    }

    async fn unbound(&mut self, _listener: &Listener) {
      println!("Unbound");
    }

    async fn connected(&mut self, sa: SockAddr, mut strm: ServerStream) {
      let sa = sa.unwrap_std();

      println!("server listener: Received an incoming connection from {sa:?}",);



      let killswitch = self.ks.clone();
      tokio::task::spawn(async move {
        let mut buf = [0u8; 5];

        println!("client: Waiting for 'hello' from client");
        let n = strm.read(&mut buf[..]).await.unwrap();
        assert_eq!(n, 5);

        println!("client: Sending 'world' to client");
        let n = strm.write(b"world").await.unwrap();
        assert_eq!(n, 5);

        println!("client: Triggering killswitch to terminate listener");
        killswitch.trigger();
      });
    }
  }
}

#[cfg(feature = "tokio")]
use {
  std::str::FromStr,
  tok::{
    connector, oneshot, AsyncReadExt, AsyncWriteExt, KillSwitch, Listener,
    MyAcceptor
  }
};

#[cfg(feature = "tokio")]
#[tokio::main]
async fn main() {
  // channel used to pass port number from the server task to the client task.
  let (tx, rx) = oneshot::channel();

  //
  // Set up and spawn client task
  //
  let jh_client = tokio::task::spawn(async move {
    let port = rx.await.unwrap();

    let inf = connector::TcpConnInfo {
      addr: format!("127.0.0.1:{port}")
    };
    let c = connector::Connector::Tcp(inf);

    let mut strm = c.connect().await.unwrap();

    println!("server: Sending 'hello' to client");
    let n = strm.write(b"hello").await.unwrap();
    assert_eq!(n, 5);

    println!("server: Waiting for 'world' reply from server");
    let mut buf = [0u8; 5];
    let n = strm.read(&mut buf[..]).await.unwrap();
    assert_eq!(n, 5);
  });

use std::{fmt, io};

/// Crate-specific errors.
#[derive(Debug)]
pub enum Error {
  /// Invalid protocol specifier.
  BadProtSpec(String),
  IO(String)
}

impl Error {
  #[allow(clippy::needless_pass_by_value)]
  pub fn bad_protspec(s: impl ToString) -> Self {
    Self::BadProtSpec(s.to_string())
  }
}

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

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

impl fmt::Display for Error {
  fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
    match self {
      Self::BadProtSpec(s) => {
        write!(f, "Unable to parse protocol specifier string; {s}")
      }
      Self::IO(s) => {
        write!(f, "I/O error; {s}")
      }
    }
  }
}

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

  /// socket path.
  #[cfg(unix)]
  Uds(PathBuf)
}

impl ProtAddr {
  #[cfg(unix)]
  #[must_use]
  pub const fn is_uds(&self) -> bool {
    match self {
      Self::Tcp(_) => false,
      Self::Uds(_) => true
    }
  }

  #[cfg(windows)]
  pub fn is_uds() -> bool {
    false
  }

  /// Returns `true` is this objects represents a TCP/IP address.
  #[must_use]
  pub const fn is_tcp(&self) -> bool {
    match self {
      Self::Tcp(_) => true,
      #[cfg(unix)]
      _ => false
    }
  }
}

impl FromStr for ProtAddr {
  type Err = ();

  /// Parse a `&str` and turn it into a `ProtAddr`.
  ///
  /// On unixy platforms if the `addr` contains one or more slashes (`/`) it is
  /// assumed the address is a unix local domain socket address.  Otherwise
  /// it is assumed the address is an IP socket address, in the form
  /// `<host>:<port>`.
  fn from_str(addr: &str) -> Result<Self, Self::Err> {
    #[cfg(unix)]
    if addr.find('/').is_some() {
      // Assume local domain socket
      Ok(Self::Uds(PathBuf::from(addr)))
    } else {
      // Assume IP socket address
      Ok(Self::Tcp(addr.to_string()))
    }

    #[cfg(windows)]
    Ok(Self::Tcp(addr.to_string()))
  }
}

impl fmt::Display for ProtAddr {
  fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
    match self {
      #[cfg(unix)]
      Self::Uds(sa) => {
        // ToDo: Return error if it's not really a valid Unicode string.
        write!(f, "{}", sa.display())
      }
      Self::Tcp(sa) => {
        write!(f, "{sa}")
      }
    }
  }
}

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

//! Utility functions specific to tokio.

pub mod client;
pub mod server;

use tokio::io::Result;

pub use client::Stream as ClientStream;
pub use server::Stream as ServerStream;

/// Unified type covering both [`ServerStream`] and [`ClientStream`] types.
pub type Stream = tokio_util::either::Either<ServerStream, ClientStream>;


/// # Errors
/// Any error returned indicates a connection failure.
///
/// # Panics
/// Only `ProtWrap::Tcp` is supported.
#[deprecated(
  since = "0.3.0",
  note = "Use `client::Connector::connect()` instead"
)]
pub async fn connect(pa: &super::ProtAddr) -> Result<client::Stream> {
  #[allow(irrefutable_let_patterns)]
  let super::ProtAddr::Tcp(addr) = pa

};

use super::Stream;

use crate::err::Error;

/// Context used to establish TCP connections.
#[derive(Clone, Debug)]
pub struct TcpConnInfo {
  /// Socket address.
  pub addr: String
}

impl FromStr for TcpConnInfo {
  type Err = Error;

  fn from_str(s: &str) -> Result<Self, Self::Err> {
    Ok(Self {
      addr: s.to_string()
    })
  }
}


/// Context used to establish unix local domain connections.
#[cfg(unix)]
#[derive(Clone, Debug)]
pub struct UdsConnInfo {
  /// Socket address pathname.
  pub fname: PathBuf
}

#[cfg(unix)]
impl FromStr for UdsConnInfo {
  type Err = Error;

  fn from_str(s: &str) -> Result<Self, Self::Err> {
    Ok(Self {
      fname: PathBuf::from(s)
    })
  }
}


/// Context used to establish TLS (based on TCP) connections.
// ToDo: Add key/cert fields
#[cfg(feature = "tls")]
#[derive(Clone, Debug)]
pub struct TlsTcpConnInfo {
  /// Socket address.
  pub addr: String
}

#[cfg(feature = "tls")]
impl FromStr for TlsTcpConnInfo {
  type Err = Error;

  fn from_str(s: &str) -> Result<Self, Self::Err> {
    Ok(Self {
      addr: s.to_string()
    })
  }
}


/// Protocol-specific connector helper.
#[derive(Clone, Debug)]
pub enum Connector {
  Tcp(TcpConnInfo),
  #[cfg(unix)]
  Uds(UdsConnInfo),
  #[cfg(feature = "tls")]
  TlsTcp(TlsTcpConnInfo)
}

impl Connector {
  /// Create a TCP connector from a string.
  ///
  /// # Errors
  /// This function will fail if the target address specification could not be
  /// parsed.
  pub fn tcp(s: &str) -> Result<Self, Error> {
    Ok(Self::Tcp(TcpConnInfo::from_str(s)?))
  }

  /// Create an unix domain socket connector from a string.
  ///
  /// # Errors
  /// This function will fail if the target address specification could not be
  /// parsed.
  #[cfg(unix)]
  pub fn uds(s: &str) -> Result<Self, Error> {
    Ok(Self::Uds(UdsConnInfo::from_str(s)?))
  }

  /// Create an TCP/TLS socket connector from a string.
  ///
  /// # Errors
  /// This function will fail if the target address specification could not be
  /// parsed.
  #[cfg(feature = "tls")]
  pub fn tls_tcp(s: &str) -> Result<Self, Error> {
    Ok(Self::TlsTcp(TlsTcpConnInfo::from_str(s)?))
  }
}


// ToDo: Add tls/tcp parameters parsing
impl FromStr for Connector {
  type Err = Error;
  fn from_str(s: &str) -> Result<Self, Self::Err> {
    #[cfg(unix)]
    if s.find('/').is_some() {
      // Assume unix domain socket
      Ok(Self::Uds(UdsConnInfo::from_str(s)?))
    } else {
      // Assume IP socket address
      Ok(Self::Tcp(TcpConnInfo::from_str(s)?))
    }

    #[cfg(windows)]
    Ok(Self::Tcp(TcpConnInfo::from_str(s)?))
  }
}

impl Connector {
  /// # Errors
  /// [`Error::IO`] indicates failure to establish connections.
  ///
  /// # Panics
  /// For now, this function will panic if:
  /// - An invalid cipher-suite configuration has been chosen.
  /// - rustls is unable to look up `localhost` name
  ///
  /// This will change in the future.
  pub async fn connect(&self) -> Result<Stream, std::io::Error> {
    match self {
      Self::Tcp(info) => {
        let strm = TcpStream::connect(&info.addr).await?;
        Ok(Stream::Tcp(strm))
      }

          .with_protocol_versions(&versions)
          .expect("inconsistent cipher-suite/versions selected")
          .with_custom_certificate_verifier(Arc::new(CertVerifier {}))
          .with_no_client_auth();

        let connector = TlsConnector::from(Arc::new(cfg));

        let raw_stream = TcpStream::connect(&info.addr).await?;

        let domain = rustls::ServerName::try_from("localhost").unwrap();
        /*
        map_err(|_| {
          io::Error::new(io::ErrorKind::InvalidInput, "invalid dnsname")
        })?;
        */

        let strm = connector.connect(domain, raw_stream).await?;

        Ok(Stream::TlsTcp(strm))
      }
    }
  }
}

}

impl SockAddr {
  /// Unwrap the [`std::net::SocketAddr`] (i.e. IPv4/IPv6) case.
  ///
  /// # Panics
  /// Will panic if the type is not `SockAddr::Std`.
  #[must_use]
  pub fn unwrap_std(self) -> std::net::SocketAddr {
    #[allow(irrefutable_let_patterns)]
    let Self::Std(s) = self
    else {
      panic!("Not SockAddr::Std()");
    };
    s
  }

  #[must_use]
  pub const fn try_as_std(&self) -> Option<&std::net::SocketAddr> {
    #[allow(irrefutable_let_patterns)]
    if let Self::Std(s) = self {
      Some(s)
    } else {
      None
    }
  }

  /// Unwrap the [`tokio::net::unix::SocketAddr`] (i.e. unix local domain
  /// socket) case.
  ///
  /// # Panics
  /// Will panic if the type is not `SockAddr::TokioUnix`.
  #[cfg(unix)]
  #[must_use]
  pub fn unwrap_tokunix(self) -> tokio::net::unix::SocketAddr {
    let Self::TokioUnix(s) = self else {
      panic!("Not SockAddr::TokioUnix()");
    };
    s
  }

  #[cfg(unix)]
  #[must_use]
  pub const fn try_as_tokunix(&self) -> Option<&tokio::net::unix::SocketAddr> {
    if let Self::TokioUnix(s) = self {
      Some(s)
    } else {
      None
    }
  }
}

impl TryFrom<SockAddr> for std::net::SocketAddr {
  type Error = SockAddr;

  fn try_from(orig: SockAddr) -> Result<Self, Self::Error> {
    match orig {
      SockAddr::Std(sa) => Ok(sa),
      #[allow(unreachable_patterns)]
      a @ SockAddr::TokioUnix(_) => Err(a)
    }
  }
}

#[cfg(unix)]
impl TryFrom<SockAddr> for tokio::net::unix::SocketAddr {
  type Error = SockAddr;

  fn try_from(orig: SockAddr) -> Result<Self, Self::Error> {
    match orig {
      SockAddr::TokioUnix(sa) => Ok(sa),
      a @ SockAddr::Std(_) => Err(a)
    }
  }
}


/// Callbacks for the [`Listener`] type.
#[async_trait]

  Tcp(TcpListenerInfo),
  #[cfg(unix)]
  Uds(UdsListenerInfo)
}

impl Listener {
  /// Create a TCP listener from a string.
  ///
  /// # Errors
  /// An error means the listener address specification string could not be
  /// parsed.
  pub fn tcp(s: &str) -> Result<Self, Error> {
    Ok(Self::Tcp(TcpListenerInfo::from_str(s)?))
  }

  /// Create an unix domain socket listener from a string.
  ///
  /// # Errors
  /// An error means the listener address specification string could not be
  /// parsed.
  #[cfg(unix)]
  pub fn uds(s: &str) -> Result<Self, Error> {
    Ok(Self::Uds(UdsListenerInfo::from_str(s)?))
  }
}

impl FromStr for Listener {
  type Err = Error;
  fn from_str(s: &str) -> Result<Self, Self::Err> {
    #[cfg(unix)]
    if s.find('/').is_some() {
      // Assume unix domain socket
      Ok(Self::Uds(UdsListenerInfo::from_str(s)?))
    } else {
      // Assume IP socket address
      Ok(Self::Tcp(TcpListenerInfo::from_str(s)?))
    }

    #[cfg(windows)]
    Ok(Self::Tcp(TcpListenerInfo::from_str(s)?))
  }
}

impl Listener {
  /// Run a listener loop.
  ///
  /// If the socket bind is successful the [`Acceptor::bound()`] of `acceptor`
  /// will be called, where the bound socket address will be passed as an
  /// argument.  (This can be used to retreive the port number if the
  /// application requested the port number to be automatically assigned.
  ///
  /// Each time a client has connected the acceptor will call
  /// [`Acceptor::connected()`] to allow the application to process the
  /// connection.  The ownership of the newly established connection will be
  /// passed to the `connected()` method.
  ///
  /// # Unix domain sockets
  /// If the listener is a unix domain socket, the socket file will be removed
  /// if the listener is aborted.
  ///
  /// # Errors
  /// [`Error::IO`] can indicate the inability to bind socket.
  ///
  /// For "forced" UDS sockets, an `ErrorKind::Other` is returned if the file
  /// already exists and is not a socket.
  pub async fn run(
    &self,
    ks: KillSwitch,
    mut acceptor: impl Acceptor + Send
  ) -> Result<(), std::io::Error> {
    match self {
      Self::Tcp(info) => {
        let listener = TcpListener::bind(&info.addr).await?;

        let sa = listener.local_addr()?;
        acceptor.bound(self, SockAddr::Std(sa)).await;

        loop {
          tokio::select! {
            ret = listener.accept() => {
              let (strm, sa) = ret?;
              let sa = SockAddr::Std(sa);
              acceptor.connected(sa, Stream::Tcp(strm)).await;
            }
            () = ks.wait() => {
              break;
            }
          }
        }

        drop(listener);

        acceptor.unbound(self).await;
      }

      #[cfg(unix)]
      Self::Uds(info) => {
        if info.mkdir {
          if let Some(dir) = info.fname.parent() {
            fs::create_dir_all(dir).await?;
          }
        }
        if info.force && info.fname.exists() {
          let md = fs::metadata(&info.fname).await?;

        loop {
          tokio::select! {
            ret = listener.accept() => {
              let (strm, sa) = ret?;
              let sa = SockAddr::TokioUnix(sa);
              acceptor.connected(sa, Stream::Uds(strm)).await;
            }
            () = ks.wait() => {
              break;
            }
          }
        }

        drop(listener);

      let killswitch = self.ks.clone();
      tokio::task::spawn(async move {
        let mut buf = [0u8; 5];

        let n = strm.read(&mut buf[..]).await.unwrap();
        assert_eq!(n, 5);
        assert_eq!(buf, *b"hello");

        let n = strm.write(b"world").await.unwrap();
        assert_eq!(n, 5);

        killswitch.trigger();
      });
    }
  }

      listener.run(killswitch, acceptor).await.unwrap();
    });

    let jh_client = tokio::task::spawn(async move {
      // Use side-channel to receive port number from server
      let port = rx.await.unwrap();

      let addr = format!("127.0.0.1:{port}");
      let c = connector::Connector::from_str(&addr).unwrap();

      let mut strm = c.connect().await.unwrap();

      println!("server: Sending 'hello' to client");
      let n = strm.write(b"hello").await.unwrap();
      assert_eq!(n, 5);

      let mut buf = [0u8; 5];
      let n = strm.read(&mut buf[..]).await.unwrap();
      assert_eq!(n, 5);
      assert_eq!(buf, *b"world");
    });

    ks.wait().await;

    jh_client.await.unwrap();
    jh_server.await.unwrap();
  }
}

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

# Change Log

## [Unreleased]

[Details](/vdiff?from=protwrap-0.3.1&to=trunk)

### Added

### Changed

### Removed

---

## [0.3.1] - 2024-10-07

[Details](/vdiff?from=protwrap-0.3.0&to=protwrap-0.3.1)

### Changed

- Derive `Debug` and `Clone` on `tokio::client::connector::Connector`.

---

## [0.3.0] - 2024-05-31

[Details](/vdiff?from=protwrap-0.2.2&to=protwrap-0.3.0)

This is a major rewrite.