273bd1b
use std::time::Duration;
use http_body_util::{BodyExt, Limited};
use hyper::StatusCode;
use hyper::body::Incoming;
use tokio::sync::oneshot;
use uuid::Uuid;
use super::response::{
HttpRequest, HttpResponse, authenticate, engine_error, json_response, text_response,
};
use super::session::{Input, Output};
use super::{SocketHub, engine};
pub(super) async fn polling_open(request: &HttpRequest, hub: &SocketHub) -> HttpResponse {
let (user_id, token) = match authenticate(request, hub).await {
Ok(user_id) => user_id,
Err(response) => return response,
};
let sid = Uuid::new_v4().to_string();
let options = hub.options();
let open = engine::open(
&sid,
options.heartbeat_interval,
options.heartbeat_timeout,
options.max_payload,
&["websocket"],
);
hub.create_session(sid, user_id, token, Output::polling(), false)
.await;
text_response(engine::encode_payload(&[open]))
}
pub(super) async fn polling_poll(hub: &SocketHub, sid: &str) -> HttpResponse {
let Some(input) = hub.input(sid).await else {
return engine_error(1, "Session ID unknown");
};
let (sender, receiver) = oneshot::channel();
if input.send(Input::Poll(sender)).await.is_err() {
return engine_error(1, "Session ID unknown");
}
let options = hub.options();
let wait = options.heartbeat_interval + options.heartbeat_timeout + Duration::from_secs(5);
match tokio::time::timeout(wait, receiver).await {
Ok(Ok(packets)) => {
let payload = if packets.is_empty() {
"6".to_string()
} else {
engine::encode_payload(&packets)
};
text_response(payload)
}
_ => text_response("6".to_string()),
}
}
pub(super) async fn polling_post(
body: Incoming,
hub: &SocketHub,
sid: Option<&str>,
) -> HttpResponse {
let Some(sid) = sid else {
return engine_error(3, "Bad request");
};
let Some(input) = hub.input(sid).await else {
return engine_error(1, "Session ID unknown");
};
let bytes = match Limited::new(body, hub.options().max_payload)
.collect()
.await
{
Ok(body) => body.to_bytes(),
Err(error) if error.is::<http_body_util::LengthLimitError>() => {
return json_response(
StatusCode::PAYLOAD_TOO_LARGE,
serde_json::json!({
"code": 3,
"message": "Payload too large",
}),
);
}
Err(_) => return engine_error(3, "Bad request"),
};
let Ok(text) = std::str::from_utf8(&bytes) else {
return engine_error(3, "Bad request");
};
let Some(packets) = engine::decode_payload(text) else {
return engine_error(3, "Bad request");
};
if input.send(Input::Packets(packets)).await.is_err() {
return engine_error(1, "Session ID unknown");
}
text_response("ok".to_string())
}