use fastwebsockets::{Frame, Payload, WebSocketError, handshake, upgrade};
use fastwebsockets_stream::{PayloadType, WebSocketStream};
use http_body_util::Empty;
use hyper::Request;
use hyper::Response;
use hyper::body::Bytes;
use hyper::body::Incoming;
use hyper::header::CONNECTION;
use hyper::header::UPGRADE;
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper_util::rt::TokioIo;
use std::future::Future;
use std::io::ErrorKind;
use std::net::Ipv4Addr;
use tokio::io::AsyncReadExt;
use tokio::net::TcpStream;
struct SpawnExecutor;
impl<F> hyper::rt::Executor<F> for SpawnExecutor
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
fn execute(&self, fut: F) {
tokio::task::spawn(fut);
}
}
#[tokio::test]
async fn payload_mismatch() {
let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0u16))
.await
.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move {
loop {
let (stream, _) = listener.accept().await.unwrap();
let io = TokioIo::new(stream);
tokio::spawn(async move {
if let Err(err) = http1::Builder::new()
.serve_connection(io, service_fn(handle))
.with_upgrades()
.await
{
println!("Error serving connection: {:?}", err);
}
});
}
});
let stream = TcpStream::connect(addr).await.unwrap();
let request = Request::builder()
.method("GET")
.uri("ws://localhost")
.header("Host", "localhost")
.header(UPGRADE, "websocket")
.header(CONNECTION, "upgrade")
.header("Sec-WebSocket-Key", handshake::generate_key())
.header("Sec-WebSocket-Version", "13")
.body(Empty::<Bytes>::new())
.unwrap();
let (mut ws, _response) = handshake::client(&SpawnExecutor, request, stream)
.await
.unwrap();
ws.write_frame(Frame::text(Payload::Borrowed(b"Hello!")))
.await
.unwrap();
let _ = ws.read_frame().await;
}
async fn handle(mut request: Request<Incoming>) -> Result<Response<Empty<Bytes>>, WebSocketError> {
assert!(upgrade::is_upgrade_request(&request));
let (response, ws_fut) = upgrade::upgrade(&mut request)?;
tokio::spawn(async move {
let mut buf = [0u8; 1024];
let mut ws_stream = WebSocketStream::new(ws_fut.await.unwrap(), PayloadType::Binary);
let result = ws_stream.read(&mut buf).await;
assert!(result.is_err());
let err = result.unwrap_err();
assert_eq!(err.kind(), ErrorKind::Other);
assert_eq!(
err.to_string(),
"The received data type is different from the stream data type".to_string()
)
});
Ok(response)
}