mirror of
https://github.com/actix/actix-extras.git
synced 2024-11-24 07:53:00 +01:00
export ws module
This commit is contained in:
parent
fd3e351c31
commit
6ab7665868
@ -1,10 +1,5 @@
|
|||||||
//! Actix actors integration for Actix web framework
|
//! Actix actors integration for Actix web framework
|
||||||
mod context;
|
mod context;
|
||||||
mod ws;
|
pub mod ws;
|
||||||
|
|
||||||
pub use self::context::HttpContext;
|
pub use self::context::HttpContext;
|
||||||
pub use self::ws::{ws_handshake, ws_start, WebsocketContext};
|
|
||||||
|
|
||||||
pub use actix_http::ws::CloseCode as WsCloseCode;
|
|
||||||
pub use actix_http::ws::ProtocolError as WsProtocolError;
|
|
||||||
pub use actix_http::ws::{Frame as WsFrame, Message as WsMessage};
|
|
||||||
|
@ -1,3 +1,4 @@
|
|||||||
|
//! Websocket integration
|
||||||
use std::collections::VecDeque;
|
use std::collections::VecDeque;
|
||||||
use std::io;
|
use std::io;
|
||||||
|
|
||||||
@ -11,9 +12,11 @@ use actix::{
|
|||||||
Message as ActixMessage, SpawnHandle,
|
Message as ActixMessage, SpawnHandle,
|
||||||
};
|
};
|
||||||
use actix_codec::{Decoder, Encoder};
|
use actix_codec::{Decoder, Encoder};
|
||||||
use actix_http::ws::{
|
use actix_http::ws::hash_key;
|
||||||
hash_key, CloseReason, Codec, Frame, HandshakeError, Message, ProtocolError,
|
pub use actix_http::ws::{
|
||||||
|
CloseCode, CloseReason, Codec, Frame, HandshakeError, Message, ProtocolError,
|
||||||
};
|
};
|
||||||
|
|
||||||
use actix_web::dev::{Head, HttpResponseBuilder};
|
use actix_web::dev::{Head, HttpResponseBuilder};
|
||||||
use actix_web::error::{Error, ErrorInternalServerError, PayloadError};
|
use actix_web::error::{Error, ErrorInternalServerError, PayloadError};
|
||||||
use actix_web::http::{header, Method, StatusCode};
|
use actix_web::http::{header, Method, StatusCode};
|
||||||
@ -23,16 +26,12 @@ use futures::sync::oneshot::Sender;
|
|||||||
use futures::{Async, Future, Poll, Stream};
|
use futures::{Async, Future, Poll, Stream};
|
||||||
|
|
||||||
/// Do websocket handshake and start ws actor.
|
/// Do websocket handshake and start ws actor.
|
||||||
pub fn ws_start<A, T>(
|
pub fn start<A, T>(actor: A, req: &HttpRequest, stream: T) -> Result<HttpResponse, Error>
|
||||||
actor: A,
|
|
||||||
req: &HttpRequest,
|
|
||||||
stream: T,
|
|
||||||
) -> Result<HttpResponse, Error>
|
|
||||||
where
|
where
|
||||||
A: Actor<Context = WebsocketContext<A>> + StreamHandler<Frame, ProtocolError>,
|
A: Actor<Context = WebsocketContext<A>> + StreamHandler<Frame, ProtocolError>,
|
||||||
T: Stream<Item = Bytes, Error = PayloadError> + 'static,
|
T: Stream<Item = Bytes, Error = PayloadError> + 'static,
|
||||||
{
|
{
|
||||||
let mut res = ws_handshake(req)?;
|
let mut res = handshake(req)?;
|
||||||
Ok(res.streaming(WebsocketContext::create(actor, stream)))
|
Ok(res.streaming(WebsocketContext::create(actor, stream)))
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -44,7 +43,7 @@ where
|
|||||||
// /// `protocols` is a sequence of known protocols. On successful handshake,
|
// /// `protocols` is a sequence of known protocols. On successful handshake,
|
||||||
// /// the returned response headers contain the first protocol in this list
|
// /// the returned response headers contain the first protocol in this list
|
||||||
// /// which the server also knows.
|
// /// which the server also knows.
|
||||||
pub fn ws_handshake(req: &HttpRequest) -> Result<HttpResponseBuilder, HandshakeError> {
|
pub fn handshake(req: &HttpRequest) -> Result<HttpResponseBuilder, HandshakeError> {
|
||||||
// WebSocket accepts only GET
|
// WebSocket accepts only GET
|
||||||
if *req.method() != Method::GET {
|
if *req.method() != Method::GET {
|
||||||
return Err(HandshakeError::GetMethodRequired);
|
return Err(HandshakeError::GetMethodRequired);
|
||||||
|
@ -9,18 +9,18 @@ use futures::{Sink, Stream};
|
|||||||
struct Ws;
|
struct Ws;
|
||||||
|
|
||||||
impl Actor for Ws {
|
impl Actor for Ws {
|
||||||
type Context = WebsocketContext<Self>;
|
type Context = ws::WebsocketContext<Self>;
|
||||||
}
|
}
|
||||||
|
|
||||||
impl StreamHandler<WsFrame, WsProtocolError> for Ws {
|
impl StreamHandler<ws::Frame, ws::ProtocolError> for Ws {
|
||||||
fn handle(&mut self, msg: WsFrame, ctx: &mut Self::Context) {
|
fn handle(&mut self, msg: ws::Frame, ctx: &mut Self::Context) {
|
||||||
match msg {
|
match msg {
|
||||||
WsFrame::Ping(msg) => ctx.pong(&msg),
|
ws::Frame::Ping(msg) => ctx.pong(&msg),
|
||||||
WsFrame::Text(text) => {
|
ws::Frame::Text(text) => {
|
||||||
ctx.text(String::from_utf8_lossy(&text.unwrap())).to_owned()
|
ctx.text(String::from_utf8_lossy(&text.unwrap())).to_owned()
|
||||||
}
|
}
|
||||||
WsFrame::Binary(bin) => ctx.binary(bin.unwrap()),
|
ws::Frame::Binary(bin) => ctx.binary(bin.unwrap()),
|
||||||
WsFrame::Close(reason) => ctx.close(reason),
|
ws::Frame::Close(reason) => ctx.close(reason),
|
||||||
_ => (),
|
_ => (),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -28,40 +28,42 @@ impl StreamHandler<WsFrame, WsProtocolError> for Ws {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_simple() {
|
fn test_simple() {
|
||||||
let mut srv =
|
let mut srv = TestServer::new(|| {
|
||||||
TestServer::new(|| {
|
|
||||||
HttpService::new(App::new().service(web::resource("/").to(
|
HttpService::new(App::new().service(web::resource("/").to(
|
||||||
|req: HttpRequest, stream: web::Payload<_>| ws_start(Ws, &req, stream),
|
|req: HttpRequest, stream: web::Payload<_>| ws::start(Ws, &req, stream),
|
||||||
)))
|
)))
|
||||||
});
|
});
|
||||||
|
|
||||||
// client service
|
// client service
|
||||||
let framed = srv.ws().unwrap();
|
let framed = srv.ws().unwrap();
|
||||||
let framed = srv
|
let framed = srv
|
||||||
.block_on(framed.send(WsMessage::Text("text".to_string())))
|
.block_on(framed.send(ws::Message::Text("text".to_string())))
|
||||||
.unwrap();
|
.unwrap();
|
||||||
let (item, framed) = srv.block_on(framed.into_future()).map_err(|_| ()).unwrap();
|
let (item, framed) = srv.block_on(framed.into_future()).map_err(|_| ()).unwrap();
|
||||||
assert_eq!(item, Some(WsFrame::Text(Some(BytesMut::from("text")))));
|
assert_eq!(item, Some(ws::Frame::Text(Some(BytesMut::from("text")))));
|
||||||
|
|
||||||
let framed = srv
|
let framed = srv
|
||||||
.block_on(framed.send(WsMessage::Binary("text".into())))
|
.block_on(framed.send(ws::Message::Binary("text".into())))
|
||||||
.unwrap();
|
.unwrap();
|
||||||
let (item, framed) = srv.block_on(framed.into_future()).map_err(|_| ()).unwrap();
|
let (item, framed) = srv.block_on(framed.into_future()).map_err(|_| ()).unwrap();
|
||||||
assert_eq!(
|
assert_eq!(
|
||||||
item,
|
item,
|
||||||
Some(WsFrame::Binary(Some(Bytes::from_static(b"text").into())))
|
Some(ws::Frame::Binary(Some(Bytes::from_static(b"text").into())))
|
||||||
);
|
);
|
||||||
|
|
||||||
let framed = srv
|
let framed = srv
|
||||||
.block_on(framed.send(WsMessage::Ping("text".into())))
|
.block_on(framed.send(ws::Message::Ping("text".into())))
|
||||||
.unwrap();
|
.unwrap();
|
||||||
let (item, framed) = srv.block_on(framed.into_future()).map_err(|_| ()).unwrap();
|
let (item, framed) = srv.block_on(framed.into_future()).map_err(|_| ()).unwrap();
|
||||||
assert_eq!(item, Some(WsFrame::Pong("text".to_string().into())));
|
assert_eq!(item, Some(ws::Frame::Pong("text".to_string().into())));
|
||||||
|
|
||||||
let framed = srv
|
let framed = srv
|
||||||
.block_on(framed.send(WsMessage::Close(Some(WsCloseCode::Normal.into()))))
|
.block_on(framed.send(ws::Message::Close(Some(ws::CloseCode::Normal.into()))))
|
||||||
.unwrap();
|
.unwrap();
|
||||||
|
|
||||||
let (item, _framed) = srv.block_on(framed.into_future()).map_err(|_| ()).unwrap();
|
let (item, _framed) = srv.block_on(framed.into_future()).map_err(|_| ()).unwrap();
|
||||||
assert_eq!(item, Some(WsFrame::Close(Some(WsCloseCode::Normal.into()))));
|
assert_eq!(
|
||||||
|
item,
|
||||||
|
Some(ws::Frame::Close(Some(ws::CloseCode::Normal.into())))
|
||||||
|
);
|
||||||
}
|
}
|
||||||
|
Loading…
Reference in New Issue
Block a user