use crate::{
Html,
healthcheck::{HealthCheck, HealthStatus},
method::{Method, ReadState, WriteState},
metrics::Metrics,
middleware::{ErrorHandler, error_handler},
request::RequestParams,
route::{self, *},
socket,
};
use async_std::sync::Arc;
use async_trait::async_trait;
use derivative::Derivative;
use futures::{
future::{BoxFuture, FutureExt},
stream::BoxStream,
};
use maud::{PreEscaped, html};
use semver::Version;
use serde::{Deserialize, Serialize, de::DeserializeOwned};
use serde_with::{DisplayFromStr, serde_as};
use snafu::{OptionExt, ResultExt, Snafu};
use std::{
borrow::Cow,
collections::hash_map::{Entry, HashMap, IntoValues, Values},
convert::Infallible,
fmt::Display,
fs,
marker::PhantomData,
ops::Index,
path::{Path, PathBuf},
};
use tide::http::content::Accept;
use vbs::version::StaticVersionType;
#[derive(Clone, Debug, Snafu, PartialEq, Eq)]
pub enum ApiError {
Route { source: RouteParseError },
ApiMustBeTable,
MissingRoutesTable,
RoutesMustBeTable,
UndefinedRoute,
HandlerAlreadyRegistered,
IncorrectMethod { expected: Method, actual: Method },
InvalidMetaTable { source: toml::de::Error },
MissingFormatVersion,
InvalidFormatVersion,
AmbiguousRoutes { route1: String, route2: String },
CannotReadToml { reason: String },
}
#[serde_as]
#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
pub struct ApiVersion {
#[serde_as(as = "Option<DisplayFromStr>")]
pub api_version: Option<Version>,
#[serde_as(as = "DisplayFromStr")]
pub spec_version: Version,
}
#[serde_as]
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
pub struct ApiMetadata {
#[serde(default = "meta_defaults::name")]
pub name: String,
#[serde(default = "meta_defaults::description")]
pub description: String,
#[serde_as(as = "DisplayFromStr")]
#[serde(default = "meta_defaults::format_version")]
pub format_version: Version,
#[serde(default = "meta_defaults::html_top")]
pub html_top: String,
#[serde(default = "meta_defaults::html_bottom")]
pub html_bottom: String,
#[serde(default = "meta_defaults::heading_entry")]
pub heading_entry: String,
#[serde(default = "meta_defaults::heading_routes")]
pub heading_routes: String,
#[serde(default = "meta_defaults::heading_parameters")]
pub heading_parameters: String,
#[serde(default = "meta_defaults::heading_description")]
pub heading_description: String,
#[serde(default = "meta_defaults::route_path")]
pub route_path: String,
#[serde(default = "meta_defaults::parameter_table_open")]
pub parameter_table_open: String,
#[serde(default = "meta_defaults::parameter_table_close")]
pub parameter_table_close: String,
#[serde(default = "meta_defaults::parameter_row")]
pub parameter_row: String,
#[serde(default = "meta_defaults::parameter_none")]
pub parameter_none: String,
}
impl Default for ApiMetadata {
fn default() -> Self {
toml::Value::Table(Default::default()).try_into().unwrap()
}
}
mod meta_defaults {
use super::Version;
pub fn name() -> String {
"default-tide-disco-api".to_string()
}
pub fn description() -> String {
"Default Tide Disco API".to_string()
}
pub fn format_version() -> Version {
"0.1.0".parse().unwrap()
}
pub fn html_top() -> String {
"
<!DOCTYPE html>
<html lang='en'>
<head>
<meta charset='utf-8'>
<title>{{NAME}} Reference</title>
<link rel='stylesheet' href='{{PUBLIC}}/css/style.css'>
<script src='{{PUBLIC}}/js/script.js'></script>
<link rel='icon' type='image/svg+xml'
href='/public/favicon.svg'>
</head>
<body>
<div><a href='/'><img src='{{PUBLIC}}/espressosys_logo.svg'
alt='Espresso Systems Logo'
/></a></div>
<h1>{{NAME}} API {{VERSION}} Reference</h1>
<p>{{SHORT_DESCRIPTION}}</p><br/>
{{LONG_DESCRIPTION}}
"
.to_string()
}
pub fn html_bottom() -> String {
"
<h1> </h1>
<p>Copyright © 2022 Espresso Systems. All rights reserved.</p>
</body>
</html>
"
.to_string()
}
pub fn heading_entry() -> String {
"<a name='{{NAME}}'><h3 class='entry'><span class='meth'>{{METHOD}}</span> {{NAME}}</h3></a>\n".to_string()
}
pub fn heading_routes() -> String {
"<h3>Routes</h3>\n".to_string()
}
pub fn heading_parameters() -> String {
"<h3>Parameters</h3>\n".to_string()
}
pub fn heading_description() -> String {
"<h3>Description</h3>\n".to_string()
}
pub fn route_path() -> String {
"<p class='path'>{{PATH}}</p>\n".to_string()
}
pub fn parameter_table_open() -> String {
"<table>\n".to_string()
}
pub fn parameter_table_close() -> String {
"</table>\n\n".to_string()
}
pub fn parameter_row() -> String {
"<tr><td class='parameter'>{{NAME}}</td><td class='type'>{{TYPE}}</td></tr>\n".to_string()
}
pub fn parameter_none() -> String {
"<div class='meta'>None</div>".to_string()
}
}
#[derive(Derivative)]
#[derivative(Debug(bound = ""))]
pub struct Api<State, Error, VER> {
inner: ApiInner<State, Error>,
_version: PhantomData<VER>,
}
#[derive(Derivative)]
#[derivative(Debug(bound = ""))]
pub(crate) struct ApiInner<State, Error> {
meta: Arc<ApiMetadata>,
name: String,
routes: HashMap<String, Route<State, Error>>,
routes_by_path: HashMap<String, Vec<String>>,
#[derivative(Debug = "ignore")]
health_check: HealthCheckHandler<State>,
api_version: Option<Version>,
#[derivative(Debug = "ignore")]
error_handler: Option<Arc<dyn ErrorHandler<Error>>>,
#[derivative(Debug = "ignore")]
version_handler: Arc<dyn VersionHandler>,
public: Option<PathBuf>,
short_description: String,
long_description: String,
}
pub(crate) trait VersionHandler:
Send + Sync + Fn(&Accept, ApiVersion) -> Result<tide::Response, RouteError<Infallible>>
{
}
impl<F> VersionHandler for F where
F: Send + Sync + Fn(&Accept, ApiVersion) -> Result<tide::Response, RouteError<Infallible>>
{
}
impl<'a, State, Error> IntoIterator for &'a ApiInner<State, Error> {
type Item = &'a Route<State, Error>;
type IntoIter = Values<'a, String, Route<State, Error>>;
fn into_iter(self) -> Self::IntoIter {
self.routes.values()
}
}
impl<State, Error> IntoIterator for ApiInner<State, Error> {
type Item = Route<State, Error>;
type IntoIter = IntoValues<String, Route<State, Error>>;
fn into_iter(self) -> Self::IntoIter {
self.routes.into_values()
}
}
impl<State, Error> Index<&str> for ApiInner<State, Error> {
type Output = Route<State, Error>;
fn index(&self, index: &str) -> &Route<State, Error> {
&self.routes[index]
}
}
pub(crate) struct RoutesWithPath<'a, State, Error> {
routes: std::slice::Iter<'a, String>,
api: &'a ApiInner<State, Error>,
}
impl<'a, State, Error> Iterator for RoutesWithPath<'a, State, Error> {
type Item = &'a Route<State, Error>;
fn next(&mut self) -> Option<Self::Item> {
Some(&self.api.routes[self.routes.next()?])
}
}
impl<State, Error> ApiInner<State, Error> {
pub(crate) fn routes_by_path(
&self,
) -> impl Iterator<Item = (&str, RoutesWithPath<'_, State, Error>)> {
self.routes_by_path.iter().map(|(path, routes)| {
(
path.as_str(),
RoutesWithPath {
routes: routes.iter(),
api: self,
},
)
})
}
pub(crate) async fn health(&self, req: RequestParams, state: &State) -> tide::Response {
(self.health_check)(req, state).await
}
pub(crate) fn version(&self) -> ApiVersion {
ApiVersion {
api_version: self.api_version.clone(),
spec_version: self.meta.format_version.clone(),
}
}
pub(crate) fn public(&self) -> Option<&PathBuf> {
self.public.as_ref()
}
pub(crate) fn set_name(&mut self, name: String) {
self.name = name;
}
pub(crate) fn documentation(&self) -> Html {
html! {
(PreEscaped(self.meta.html_top
.replace("{{NAME}}", &self.name)
.replace("{{SHORT_DESCRIPTION}}", &self.short_description)
.replace("{{LONG_DESCRIPTION}}", &self.long_description)
.replace("{{VERSION}}", &match &self.api_version {
Some(version) => version.to_string(),
None => "(no version)".to_string(),
})
.replace("{{FORMAT_VERSION}}", &self.meta.format_version.to_string())
.replace("{{PUBLIC}}", &format!("/public/{}", self.name))))
@for route in self.routes.values() {
(route.documentation())
}
(PreEscaped(&self.meta.html_bottom))
}
}
pub(crate) fn short_description(&self) -> &str {
&self.short_description
}
pub(crate) fn error_handler(&self) -> Arc<dyn ErrorHandler<Error>> {
self.error_handler.clone().unwrap()
}
pub(crate) fn version_handler(&self) -> Arc<dyn VersionHandler> {
self.version_handler.clone()
}
}
impl<State, Error, VER> Api<State, Error, VER>
where
State: 'static,
Error: 'static,
VER: StaticVersionType + 'static,
{
pub fn new(api: impl Into<toml::Value>) -> Result<Self, ApiError> {
let mut api = api.into();
let meta = match api
.as_table_mut()
.context(ApiMustBeTableSnafu)?
.remove("meta")
{
Some(meta) => toml::Value::try_into(meta)
.map_err(|source| ApiError::InvalidMetaTable { source })?,
None => ApiMetadata::default(),
};
let meta = Arc::new(meta);
let routes = match api.get("route") {
Some(routes) => routes.as_table().context(RoutesMustBeTableSnafu)?,
None => return Err(ApiError::MissingRoutesTable),
};
let routes = routes
.into_iter()
.map(|(name, spec)| {
let route = Route::new(name.clone(), spec, meta.clone()).context(RouteSnafu)?;
Ok((route.name(), route))
})
.collect::<Result<HashMap<_, _>, _>>()?;
let mut routes_by_path = HashMap::new();
for route in routes.values() {
for path in route.patterns() {
match routes_by_path.entry(path.clone()) {
Entry::Vacant(e) => e.insert(Vec::new()).push(route.name().clone()),
Entry::Occupied(mut e) => {
if let Some(ambiguous_name) = e
.get()
.iter()
.find(|name| routes[*name].method() == route.method())
{
return Err(ApiError::AmbiguousRoutes {
route1: route.name(),
route2: ambiguous_name.clone(),
});
}
e.get_mut().push(route.name());
}
}
}
}
let blocks = markdown::tokenize(&meta.description);
let (short_description, long_description) = match blocks.split_first() {
Some((short, long)) => {
let render = |blocks| markdown::to_html(&markdown::generate_markdown(blocks));
let short = render(vec![short.clone()]);
let long = render(long.to_vec());
let short = short.strip_prefix("<p>").unwrap_or(&short);
let short = short.strip_suffix("</p>").unwrap_or(short);
let short = short.to_string();
(short, long)
}
None => Default::default(),
};
Ok(Self {
inner: ApiInner {
name: meta.name.clone(),
meta,
routes,
routes_by_path,
health_check: Box::new(Self::default_health_check),
api_version: None,
error_handler: None,
version_handler: Arc::new(|accept, version| {
respond_with(accept, version, VER::instance())
}),
public: None,
short_description,
long_description,
},
_version: Default::default(),
})
}
pub fn from_file<P: AsRef<Path>>(path: P) -> Result<Self, ApiError> {
let bytes = fs::read(path).map_err(|err| ApiError::CannotReadToml {
reason: err.to_string(),
})?;
let string = std::str::from_utf8(&bytes).map_err(|err| ApiError::CannotReadToml {
reason: err.to_string(),
})?;
Self::new(toml::from_str::<toml::Value>(string).map_err(|err| {
ApiError::CannotReadToml {
reason: err.to_string(),
}
})?)
}
pub fn with_version(&mut self, version: Version) -> &mut Self {
self.inner.api_version = Some(version);
self
}
pub fn with_public(&mut self, dir: PathBuf) -> &mut Self {
self.inner.public = Some(dir);
self
}
pub fn at<F, T>(&mut self, name: &str, handler: F) -> Result<&mut Self, ApiError>
where
F: 'static + Send + Sync + Fn(RequestParams, &State) -> BoxFuture<'_, Result<T, Error>>,
T: Serialize,
State: 'static + Send + Sync,
VER: 'static + Send + Sync,
{
let route = self
.inner
.routes
.get_mut(name)
.ok_or(ApiError::UndefinedRoute)?;
if route.has_handler() {
return Err(ApiError::HandlerAlreadyRegistered);
}
if !route.method().is_http() {
return Err(ApiError::IncorrectMethod {
expected: Method::get(),
actual: route.method(),
});
}
route
.set_fn_handler(handler, VER::instance())
.unwrap_or_else(|_| panic!("unexpected failure in set_fn_handler"));
Ok(self)
}
fn method_immutable<F, T>(
&mut self,
method: Method,
name: &str,
handler: F,
) -> Result<&mut Self, ApiError>
where
F: 'static
+ Send
+ Sync
+ Fn(RequestParams, &<State as ReadState>::State) -> BoxFuture<'_, Result<T, Error>>,
T: Serialize,
State: 'static + Send + Sync + ReadState,
VER: 'static + Send + Sync + StaticVersionType,
{
assert!(method.is_http() && !method.is_mutable());
let route = self
.inner
.routes
.get_mut(name)
.ok_or(ApiError::UndefinedRoute)?;
if route.method() != method {
return Err(ApiError::IncorrectMethod {
expected: method,
actual: route.method(),
});
}
if route.has_handler() {
return Err(ApiError::HandlerAlreadyRegistered);
}
route
.set_handler(ReadHandler::<_, VER>::from(handler))
.unwrap_or_else(|_| panic!("unexpected failure in set_handler"));
Ok(self)
}
pub fn get<F, T>(&mut self, name: &str, handler: F) -> Result<&mut Self, ApiError>
where
F: 'static
+ Send
+ Sync
+ Fn(RequestParams, &<State as ReadState>::State) -> BoxFuture<'_, Result<T, Error>>,
T: Serialize,
State: 'static + Send + Sync + ReadState,
VER: 'static + Send + Sync,
{
self.method_immutable(Method::get(), name, handler)
}
fn method_mutable<F, T>(
&mut self,
method: Method,
name: &str,
handler: F,
) -> Result<&mut Self, ApiError>
where
F: 'static
+ Send
+ Sync
+ Fn(RequestParams, &mut <State as ReadState>::State) -> BoxFuture<'_, Result<T, Error>>,
T: Serialize,
State: 'static + Send + Sync + WriteState,
VER: 'static + Send + Sync,
{
assert!(method.is_http() && method.is_mutable());
let route = self
.inner
.routes
.get_mut(name)
.ok_or(ApiError::UndefinedRoute)?;
if route.method() != method {
return Err(ApiError::IncorrectMethod {
expected: method,
actual: route.method(),
});
}
if route.has_handler() {
return Err(ApiError::HandlerAlreadyRegistered);
}
route
.set_handler(WriteHandler::<_, VER>::from(handler))
.unwrap_or_else(|_| panic!("unexpected failure in set_handler"));
Ok(self)
}
pub fn post<F, T>(&mut self, name: &str, handler: F) -> Result<&mut Self, ApiError>
where
F: 'static
+ Send
+ Sync
+ Fn(RequestParams, &mut <State as ReadState>::State) -> BoxFuture<'_, Result<T, Error>>,
T: Serialize,
State: 'static + Send + Sync + WriteState,
VER: 'static + Send + Sync,
{
self.method_mutable(Method::post(), name, handler)
}
pub fn put<F, T>(&mut self, name: &str, handler: F) -> Result<&mut Self, ApiError>
where
F: 'static
+ Send
+ Sync
+ Fn(RequestParams, &mut <State as ReadState>::State) -> BoxFuture<'_, Result<T, Error>>,
T: Serialize,
State: 'static + Send + Sync + WriteState,
VER: 'static + Send + Sync,
{
self.method_mutable(Method::put(), name, handler)
}
pub fn delete<F, T>(&mut self, name: &str, handler: F) -> Result<&mut Self, ApiError>
where
F: 'static
+ Send
+ Sync
+ Fn(RequestParams, &mut <State as ReadState>::State) -> BoxFuture<'_, Result<T, Error>>,
T: Serialize,
State: 'static + Send + Sync + WriteState,
VER: 'static + Send + Sync,
{
self.method_mutable(Method::delete(), name, handler)
}
pub fn socket<F, ToClient, FromClient>(
&mut self,
name: &str,
handler: F,
) -> Result<&mut Self, ApiError>
where
F: 'static
+ Send
+ Sync
+ Fn(
RequestParams,
socket::Connection<ToClient, FromClient, Error, VER>,
&State,
) -> BoxFuture<'_, Result<(), Error>>,
ToClient: 'static + Serialize + ?Sized,
FromClient: 'static + DeserializeOwned,
State: 'static + Send + Sync,
Error: 'static + Send + Display,
{
self.register_socket_handler(name, socket::handler(handler))
}
pub fn stream<F, Msg>(&mut self, name: &str, handler: F) -> Result<&mut Self, ApiError>
where
F: 'static + Send + Sync + Fn(RequestParams, &State) -> BoxStream<Result<Msg, Error>>,
Msg: 'static + Serialize + Send + Sync,
State: 'static + Send + Sync,
Error: 'static + Send + Display,
VER: 'static + Send + Sync,
{
self.register_socket_handler(name, socket::stream_handler::<_, _, _, _, VER>(handler))
}
fn register_socket_handler(
&mut self,
name: &str,
handler: socket::Handler<State, Error>,
) -> Result<&mut Self, ApiError> {
let route = self
.inner
.routes
.get_mut(name)
.ok_or(ApiError::UndefinedRoute)?;
if route.method() != Method::Socket {
return Err(ApiError::IncorrectMethod {
expected: Method::Socket,
actual: route.method(),
});
}
if route.has_handler() {
return Err(ApiError::HandlerAlreadyRegistered);
}
route
.set_socket_handler(handler)
.unwrap_or_else(|_| panic!("unexpected failure in set_socket_handler"));
Ok(self)
}
pub fn metrics<F, T>(&mut self, name: &str, handler: F) -> Result<&mut Self, ApiError>
where
F: 'static
+ Send
+ Sync
+ Fn(RequestParams, &State::State) -> BoxFuture<Result<Cow<T>, Error>>,
T: 'static + Clone + Metrics,
State: 'static + Send + Sync + ReadState,
Error: 'static,
VER: 'static + Send + Sync,
{
let route = self
.inner
.routes
.get_mut(name)
.ok_or(ApiError::UndefinedRoute)?;
if route.method() != Method::Metrics {
return Err(ApiError::IncorrectMethod {
expected: Method::Metrics,
actual: route.method(),
});
}
if route.has_handler() {
return Err(ApiError::HandlerAlreadyRegistered);
}
route
.set_metrics_handler(handler)
.unwrap_or_else(|_| panic!("unexpected failure in set_metrics_handler"));
Ok(self)
}
pub fn with_health_check<H>(
&mut self,
handler: impl 'static + Send + Sync + Fn(&State) -> BoxFuture<H>,
) -> &mut Self
where
State: 'static + Send + Sync,
H: 'static + HealthCheck,
VER: 'static + Send + Sync,
{
self.inner.health_check = route::health_check_handler::<_, _, VER>(handler);
self
}
pub(crate) fn map_err<Error2>(
self,
f: impl 'static + Clone + Send + Sync + Fn(Error) -> Error2,
) -> Api<State, Error2, VER>
where
Error: 'static + Send + Sync,
Error2: 'static,
State: 'static + Send + Sync,
{
Api {
inner: ApiInner {
meta: self.inner.meta,
name: self.inner.name,
routes: self
.inner
.routes
.into_iter()
.map(|(name, route)| (name, route.map_err(f.clone())))
.collect(),
routes_by_path: self.inner.routes_by_path,
health_check: self.inner.health_check,
api_version: self.inner.api_version,
error_handler: None,
version_handler: self.inner.version_handler,
public: self.inner.public,
short_description: self.inner.short_description,
long_description: self.inner.long_description,
},
_version: Default::default(),
}
}
pub(crate) fn into_inner(mut self) -> ApiInner<State, Error>
where
Error: crate::Error,
{
self.inner.error_handler = Some(error_handler::<Error, VER>());
self.inner
}
fn default_health_check(req: RequestParams, _state: &State) -> BoxFuture<'_, tide::Response> {
async move {
route::health_check_response::<_, VER>(
&req.accept().unwrap_or_else(|_| {
let mut accept = Accept::new();
accept.set_wildcard(true);
accept
}),
HealthStatus::Available,
)
}
.boxed()
}
}
struct ReadHandler<F, VER> {
handler: F,
_version: PhantomData<VER>,
}
impl<F, VER> From<F> for ReadHandler<F, VER> {
fn from(f: F) -> Self {
Self {
handler: f,
_version: Default::default(),
}
}
}
#[async_trait]
impl<State, Error, F, R, VER> Handler<State, Error> for ReadHandler<F, VER>
where
F: 'static
+ Send
+ Sync
+ Fn(RequestParams, &<State as ReadState>::State) -> BoxFuture<'_, Result<R, Error>>,
R: Serialize,
State: 'static + Send + Sync + ReadState,
VER: 'static + Send + Sync + StaticVersionType,
{
async fn handle(
&self,
req: RequestParams,
state: &State,
) -> Result<tide::Response, RouteError<Error>> {
let accept = req.accept()?;
response_from_result(
&accept,
state.read(|state| (self.handler)(req, state)).await,
VER::instance(),
)
}
}
struct WriteHandler<F, VER> {
handler: F,
_version: PhantomData<VER>,
}
impl<F, VER> From<F> for WriteHandler<F, VER> {
fn from(f: F) -> Self {
Self {
handler: f,
_version: Default::default(),
}
}
}
#[async_trait]
impl<State, Error, F, R, VER> Handler<State, Error> for WriteHandler<F, VER>
where
F: 'static
+ Send
+ Sync
+ Fn(RequestParams, &mut <State as ReadState>::State) -> BoxFuture<'_, Result<R, Error>>,
R: Serialize,
State: 'static + Send + Sync + WriteState,
VER: 'static + Send + Sync + StaticVersionType,
{
async fn handle(
&self,
req: RequestParams,
state: &State,
) -> Result<tide::Response, RouteError<Error>> {
let accept = req.accept()?;
response_from_result(
&accept,
state.write(|state| (self.handler)(req, state)).await,
VER::instance(),
)
}
}
#[cfg(test)]
mod test {
use crate::{
App, StatusCode, Url,
error::{Error, ServerError},
healthcheck::HealthStatus,
socket::Connection,
testing::{Client, setup_test, test_ws_client, test_ws_client_with_headers},
};
use async_std::{sync::RwLock, task::spawn};
use async_tungstenite::{
WebSocketStream,
tungstenite::{http::header::*, protocol::Message, protocol::frame::coding::CloseCode},
};
use futures::{
AsyncRead, AsyncWrite, FutureExt, SinkExt, StreamExt,
stream::{iter, once, repeat},
};
use portpicker::pick_unused_port;
use prometheus::{Counter, Registry};
use std::borrow::Cow;
use toml::toml;
use vbs::{
BinarySerializer, Serializer,
version::{StaticVersion, StaticVersionType},
};
#[cfg(windows)]
use async_tungstenite::tungstenite::Error as WsError;
#[cfg(windows)]
use std::io::ErrorKind;
type StaticVer01 = StaticVersion<0, 1>;
type SerializerV01 = Serializer<StaticVersion<0, 1>>;
async fn check_stream_closed<S>(mut conn: WebSocketStream<S>)
where
S: AsyncRead + AsyncWrite + Unpin,
{
let msg = conn.next().await;
#[cfg(not(windows))]
assert!(msg.is_none(), "{:?}", msg);
#[cfg(windows)]
match msg {
None => {}
Some(Err(WsError::Io(err))) if err.kind() == ErrorKind::ConnectionAborted => {}
msg => panic!(
"expected end of stream or ConnectionAborted error, got {:?}",
msg
),
}
}
#[async_std::test]
async fn test_socket_endpoint() {
setup_test();
let mut app = App::<_, ServerError>::with_state(RwLock::new(()));
let api_toml = toml! {
[meta]
FORMAT_VERSION = "0.1.0"
[route.echo]
PATH = ["/echo"]
METHOD = "SOCKET"
[route.once]
PATH = ["/once"]
METHOD = "SOCKET"
[route.error]
PATH = ["/error"]
METHOD = "SOCKET"
};
{
let mut api = app
.module::<ServerError, StaticVer01>("mod", api_toml)
.unwrap();
api.socket(
"echo",
|_req, mut conn: Connection<String, String, _, StaticVer01>, _state| {
async move {
while let Some(msg) = conn.next().await {
conn.send(&msg?).await?;
}
Ok(())
}
.boxed()
},
)
.unwrap()
.socket(
"once",
|_req, mut conn: Connection<str, (), _, StaticVer01>, _state| {
async move {
conn.send("msg").boxed().await?;
Ok(())
}
.boxed()
},
)
.unwrap()
.socket(
"error",
|_req, _conn: Connection<(), (), _, StaticVer01>, _state| {
async move {
Err(ServerError::catch_all(
StatusCode::INTERNAL_SERVER_ERROR,
"an error message".to_string(),
))
}
.boxed()
},
)
.unwrap();
}
let port = pick_unused_port().unwrap();
let url: Url = format!("http://localhost:{}", port).parse().unwrap();
spawn(app.serve(format!("0.0.0.0:{}", port), StaticVer01::instance()));
let mut conn = test_ws_client_with_headers(
url.join("mod/echo").unwrap(),
&[(ACCEPT, "application/json")],
)
.await;
conn.send(Message::Text(serde_json::to_string("hello").unwrap()))
.await
.unwrap();
assert_eq!(
conn.next().await.unwrap().unwrap(),
Message::Text(serde_json::to_string("hello").unwrap())
);
conn.send(Message::Binary(
SerializerV01::serialize("goodbye").unwrap(),
))
.await
.unwrap();
assert_eq!(
conn.next().await.unwrap().unwrap(),
Message::Text(serde_json::to_string("goodbye").unwrap())
);
let mut conn = test_ws_client_with_headers(
url.join("mod/echo").unwrap(),
&[(ACCEPT, "application/octet-stream")],
)
.await;
conn.send(Message::Text(serde_json::to_string("hello").unwrap()))
.await
.unwrap();
assert_eq!(
conn.next().await.unwrap().unwrap(),
Message::Binary(SerializerV01::serialize("hello").unwrap())
);
conn.send(Message::Binary(
SerializerV01::serialize("goodbye").unwrap(),
))
.await
.unwrap();
assert_eq!(
conn.next().await.unwrap().unwrap(),
Message::Binary(SerializerV01::serialize("goodbye").unwrap())
);
let mut conn = test_ws_client(url.join("mod/once").unwrap()).await;
assert_eq!(
conn.next().await.unwrap().unwrap(),
Message::Text(serde_json::to_string("msg").unwrap())
);
match conn.next().await.unwrap().unwrap() {
Message::Close(None) => {}
msg => panic!("expected normal close frame, got {:?}", msg),
};
check_stream_closed(conn).await;
let mut conn = test_ws_client(url.join("mod/error").unwrap()).await;
match conn.next().await.unwrap().unwrap() {
Message::Close(Some(frame)) => {
assert_eq!(frame.code, CloseCode::Error);
assert_eq!(frame.reason, "Error 500: an error message");
}
msg => panic!("expected error close frame, got {:?}", msg),
}
check_stream_closed(conn).await;
}
#[async_std::test]
async fn test_stream_endpoint() {
setup_test();
let mut app = App::<_, ServerError>::with_state(RwLock::new(()));
let api_toml = toml! {
[meta]
FORMAT_VERSION = "0.1.0"
[route.nat]
PATH = ["/nat"]
METHOD = "SOCKET"
[route.once]
PATH = ["/once"]
METHOD = "SOCKET"
[route.error]
PATH = ["/error"]
METHOD = "SOCKET"
};
{
let mut api = app
.module::<ServerError, StaticVer01>("mod", api_toml)
.unwrap();
api.stream("nat", |_req, _state| iter(0..).map(Ok).boxed())
.unwrap()
.stream("once", |_req, _state| once(async { Ok(0) }).boxed())
.unwrap()
.stream::<_, ()>("error", |_req, _state| {
repeat(Err(ServerError::catch_all(
StatusCode::INTERNAL_SERVER_ERROR,
"an error message".to_string(),
)))
.boxed()
})
.unwrap();
}
let port = pick_unused_port().unwrap();
let url: Url = format!("http://localhost:{}", port).parse().unwrap();
spawn(app.serve(format!("0.0.0.0:{}", port), StaticVer01::instance()));
let mut conn = test_ws_client(url.join("mod/nat").unwrap()).await;
for i in 0..100 {
assert_eq!(
conn.next().await.unwrap().unwrap(),
Message::Text(serde_json::to_string(&i).unwrap())
);
}
let mut conn = test_ws_client(url.join("mod/once").unwrap()).await;
assert_eq!(
conn.next().await.unwrap().unwrap(),
Message::Text(serde_json::to_string(&0).unwrap())
);
match conn.next().await.unwrap().unwrap() {
Message::Close(None) => {}
msg => panic!("expected normal close frame, got {:?}", msg),
}
check_stream_closed(conn).await;
let mut conn = test_ws_client(url.join("mod/error").unwrap()).await;
match conn.next().await.unwrap().unwrap() {
Message::Close(Some(frame)) => {
assert_eq!(frame.code, CloseCode::Error);
assert_eq!(frame.reason, "Error 500: an error message");
}
msg => panic!("expected error close frame, got {:?}", msg),
}
check_stream_closed(conn).await;
}
#[async_std::test]
async fn test_custom_healthcheck() {
setup_test();
let mut app = App::<_, ServerError>::with_state(HealthStatus::Available);
let api_toml = toml! {
[meta]
FORMAT_VERSION = "0.1.0"
[route.dummy]
PATH = ["/dummy"]
};
{
let mut api = app
.module::<ServerError, StaticVer01>("mod", api_toml)
.unwrap();
api.with_health_check(|state| async move { *state }.boxed());
}
let port = pick_unused_port().unwrap();
let url: Url = format!("http://localhost:{}", port).parse().unwrap();
spawn(app.serve(format!("0.0.0.0:{}", port), StaticVer01::instance()));
let client = Client::new(url).await;
let res = client.get("/mod/healthcheck").send().await.unwrap();
assert_eq!(res.status(), StatusCode::OK);
assert_eq!(
res.json::<HealthStatus>().await.unwrap(),
HealthStatus::Available
);
}
#[async_std::test]
async fn test_metrics_endpoint() {
setup_test();
struct State {
metrics: Registry,
counter: Counter,
}
let counter = Counter::new(
"counter",
"count of how many times metrics have been exported",
)
.unwrap();
let metrics = Registry::new();
metrics.register(Box::new(counter.clone())).unwrap();
let state = State { metrics, counter };
let mut app = App::<_, ServerError>::with_state(RwLock::new(state));
let api_toml = toml! {
[meta]
FORMAT_VERSION = "0.1.0"
[route.metrics]
PATH = ["/metrics"]
METHOD = "METRICS"
};
{
let mut api = app
.module::<ServerError, StaticVer01>("mod", api_toml)
.unwrap();
api.metrics("metrics", |_req, state| {
async move {
state.counter.inc();
Ok(Cow::Borrowed(&state.metrics))
}
.boxed()
})
.unwrap();
}
let port = pick_unused_port().unwrap();
let url: Url = format!("http://localhost:{port}").parse().unwrap();
spawn(app.serve(format!("0.0.0.0:{port}"), StaticVer01::instance()));
let client = Client::new(url).await;
for i in 1..5 {
tracing::info!("making metrics request {i}");
let expected = format!(
"# HELP counter count of how many times metrics have been exported\n# TYPE counter counter\ncounter {i}\n"
);
let res = client.get("mod/metrics").send().await.unwrap();
assert_eq!(res.status(), StatusCode::OK);
assert_eq!(res.text().await.unwrap(), expected);
}
}
}