use std::future::Future;
use std::pin::Pin;
use std::sync::Mutex;
use gunnar_core::{Fingerprint, Level, Principal, Repo};
use tokio::io::{AsyncRead, AsyncWrite};
use crate::command::GitService;
pub trait DuplexStream: AsyncRead + AsyncWrite + Unpin + Send {}
impl<T: AsyncRead + AsyncWrite + Unpin + Send> DuplexStream for T {}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum Delivery {
#[default]
Session,
Advertisement,
StatelessRpc,
}
#[derive(Debug, Clone)]
pub struct ServiceRequest {
pub service: GitService,
pub repo: Repo,
pub prospective: bool,
pub principal: Principal,
pub access: Level,
pub key_fingerprint: Option<Fingerprint>,
pub git_protocol: Option<String>,
pub delivery: Delivery,
pub credential: Option<String>,
pub plan_notice: Option<String>,
pub pack_ceiling: Option<crate::plan::PackCeiling>,
pub pack_mark: Option<gunnar_lfs::Mark>,
pub lfs_allowance: Option<gunnar_lfs::NamespaceAllowance>,
pub cancel: crate::channel::Cancel,
}
impl ServiceRequest {
pub fn with_git_protocol(mut self, value: Option<String>) -> Self {
self.git_protocol = value;
self
}
pub fn with_delivery(mut self, delivery: Delivery) -> Self {
self.delivery = delivery;
self
}
pub fn with_credential(mut self, credential: Option<String>) -> Self {
self.credential = credential;
self
}
}
#[derive(Debug, thiserror::Error)]
pub enum TransportError {
#[error("protocol error: {0}")]
Protocol(String),
#[error("i/o error: {0}")]
Io(#[from] std::io::Error),
#[error("not implemented: {0}; the wire seam is not connected to gunnar-wire yet")]
Unimplemented(&'static str),
#[error("{0}")]
Refused(String),
}
impl TransportError {
pub fn client_message(&self) -> Option<String> {
match self {
TransportError::Protocol(_) | TransportError::Io(_) => None,
TransportError::Unimplemented(_) => Some(self.to_string()),
TransportError::Refused(message) => Some(message.clone()),
}
}
}
pub struct LfsAccess {
pub store: std::sync::Arc<dyn gunnar_lfs::LfsStore>,
pub limits: gunnar_lfs::Limits,
pub allowance: Option<gunnar_lfs::NamespaceAllowance>,
}
impl std::fmt::Debug for LfsAccess {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("LfsAccess")
.field("limits", &self.limits)
.finish_non_exhaustive()
}
}
pub trait LfsTransport: Send + Sync + 'static {
fn open_lfs(&self, request: &ServiceRequest) -> Result<LfsAccess, TransportError>;
}
pub trait GitTransport: Send + Sync + 'static {
fn serve<'a>(
&'a self,
request: ServiceRequest,
stream: Box<dyn DuplexStream>,
) -> Pin<Box<dyn Future<Output = Result<(), TransportError>> + Send + 'a>>;
}
#[derive(Debug, Default)]
pub struct BannerTransport {
served: Mutex<Vec<String>>,
git_protocols: Mutex<Vec<Option<String>>>,
settle_after_release: bool,
fail_at_the_end: bool,
refuse_before_writing: Option<String>,
hold: Option<tokio::sync::watch::Receiver<bool>>,
}
#[derive(Debug)]
pub struct TransportHold {
tx: tokio::sync::watch::Sender<bool>,
}
impl TransportHold {
pub fn release(&self) {
let _ = self.tx.send(true);
}
}
impl Drop for TransportHold {
fn drop(&mut self) {
self.release();
}
}
impl BannerTransport {
pub fn new() -> Self {
Self::default()
}
pub fn settling_after_it_releases_the_stream() -> Self {
BannerTransport {
settle_after_release: true,
..Self::default()
}
}
pub fn holding() -> (Self, TransportHold) {
let (tx, rx) = tokio::sync::watch::channel(false);
(
BannerTransport {
hold: Some(rx),
..Self::default()
},
TransportHold { tx },
)
}
pub fn refusing(message: impl Into<String>) -> Self {
Self {
refuse_before_writing: Some(message.into()),
..Self::default()
}
}
pub fn that_then_fails(mut self) -> Self {
self.fail_at_the_end = true;
self
}
pub fn served(&self) -> Vec<String> {
self.served.lock().expect("banner transport lock").clone()
}
pub fn git_protocols(&self) -> Vec<Option<String>> {
self.git_protocols
.lock()
.expect("banner transport lock")
.clone()
}
pub fn banner(request: &ServiceRequest) -> String {
format!(
"{} {} {}\n",
request.service.verb(),
request.repo.key(),
request.principal.id
)
}
}
impl GitTransport for BannerTransport {
fn serve<'a>(
&'a self,
request: ServiceRequest,
mut stream: Box<dyn DuplexStream>,
) -> Pin<Box<dyn Future<Output = Result<(), TransportError>> + Send + 'a>> {
Box::pin(async move {
use tokio::io::AsyncWriteExt;
if let Some(message) = &self.refuse_before_writing {
return Err(TransportError::Refused(message.clone()));
}
let banner = BannerTransport::banner(&request);
self.served
.lock()
.expect("banner transport lock")
.push(banner.trim_end().to_string());
self.git_protocols
.lock()
.expect("banner transport lock")
.push(request.git_protocol.clone());
stream.write_all(banner.as_bytes()).await?;
stream.flush().await?;
if let Some(hold) = &self.hold {
let mut hold = hold.clone();
let released = async {
while !*hold.borrow_and_update() {
if hold.changed().await.is_err() {
break;
}
}
};
tokio::select! {
() = released => {}
() = request.cancel.cancelled() => {
return Err(TransportError::Io(std::io::Error::new(
std::io::ErrorKind::ConnectionAborted,
"the banner double was told to stop: its client is gone",
)));
}
}
}
stream.shutdown().await?;
if self.settle_after_release {
drop(stream);
for _ in 0..16 {
tokio::task::yield_now().await;
}
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
}
if self.fail_at_the_end {
return Err(TransportError::Protocol(
"the banner double was asked to report a failed transfer".to_string(),
));
}
Ok(())
})
}
}