use actix_http::{
body::{BodyStream, MessageBody},
ws::handshake,
};
use actix_web::{HttpRequest, HttpResponse, web};
use tokio::sync::{mpsc::channel, oneshot};
mod session;
mod stream;
pub(crate) use self::{session::Session, stream::MessageStream};
pub(crate) fn handle(
req: &HttpRequest,
body: web::Payload,
) -> Result<(HttpResponse, Session, MessageStream), actix_web::Error> {
let mut response = handshake(req.head())?;
let (tx, rx) = channel(32);
let (connection_closed_tx, connection_closed_rx) = oneshot::channel();
Ok((
response
.message_body(
BodyStream::new(
stream::StreamingBody::new(rx)
.with_connection_close_signal(connection_closed_tx),
)
.boxed(),
)?
.into(),
Session::new(tx),
MessageStream::new(body.into_inner()).with_connection_close_signal(connection_closed_rx),
))
}