use std::{
collections::HashMap,
pin::Pin,
sync::Arc,
task::{
Context,
Poll,
},
};
use async_trait::async_trait;
use futures::Stream;
use muxado::Error as MuxadoError;
use thiserror::Error;
use tokio::sync::mpsc::Receiver;
use crate::{
Session,
config::{
HttpTunnelBuilder,
LabeledTunnelBuilder,
TcpTunnelBuilder,
TlsTunnelBuilder,
},
conn::{
ConnInner,
EdgeConn,
EndpointConn,
},
internals::raw_session::RpcError,
session::ConnectError,
};
#[derive(Error, Debug, Clone)]
#[non_exhaustive]
pub enum AcceptError {
#[error("transport error")]
Transport(#[from] MuxadoError),
#[error("reconnect error")]
Reconnect(#[from] Arc<ConnectError>),
#[error("listener closed: {message}{}", error_code.clone().map(|s| format!(", {s}")).unwrap_or_else(String::new))]
ListenerClosed {
message: String,
error_code: Option<String>,
},
}
#[derive(Clone)]
pub(crate) struct TunnelInnerInfo {
pub(crate) id: String,
pub(crate) proto: String,
pub(crate) url: String,
pub(crate) labels: HashMap<String, String>,
pub(crate) forwards_to: String,
pub(crate) metadata: String,
}
pub(crate) struct TunnelInner {
pub(crate) info: TunnelInnerInfo,
pub(crate) incoming: Option<Receiver<Result<ConnInner, AcceptError>>>,
pub(crate) session: Session,
}
impl Drop for TunnelInner {
fn drop(&mut self) {
let id = self.id().to_string();
let sess = self.session.clone();
let rt = sess.runtime();
rt.spawn(async move { sess.close_tunnel(&id).await });
}
}
macro_rules! tunnel_trait {
($($hyper_bound:tt)*) => {
pub trait Tunnel:
Stream<Item = Result<<Self as Tunnel>::Conn, AcceptError>>
+ TunnelInfo
+ TunnelCloser
$($hyper_bound)*
+ Unpin
+ Send
+ 'static
{
type Conn: crate::Conn;
}
pub trait TunnelInfo {
fn id(&self) -> &str;
fn forwards_to(&self) -> &str;
fn metadata(&self) -> &str;
}
#[async_trait]
pub trait TunnelCloser {
async fn close(&mut self) -> Result<(), RpcError>;
}
}
}
tunnel_trait!();
pub trait EndpointInfo {
fn url(&self) -> &str;
fn proto(&self) -> &str;
}
pub trait EdgeInfo {
fn labels(&self) -> &HashMap<String, String>;
}
impl Stream for TunnelInner {
type Item = Result<ConnInner, AcceptError>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
self.incoming
.as_mut()
.expect("tunnel inner lacks a receiver")
.poll_recv(cx)
}
}
impl TunnelInner {
pub fn id(&self) -> &str {
&self.info.id
}
pub fn url(&self) -> &str {
&self.info.url
}
pub async fn close(&mut self) -> Result<(), RpcError> {
self.session.close_tunnel(self.id()).await?;
if let Some(r) = self.incoming.as_mut() {
r.close()
}
Ok(())
}
pub fn proto(&self) -> &str {
&self.info.proto
}
pub fn labels(&self) -> &HashMap<String, String> {
&self.info.labels
}
pub fn forwards_to(&self) -> &str {
&self.info.forwards_to
}
pub fn metadata(&self) -> &str {
&self.info.metadata
}
pub(crate) fn make_info(&self) -> TunnelInner {
TunnelInner {
info: self.info.clone(),
incoming: None,
session: self.session.clone(),
}
}
}
macro_rules! make_tunnel_type {
($(#[$outer:meta])* $wrapper:ident, $builder:tt, $conn:tt, $($m:tt),*) => {
$(#[$outer])*
pub struct $wrapper {
pub(crate) inner: TunnelInner,
}
impl $wrapper {
pub(crate) fn make_info(&self) -> $wrapper {
$wrapper {
inner: self.inner.make_info(),
}
}
}
impl Tunnel for $wrapper {
type Conn = $conn;
}
impl TunnelInfo for $wrapper {
fn id(&self) -> &str {
self.inner.id()
}
fn forwards_to(&self) -> &str {
self.inner.forwards_to()
}
fn metadata(&self) -> &str {
self.inner.metadata()
}
}
#[async_trait]
impl TunnelCloser for $wrapper {
async fn close(&mut self) -> Result<(), RpcError> {
self.inner.close().await
}
}
impl $wrapper {
pub fn builder(session: Session) -> $builder {
$builder::from(session)
}
}
$(
make_tunnel_type!($m; $wrapper);
)*
impl Stream for $wrapper {
type Item = Result<$conn, AcceptError>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
Pin::new(&mut self.inner).poll_next(cx).map(|o| o.map(|r| r.map(|c| $conn { inner: c })))
}
}
};
(endpoint; $wrapper:ty) => {
impl EndpointInfo for $wrapper {
fn url(&self) -> &str {
self.inner.url()
}
fn proto(&self) -> &str {
self.inner.proto()
}
}
};
(edge; $wrapper:ty) => {
impl EdgeInfo for $wrapper {
fn labels(&self) -> &HashMap<String, String> {
self.inner.labels()
}
}
};
}
make_tunnel_type! {
HttpTunnel, HttpTunnelBuilder, EndpointConn, endpoint
}
make_tunnel_type! {
TcpTunnel, TcpTunnelBuilder, EndpointConn, endpoint
}
make_tunnel_type! {
TlsTunnel, TlsTunnelBuilder, EndpointConn, endpoint
}
make_tunnel_type! {
LabeledTunnel, LabeledTunnelBuilder, EdgeConn, edge
}