mini-serve 0.6.0

use std::fmt;
use std::time::Duration;

use hyper::{body::Frame, StatusCode};
use http_body::Body;
use http_body_util::combinators::BoxBody;
use hyper::body::Bytes;
use mini_serve::{RouteBuilder, handler, BodyError};
use tokio::net::TcpListener;
use tokio::sync::oneshot;

/// A test double body that yields some frames then errors.
struct FailingBody {
	index: std::sync::atomic::AtomicUsize,
}

impl FailingBody {
	fn new() -> Self {
		FailingBody {
			index: std::sync::atomic::AtomicUsize::new(0),
		}
	}
}

#[derive(Debug)]
struct TestBodyError;
impl fmt::Display for TestBodyError {
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		write!(f, "simulated mid-stream read error")
	}
}
impl std::error::Error for TestBodyError {}

impl Body for FailingBody {
	type Data = Bytes;
	type Error = BodyError;

	fn poll_frame(
		self: std::pin::Pin<&mut Self>,
		_cx: &mut std::task::Context<'_>,
	) -> std::task::Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
		let idx = self.index.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
		match idx {
			0 => std::task::Poll::Ready(Some(Ok(Frame::data(Bytes::from("chunk1"))))),
			1 => std::task::Poll::Ready(Some(Ok(Frame::data(Bytes::from("chunk2"))))),
			_ => std::task::Poll::Ready(Some(Err(BodyError::new(TestBodyError)))),
		}
	}

	fn is_end_stream(&self) -> bool {
		let idx = self.index.load(std::sync::atomic::Ordering::SeqCst);
		idx > 2
	}
}

#[tokio::test]
async fn body_stream_error_aborts_connection_not_empty_200() {
	let app = RouteBuilder::stateless()
		.get("/stream_error", handler(|_req, _state| async {
			let body = BoxBody::new(FailingBody::new());
			Ok(hyper::Response::builder()
				.status(StatusCode::OK)
				.body(body)
				.unwrap())
		}))
		.seal();

	let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
	let addr = listener.local_addr().unwrap();
	let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();

	let run_task = tokio::spawn(async move {
		app.run(listener, async move {
			let _ = shutdown_rx.await;
		})
		.await
	});

	let client = reqwest::Client::builder()
		.build()
		.unwrap();
	let res = client
		.get(format!("http://{addr}/stream_error"))
		.send()
		.await;

	let (is_err, content_len, body_text) = match res {
		Ok(resp) => {
			let status = resp.status();
			let text = resp.text().await;
			(
				false,
				status == StatusCode::OK,
				text.unwrap_or_default(),
			)
		}
		Err(err) => {
			(
				true,
				err.is_body(),
				String::new(),
			)
		}
	};

	// The connection should fail (body stream error aborts it), not return an empty 200.
	// reqwest will see this as either an error (preferred, connection reset) or
	// an incomplete body.
	assert!(
		is_err || (content_len && body_text.len() < 12),
		"expected connection error or incomplete body, got status OK with full body"
	);

	shutdown_tx.send(()).unwrap();
	let result = tokio::time::timeout(Duration::from_secs(2), run_task)
		.await
		.expect("run() did not return after shutdown")
		.expect("run task panicked");
	assert!(result.is_ok());
}