mirror of
https://github.com/fafhrd91/actix-web
synced 2024-11-24 00:21:08 +01:00
migrate actix-web-actors
This commit is contained in:
parent
a791aab418
commit
a153374b61
@ -94,7 +94,7 @@ open-ssl = { version="0.10", package = "openssl", optional = true }
|
|||||||
rust-tls = { version = "0.16.0", package = "rustls", optional = true }
|
rust-tls = { version = "0.16.0", package = "rustls", optional = true }
|
||||||
|
|
||||||
[dev-dependencies]
|
[dev-dependencies]
|
||||||
# actix = "0.8.3"
|
actix = "0.9.0-alpha.1"
|
||||||
rand = "0.7"
|
rand = "0.7"
|
||||||
env_logger = "0.6"
|
env_logger = "0.6"
|
||||||
serde_derive = "1.0"
|
serde_derive = "1.0"
|
||||||
|
@ -1,5 +1,9 @@
|
|||||||
# Changes
|
# Changes
|
||||||
|
|
||||||
|
## [2.0.0-alpha.1] - 2019-12-15
|
||||||
|
|
||||||
|
* Migrate to actix-web 2.0.0
|
||||||
|
|
||||||
## [1.0.4] - 2019-12-07
|
## [1.0.4] - 2019-12-07
|
||||||
|
|
||||||
* Allow comma-separated websocket subprotocols without spaces (#1172)
|
* Allow comma-separated websocket subprotocols without spaces (#1172)
|
||||||
|
@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "actix-web-actors"
|
name = "actix-web-actors"
|
||||||
version = "1.0.4"
|
version = "2.0.0-alpha.1"
|
||||||
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
||||||
description = "Actix actors support for actix web framework."
|
description = "Actix actors support for actix web framework."
|
||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
@ -9,8 +9,6 @@ homepage = "https://actix.rs"
|
|||||||
repository = "https://github.com/actix/actix-web.git"
|
repository = "https://github.com/actix/actix-web.git"
|
||||||
documentation = "https://docs.rs/actix-web-actors/"
|
documentation = "https://docs.rs/actix-web-actors/"
|
||||||
license = "MIT/Apache-2.0"
|
license = "MIT/Apache-2.0"
|
||||||
exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
|
|
||||||
workspace = ".."
|
|
||||||
edition = "2018"
|
edition = "2018"
|
||||||
|
|
||||||
[lib]
|
[lib]
|
||||||
@ -18,13 +16,14 @@ name = "actix_web_actors"
|
|||||||
path = "src/lib.rs"
|
path = "src/lib.rs"
|
||||||
|
|
||||||
[dependencies]
|
[dependencies]
|
||||||
actix = "0.8.3"
|
actix = "0.9.0-alpha.1"
|
||||||
actix-web = "1.0.9"
|
actix-web = "2.0.0-alpha.5"
|
||||||
actix-http = "0.2.11"
|
actix-http = "1.0.0"
|
||||||
actix-codec = "0.1.2"
|
actix-codec = "0.2.0"
|
||||||
bytes = "0.4"
|
bytes = "0.5.2"
|
||||||
futures = "0.1.25"
|
futures = "0.3.1"
|
||||||
|
pin-project = "0.4.6"
|
||||||
|
|
||||||
[dev-dependencies]
|
[dev-dependencies]
|
||||||
|
actix-rt = "1.0.0"
|
||||||
env_logger = "0.6"
|
env_logger = "0.6"
|
||||||
actix-http-test = { version = "0.2.4", features=["ssl"] }
|
|
||||||
|
@ -1,4 +1,6 @@
|
|||||||
use std::collections::VecDeque;
|
use std::collections::VecDeque;
|
||||||
|
use std::pin::Pin;
|
||||||
|
use std::task::{Context, Poll};
|
||||||
|
|
||||||
use actix::dev::{
|
use actix::dev::{
|
||||||
AsyncContextParts, ContextFut, ContextParts, Envelope, Mailbox, ToEnvelope,
|
AsyncContextParts, ContextFut, ContextParts, Envelope, Mailbox, ToEnvelope,
|
||||||
@ -7,10 +9,10 @@ use actix::fut::ActorFuture;
|
|||||||
use actix::{
|
use actix::{
|
||||||
Actor, ActorContext, ActorState, Addr, AsyncContext, Handler, Message, SpawnHandle,
|
Actor, ActorContext, ActorState, Addr, AsyncContext, Handler, Message, SpawnHandle,
|
||||||
};
|
};
|
||||||
use actix_web::error::{Error, ErrorInternalServerError};
|
use actix_web::error::Error;
|
||||||
use bytes::Bytes;
|
use bytes::Bytes;
|
||||||
use futures::sync::oneshot::Sender;
|
use futures::channel::oneshot::Sender;
|
||||||
use futures::{Async, Future, Poll, Stream};
|
use futures::{Future, Stream};
|
||||||
|
|
||||||
/// Execution context for http actors
|
/// Execution context for http actors
|
||||||
pub struct HttpContext<A>
|
pub struct HttpContext<A>
|
||||||
@ -43,7 +45,7 @@ where
|
|||||||
#[inline]
|
#[inline]
|
||||||
fn spawn<F>(&mut self, fut: F) -> SpawnHandle
|
fn spawn<F>(&mut self, fut: F) -> SpawnHandle
|
||||||
where
|
where
|
||||||
F: ActorFuture<Item = (), Error = (), Actor = A> + 'static,
|
F: ActorFuture<Output = (), Actor = A> + 'static,
|
||||||
{
|
{
|
||||||
self.inner.spawn(fut)
|
self.inner.spawn(fut)
|
||||||
}
|
}
|
||||||
@ -51,7 +53,7 @@ where
|
|||||||
#[inline]
|
#[inline]
|
||||||
fn wait<F>(&mut self, fut: F)
|
fn wait<F>(&mut self, fut: F)
|
||||||
where
|
where
|
||||||
F: ActorFuture<Item = (), Error = (), Actor = A> + 'static,
|
F: ActorFuture<Output = (), Actor = A> + 'static,
|
||||||
{
|
{
|
||||||
self.inner.wait(fut)
|
self.inner.wait(fut)
|
||||||
}
|
}
|
||||||
@ -81,7 +83,7 @@ where
|
|||||||
{
|
{
|
||||||
#[inline]
|
#[inline]
|
||||||
/// Create a new HTTP Context from a request and an actor
|
/// Create a new HTTP Context from a request and an actor
|
||||||
pub fn create(actor: A) -> impl Stream<Item = Bytes, Error = Error> {
|
pub fn create(actor: A) -> impl Stream<Item = Result<Bytes, Error>> {
|
||||||
let mb = Mailbox::default();
|
let mb = Mailbox::default();
|
||||||
let ctx = HttpContext {
|
let ctx = HttpContext {
|
||||||
inner: ContextParts::new(mb.sender_producer()),
|
inner: ContextParts::new(mb.sender_producer()),
|
||||||
@ -91,7 +93,7 @@ where
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Create a new HTTP Context
|
/// Create a new HTTP Context
|
||||||
pub fn with_factory<F>(f: F) -> impl Stream<Item = Bytes, Error = Error>
|
pub fn with_factory<F>(f: F) -> impl Stream<Item = Result<Bytes, Error>>
|
||||||
where
|
where
|
||||||
F: FnOnce(&mut Self) -> A + 'static,
|
F: FnOnce(&mut Self) -> A + 'static,
|
||||||
{
|
{
|
||||||
@ -160,24 +162,23 @@ impl<A> Stream for HttpContextFut<A>
|
|||||||
where
|
where
|
||||||
A: Actor<Context = HttpContext<A>>,
|
A: Actor<Context = HttpContext<A>>,
|
||||||
{
|
{
|
||||||
type Item = Bytes;
|
type Item = Result<Bytes, Error>;
|
||||||
type Error = Error;
|
|
||||||
|
|
||||||
fn poll(&mut self) -> Poll<Option<Bytes>, Error> {
|
fn poll_next(
|
||||||
|
mut self: Pin<&mut Self>,
|
||||||
|
cx: &mut Context<'_>,
|
||||||
|
) -> Poll<Option<Self::Item>> {
|
||||||
if self.fut.alive() {
|
if self.fut.alive() {
|
||||||
match self.fut.poll() {
|
let _ = Pin::new(&mut self.fut).poll(cx);
|
||||||
Ok(Async::NotReady) | Ok(Async::Ready(())) => (),
|
|
||||||
Err(_) => return Err(ErrorInternalServerError("error")),
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// frames
|
// frames
|
||||||
if let Some(data) = self.fut.ctx().stream.pop_front() {
|
if let Some(data) = self.fut.ctx().stream.pop_front() {
|
||||||
Ok(Async::Ready(data))
|
Poll::Ready(data.map(|b| Ok(b)))
|
||||||
} else if self.fut.alive() {
|
} else if self.fut.alive() {
|
||||||
Ok(Async::NotReady)
|
Poll::Pending
|
||||||
} else {
|
} else {
|
||||||
Ok(Async::Ready(None))
|
Poll::Ready(None)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -199,9 +200,9 @@ mod tests {
|
|||||||
|
|
||||||
use actix::Actor;
|
use actix::Actor;
|
||||||
use actix_web::http::StatusCode;
|
use actix_web::http::StatusCode;
|
||||||
use actix_web::test::{block_on, call_service, init_service, TestRequest};
|
use actix_web::test::{call_service, init_service, read_body, TestRequest};
|
||||||
use actix_web::{web, App, HttpResponse};
|
use actix_web::{web, App, HttpResponse};
|
||||||
use bytes::{Bytes, BytesMut};
|
use bytes::Bytes;
|
||||||
|
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|
||||||
@ -223,31 +224,25 @@ mod tests {
|
|||||||
if self.count > 3 {
|
if self.count > 3 {
|
||||||
ctx.write_eof()
|
ctx.write_eof()
|
||||||
} else {
|
} else {
|
||||||
ctx.write(Bytes::from(format!("LINE-{}", self.count).as_bytes()));
|
ctx.write(Bytes::from(format!("LINE-{}", self.count)));
|
||||||
ctx.run_later(Duration::from_millis(100), |slf, ctx| slf.write(ctx));
|
ctx.run_later(Duration::from_millis(100), |slf, ctx| slf.write(ctx));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[actix_rt::test]
|
||||||
fn test_default_resource() {
|
async fn test_default_resource() {
|
||||||
let mut srv =
|
let mut srv =
|
||||||
init_service(App::new().service(web::resource("/test").to(|| {
|
init_service(App::new().service(web::resource("/test").to(|| {
|
||||||
HttpResponse::Ok().streaming(HttpContext::create(MyActor { count: 0 }))
|
HttpResponse::Ok().streaming(HttpContext::create(MyActor { count: 0 }))
|
||||||
})));
|
})))
|
||||||
|
.await;
|
||||||
|
|
||||||
let req = TestRequest::with_uri("/test").to_request();
|
let req = TestRequest::with_uri("/test").to_request();
|
||||||
let mut resp = call_service(&mut srv, req);
|
let resp = call_service(&mut srv, req).await;
|
||||||
assert_eq!(resp.status(), StatusCode::OK);
|
assert_eq!(resp.status(), StatusCode::OK);
|
||||||
|
|
||||||
let body = block_on(resp.take_body().fold(
|
let body = read_body(resp).await;
|
||||||
BytesMut::new(),
|
assert_eq!(body, Bytes::from_static(b"LINE-1LINE-2LINE-3"));
|
||||||
move |mut body, chunk| {
|
|
||||||
body.extend_from_slice(&chunk);
|
|
||||||
Ok::<_, Error>(body)
|
|
||||||
},
|
|
||||||
))
|
|
||||||
.unwrap();
|
|
||||||
assert_eq!(body.freeze(), Bytes::from_static(b"LINE-1LINE-2LINE-3"));
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -1,6 +1,8 @@
|
|||||||
//! Websocket integration
|
//! Websocket integration
|
||||||
use std::collections::VecDeque;
|
use std::collections::VecDeque;
|
||||||
use std::io;
|
use std::io;
|
||||||
|
use std::pin::Pin;
|
||||||
|
use std::task::{Context, Poll};
|
||||||
|
|
||||||
use actix::dev::{
|
use actix::dev::{
|
||||||
AsyncContextParts, ContextFut, ContextParts, Envelope, Mailbox, StreamHandler,
|
AsyncContextParts, ContextFut, ContextParts, Envelope, Mailbox, StreamHandler,
|
||||||
@ -16,20 +18,20 @@ use actix_http::ws::{hash_key, Codec};
|
|||||||
pub use actix_http::ws::{
|
pub use actix_http::ws::{
|
||||||
CloseCode, CloseReason, Frame, HandshakeError, Message, ProtocolError,
|
CloseCode, CloseReason, Frame, HandshakeError, Message, ProtocolError,
|
||||||
};
|
};
|
||||||
|
|
||||||
use actix_web::dev::HttpResponseBuilder;
|
use actix_web::dev::HttpResponseBuilder;
|
||||||
use actix_web::error::{Error, ErrorInternalServerError, PayloadError};
|
use actix_web::error::{Error, PayloadError};
|
||||||
use actix_web::http::{header, Method, StatusCode};
|
use actix_web::http::{header, Method, StatusCode};
|
||||||
use actix_web::{HttpRequest, HttpResponse};
|
use actix_web::{HttpRequest, HttpResponse};
|
||||||
use bytes::{Bytes, BytesMut};
|
use bytes::{Bytes, BytesMut};
|
||||||
use futures::sync::oneshot::Sender;
|
use futures::channel::oneshot::Sender;
|
||||||
use futures::{Async, Future, Poll, Stream};
|
use futures::{Future, Stream};
|
||||||
|
|
||||||
/// Do websocket handshake and start ws actor.
|
/// Do websocket handshake and start ws actor.
|
||||||
pub fn start<A, T>(actor: A, req: &HttpRequest, stream: T) -> Result<HttpResponse, Error>
|
pub fn start<A, T>(actor: A, req: &HttpRequest, stream: T) -> Result<HttpResponse, Error>
|
||||||
where
|
where
|
||||||
A: Actor<Context = WebsocketContext<A>> + StreamHandler<Message, ProtocolError>,
|
A: Actor<Context = WebsocketContext<A>>
|
||||||
T: Stream<Item = Bytes, Error = PayloadError> + 'static,
|
+ StreamHandler<Result<Message, ProtocolError>>,
|
||||||
|
T: Stream<Item = Result<Bytes, PayloadError>> + 'static,
|
||||||
{
|
{
|
||||||
let mut res = handshake(req)?;
|
let mut res = handshake(req)?;
|
||||||
Ok(res.streaming(WebsocketContext::create(actor, stream)))
|
Ok(res.streaming(WebsocketContext::create(actor, stream)))
|
||||||
@ -52,8 +54,9 @@ pub fn start_with_addr<A, T>(
|
|||||||
stream: T,
|
stream: T,
|
||||||
) -> Result<(Addr<A>, HttpResponse), Error>
|
) -> Result<(Addr<A>, HttpResponse), Error>
|
||||||
where
|
where
|
||||||
A: Actor<Context = WebsocketContext<A>> + StreamHandler<Message, ProtocolError>,
|
A: Actor<Context = WebsocketContext<A>>
|
||||||
T: Stream<Item = Bytes, Error = PayloadError> + 'static,
|
+ StreamHandler<Result<Message, ProtocolError>>,
|
||||||
|
T: Stream<Item = Result<Bytes, PayloadError>> + 'static,
|
||||||
{
|
{
|
||||||
let mut res = handshake(req)?;
|
let mut res = handshake(req)?;
|
||||||
let (addr, out_stream) = WebsocketContext::create_with_addr(actor, stream);
|
let (addr, out_stream) = WebsocketContext::create_with_addr(actor, stream);
|
||||||
@ -70,8 +73,9 @@ pub fn start_with_protocols<A, T>(
|
|||||||
stream: T,
|
stream: T,
|
||||||
) -> Result<HttpResponse, Error>
|
) -> Result<HttpResponse, Error>
|
||||||
where
|
where
|
||||||
A: Actor<Context = WebsocketContext<A>> + StreamHandler<Message, ProtocolError>,
|
A: Actor<Context = WebsocketContext<A>>
|
||||||
T: Stream<Item = Bytes, Error = PayloadError> + 'static,
|
+ StreamHandler<Result<Message, ProtocolError>>,
|
||||||
|
T: Stream<Item = Result<Bytes, PayloadError>> + 'static,
|
||||||
{
|
{
|
||||||
let mut res = handshake_with_protocols(req, protocols)?;
|
let mut res = handshake_with_protocols(req, protocols)?;
|
||||||
Ok(res.streaming(WebsocketContext::create(actor, stream)))
|
Ok(res.streaming(WebsocketContext::create(actor, stream)))
|
||||||
@ -202,14 +206,14 @@ where
|
|||||||
{
|
{
|
||||||
fn spawn<F>(&mut self, fut: F) -> SpawnHandle
|
fn spawn<F>(&mut self, fut: F) -> SpawnHandle
|
||||||
where
|
where
|
||||||
F: ActorFuture<Item = (), Error = (), Actor = A> + 'static,
|
F: ActorFuture<Output = (), Actor = A> + 'static,
|
||||||
{
|
{
|
||||||
self.inner.spawn(fut)
|
self.inner.spawn(fut)
|
||||||
}
|
}
|
||||||
|
|
||||||
fn wait<F>(&mut self, fut: F)
|
fn wait<F>(&mut self, fut: F)
|
||||||
where
|
where
|
||||||
F: ActorFuture<Item = (), Error = (), Actor = A> + 'static,
|
F: ActorFuture<Output = (), Actor = A> + 'static,
|
||||||
{
|
{
|
||||||
self.inner.wait(fut)
|
self.inner.wait(fut)
|
||||||
}
|
}
|
||||||
@ -238,10 +242,10 @@ where
|
|||||||
{
|
{
|
||||||
#[inline]
|
#[inline]
|
||||||
/// Create a new Websocket context from a request and an actor
|
/// Create a new Websocket context from a request and an actor
|
||||||
pub fn create<S>(actor: A, stream: S) -> impl Stream<Item = Bytes, Error = Error>
|
pub fn create<S>(actor: A, stream: S) -> impl Stream<Item = Result<Bytes, Error>>
|
||||||
where
|
where
|
||||||
A: StreamHandler<Message, ProtocolError>,
|
A: StreamHandler<Result<Message, ProtocolError>>,
|
||||||
S: Stream<Item = Bytes, Error = PayloadError> + 'static,
|
S: Stream<Item = Result<Bytes, PayloadError>> + 'static,
|
||||||
{
|
{
|
||||||
let (_, stream) = WebsocketContext::create_with_addr(actor, stream);
|
let (_, stream) = WebsocketContext::create_with_addr(actor, stream);
|
||||||
stream
|
stream
|
||||||
@ -256,10 +260,10 @@ where
|
|||||||
pub fn create_with_addr<S>(
|
pub fn create_with_addr<S>(
|
||||||
actor: A,
|
actor: A,
|
||||||
stream: S,
|
stream: S,
|
||||||
) -> (Addr<A>, impl Stream<Item = Bytes, Error = Error>)
|
) -> (Addr<A>, impl Stream<Item = Result<Bytes, Error>>)
|
||||||
where
|
where
|
||||||
A: StreamHandler<Message, ProtocolError>,
|
A: StreamHandler<Result<Message, ProtocolError>>,
|
||||||
S: Stream<Item = Bytes, Error = PayloadError> + 'static,
|
S: Stream<Item = Result<Bytes, PayloadError>> + 'static,
|
||||||
{
|
{
|
||||||
let mb = Mailbox::default();
|
let mb = Mailbox::default();
|
||||||
let mut ctx = WebsocketContext {
|
let mut ctx = WebsocketContext {
|
||||||
@ -279,10 +283,10 @@ where
|
|||||||
actor: A,
|
actor: A,
|
||||||
stream: S,
|
stream: S,
|
||||||
codec: Codec,
|
codec: Codec,
|
||||||
) -> impl Stream<Item = Bytes, Error = Error>
|
) -> impl Stream<Item = Result<Bytes, Error>>
|
||||||
where
|
where
|
||||||
A: StreamHandler<Message, ProtocolError>,
|
A: StreamHandler<Result<Message, ProtocolError>>,
|
||||||
S: Stream<Item = Bytes, Error = PayloadError> + 'static,
|
S: Stream<Item = Result<Bytes, PayloadError>> + 'static,
|
||||||
{
|
{
|
||||||
let mb = Mailbox::default();
|
let mb = Mailbox::default();
|
||||||
let mut ctx = WebsocketContext {
|
let mut ctx = WebsocketContext {
|
||||||
@ -298,11 +302,11 @@ where
|
|||||||
pub fn with_factory<S, F>(
|
pub fn with_factory<S, F>(
|
||||||
stream: S,
|
stream: S,
|
||||||
f: F,
|
f: F,
|
||||||
) -> impl Stream<Item = Bytes, Error = Error>
|
) -> impl Stream<Item = Result<Bytes, Error>>
|
||||||
where
|
where
|
||||||
F: FnOnce(&mut Self) -> A + 'static,
|
F: FnOnce(&mut Self) -> A + 'static,
|
||||||
A: StreamHandler<Message, ProtocolError>,
|
A: StreamHandler<Result<Message, ProtocolError>>,
|
||||||
S: Stream<Item = Bytes, Error = PayloadError> + 'static,
|
S: Stream<Item = Result<Bytes, PayloadError>> + 'static,
|
||||||
{
|
{
|
||||||
let mb = Mailbox::default();
|
let mb = Mailbox::default();
|
||||||
let mut ctx = WebsocketContext {
|
let mut ctx = WebsocketContext {
|
||||||
@ -346,14 +350,14 @@ where
|
|||||||
|
|
||||||
/// Send ping frame
|
/// Send ping frame
|
||||||
#[inline]
|
#[inline]
|
||||||
pub fn ping(&mut self, message: &str) {
|
pub fn ping(&mut self, message: &[u8]) {
|
||||||
self.write_raw(Message::Ping(message.to_string()));
|
self.write_raw(Message::Ping(Bytes::copy_from_slice(message)));
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Send pong frame
|
/// Send pong frame
|
||||||
#[inline]
|
#[inline]
|
||||||
pub fn pong(&mut self, message: &str) {
|
pub fn pong(&mut self, message: &[u8]) {
|
||||||
self.write_raw(Message::Pong(message.to_string()));
|
self.write_raw(Message::Pong(Bytes::copy_from_slice(message)));
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Send close frame
|
/// Send close frame
|
||||||
@ -415,30 +419,34 @@ impl<A> Stream for WebsocketContextFut<A>
|
|||||||
where
|
where
|
||||||
A: Actor<Context = WebsocketContext<A>>,
|
A: Actor<Context = WebsocketContext<A>>,
|
||||||
{
|
{
|
||||||
type Item = Bytes;
|
type Item = Result<Bytes, Error>;
|
||||||
type Error = Error;
|
|
||||||
|
|
||||||
fn poll(&mut self) -> Poll<Option<Bytes>, Error> {
|
fn poll_next(
|
||||||
if self.fut.alive() && self.fut.poll().is_err() {
|
self: Pin<&mut Self>,
|
||||||
return Err(ErrorInternalServerError("error"));
|
cx: &mut Context<'_>,
|
||||||
|
) -> Poll<Option<Self::Item>> {
|
||||||
|
let this = self.get_mut();
|
||||||
|
|
||||||
|
if this.fut.alive() {
|
||||||
|
let _ = Pin::new(&mut this.fut).poll(cx);
|
||||||
}
|
}
|
||||||
|
|
||||||
// encode messages
|
// encode messages
|
||||||
while let Some(item) = self.fut.ctx().messages.pop_front() {
|
while let Some(item) = this.fut.ctx().messages.pop_front() {
|
||||||
if let Some(msg) = item {
|
if let Some(msg) = item {
|
||||||
self.encoder.encode(msg, &mut self.buf)?;
|
this.encoder.encode(msg, &mut this.buf)?;
|
||||||
} else {
|
} else {
|
||||||
self.closed = true;
|
this.closed = true;
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if !self.buf.is_empty() {
|
if !this.buf.is_empty() {
|
||||||
Ok(Async::Ready(Some(self.buf.take().freeze())))
|
Poll::Ready(Some(Ok(this.buf.split().freeze())))
|
||||||
} else if self.fut.alive() && !self.closed {
|
} else if this.fut.alive() && !this.closed {
|
||||||
Ok(Async::NotReady)
|
Poll::Pending
|
||||||
} else {
|
} else {
|
||||||
Ok(Async::Ready(None))
|
Poll::Ready(None)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -454,7 +462,9 @@ where
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[pin_project::pin_project]
|
||||||
struct WsStream<S> {
|
struct WsStream<S> {
|
||||||
|
#[pin]
|
||||||
stream: S,
|
stream: S,
|
||||||
decoder: Codec,
|
decoder: Codec,
|
||||||
buf: BytesMut,
|
buf: BytesMut,
|
||||||
@ -463,7 +473,7 @@ struct WsStream<S> {
|
|||||||
|
|
||||||
impl<S> WsStream<S>
|
impl<S> WsStream<S>
|
||||||
where
|
where
|
||||||
S: Stream<Item = Bytes, Error = PayloadError>,
|
S: Stream<Item = Result<Bytes, PayloadError>>,
|
||||||
{
|
{
|
||||||
fn new(stream: S, codec: Codec) -> Self {
|
fn new(stream: S, codec: Codec) -> Self {
|
||||||
Self {
|
Self {
|
||||||
@ -477,62 +487,64 @@ where
|
|||||||
|
|
||||||
impl<S> Stream for WsStream<S>
|
impl<S> Stream for WsStream<S>
|
||||||
where
|
where
|
||||||
S: Stream<Item = Bytes, Error = PayloadError>,
|
S: Stream<Item = Result<Bytes, PayloadError>>,
|
||||||
{
|
{
|
||||||
type Item = Message;
|
type Item = Result<Message, ProtocolError>;
|
||||||
type Error = ProtocolError;
|
|
||||||
|
|
||||||
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
|
fn poll_next(
|
||||||
if !self.closed {
|
mut self: Pin<&mut Self>,
|
||||||
|
cx: &mut Context<'_>,
|
||||||
|
) -> Poll<Option<Self::Item>> {
|
||||||
|
let mut this = self.as_mut().project();
|
||||||
|
|
||||||
|
if !*this.closed {
|
||||||
loop {
|
loop {
|
||||||
match self.stream.poll() {
|
this = self.as_mut().project();
|
||||||
Ok(Async::Ready(Some(chunk))) => {
|
match Pin::new(&mut this.stream).poll_next(cx) {
|
||||||
self.buf.extend_from_slice(&chunk[..]);
|
Poll::Ready(Some(Ok(chunk))) => {
|
||||||
|
this.buf.extend_from_slice(&chunk[..]);
|
||||||
}
|
}
|
||||||
Ok(Async::Ready(None)) => {
|
Poll::Ready(None) => {
|
||||||
self.closed = true;
|
*this.closed = true;
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
Ok(Async::NotReady) => break,
|
Poll::Pending => break,
|
||||||
Err(e) => {
|
Poll::Ready(Some(Err(e))) => {
|
||||||
return Err(ProtocolError::Io(io::Error::new(
|
return Poll::Ready(Some(Err(ProtocolError::Io(
|
||||||
io::ErrorKind::Other,
|
io::Error::new(io::ErrorKind::Other, format!("{}", e)),
|
||||||
format!("{}", e),
|
))));
|
||||||
)));
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
match self.decoder.decode(&mut self.buf)? {
|
match this.decoder.decode(this.buf)? {
|
||||||
None => {
|
None => {
|
||||||
if self.closed {
|
if *this.closed {
|
||||||
Ok(Async::Ready(None))
|
Poll::Ready(None)
|
||||||
} else {
|
} else {
|
||||||
Ok(Async::NotReady)
|
Poll::Pending
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
Some(frm) => {
|
Some(frm) => {
|
||||||
let msg = match frm {
|
let msg = match frm {
|
||||||
Frame::Text(data) => {
|
Frame::Text(data) => Message::Text(
|
||||||
if let Some(data) = data {
|
std::str::from_utf8(&data)
|
||||||
Message::Text(
|
.map_err(|e| {
|
||||||
std::str::from_utf8(&data)
|
ProtocolError::Io(io::Error::new(
|
||||||
.map_err(|_| ProtocolError::BadEncoding)?
|
io::ErrorKind::Other,
|
||||||
.to_string(),
|
format!("{}", e),
|
||||||
)
|
))
|
||||||
} else {
|
})?
|
||||||
Message::Text(String::new())
|
.to_string(),
|
||||||
}
|
|
||||||
}
|
|
||||||
Frame::Binary(data) => Message::Binary(
|
|
||||||
data.map(|b| b.freeze()).unwrap_or_else(Bytes::new),
|
|
||||||
),
|
),
|
||||||
|
Frame::Binary(data) => Message::Binary(data),
|
||||||
Frame::Ping(s) => Message::Ping(s),
|
Frame::Ping(s) => Message::Ping(s),
|
||||||
Frame::Pong(s) => Message::Pong(s),
|
Frame::Pong(s) => Message::Pong(s),
|
||||||
Frame::Close(reason) => Message::Close(reason),
|
Frame::Close(reason) => Message::Close(reason),
|
||||||
|
Frame::Continuation(item) => Message::Continuation(item),
|
||||||
};
|
};
|
||||||
Ok(Async::Ready(Some(msg)))
|
Poll::Ready(Some(Ok(msg)))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -1,10 +1,8 @@
|
|||||||
use actix::prelude::*;
|
use actix::prelude::*;
|
||||||
use actix_http::HttpService;
|
use actix_web::{test, web, App, HttpRequest};
|
||||||
use actix_http_test::TestServer;
|
|
||||||
use actix_web::{web, App, HttpRequest};
|
|
||||||
use actix_web_actors::*;
|
use actix_web_actors::*;
|
||||||
use bytes::{Bytes, BytesMut};
|
use bytes::Bytes;
|
||||||
use futures::{Sink, Stream};
|
use futures::{SinkExt, StreamExt};
|
||||||
|
|
||||||
struct Ws;
|
struct Ws;
|
||||||
|
|
||||||
@ -12,9 +10,13 @@ impl Actor for Ws {
|
|||||||
type Context = ws::WebsocketContext<Self>;
|
type Context = ws::WebsocketContext<Self>;
|
||||||
}
|
}
|
||||||
|
|
||||||
impl StreamHandler<ws::Message, ws::ProtocolError> for Ws {
|
impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for Ws {
|
||||||
fn handle(&mut self, msg: ws::Message, ctx: &mut Self::Context) {
|
fn handle(
|
||||||
match msg {
|
&mut self,
|
||||||
|
msg: Result<ws::Message, ws::ProtocolError>,
|
||||||
|
ctx: &mut Self::Context,
|
||||||
|
) {
|
||||||
|
match msg.unwrap() {
|
||||||
ws::Message::Ping(msg) => ctx.pong(&msg),
|
ws::Message::Ping(msg) => ctx.pong(&msg),
|
||||||
ws::Message::Text(text) => ctx.text(text),
|
ws::Message::Text(text) => ctx.text(text),
|
||||||
ws::Message::Binary(bin) => ctx.binary(bin),
|
ws::Message::Binary(bin) => ctx.binary(bin),
|
||||||
@ -24,45 +26,42 @@ impl StreamHandler<ws::Message, ws::ProtocolError> for Ws {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[actix_rt::test]
|
||||||
fn test_simple() {
|
async fn test_simple() {
|
||||||
let mut srv =
|
let mut srv = test::start(|| {
|
||||||
TestServer::new(|| {
|
App::new().service(web::resource("/").to(
|
||||||
HttpService::new(App::new().service(web::resource("/").to(
|
|req: HttpRequest, stream: web::Payload| {
|
||||||
|req: HttpRequest, stream: web::Payload| ws::start(Ws, &req, stream),
|
async move { ws::start(Ws, &req, stream) }
|
||||||
)))
|
},
|
||||||
});
|
))
|
||||||
|
});
|
||||||
|
|
||||||
// client service
|
// client service
|
||||||
let framed = srv.ws().unwrap();
|
let mut framed = srv.ws().await.unwrap();
|
||||||
let framed = srv
|
framed
|
||||||
.block_on(framed.send(ws::Message::Text("text".to_string())))
|
.send(ws::Message::Text("text".to_string()))
|
||||||
.unwrap();
|
.await
|
||||||
let (item, framed) = srv.block_on(framed.into_future()).map_err(|_| ()).unwrap();
|
|
||||||
assert_eq!(item, Some(ws::Frame::Text(Some(BytesMut::from("text")))));
|
|
||||||
|
|
||||||
let framed = srv
|
|
||||||
.block_on(framed.send(ws::Message::Binary("text".into())))
|
|
||||||
.unwrap();
|
|
||||||
let (item, framed) = srv.block_on(framed.into_future()).map_err(|_| ()).unwrap();
|
|
||||||
assert_eq!(
|
|
||||||
item,
|
|
||||||
Some(ws::Frame::Binary(Some(Bytes::from_static(b"text").into())))
|
|
||||||
);
|
|
||||||
|
|
||||||
let framed = srv
|
|
||||||
.block_on(framed.send(ws::Message::Ping("text".into())))
|
|
||||||
.unwrap();
|
|
||||||
let (item, framed) = srv.block_on(framed.into_future()).map_err(|_| ()).unwrap();
|
|
||||||
assert_eq!(item, Some(ws::Frame::Pong("text".to_string().into())));
|
|
||||||
|
|
||||||
let framed = srv
|
|
||||||
.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.next().await.unwrap().unwrap();
|
||||||
assert_eq!(
|
assert_eq!(item, ws::Frame::Text(Bytes::from_static(b"text")));
|
||||||
item,
|
|
||||||
Some(ws::Frame::Close(Some(ws::CloseCode::Normal.into())))
|
framed
|
||||||
);
|
.send(ws::Message::Binary("text".into()))
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
let item = framed.next().await.unwrap().unwrap();
|
||||||
|
assert_eq!(item, ws::Frame::Binary(Bytes::from_static(b"text").into()));
|
||||||
|
|
||||||
|
framed.send(ws::Message::Ping("text".into())).await.unwrap();
|
||||||
|
let item = framed.next().await.unwrap().unwrap();
|
||||||
|
assert_eq!(item, ws::Frame::Pong(Bytes::copy_from_slice(b"text")));
|
||||||
|
|
||||||
|
framed
|
||||||
|
.send(ws::Message::Close(Some(ws::CloseCode::Normal.into())))
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let item = framed.next().await.unwrap().unwrap();
|
||||||
|
assert_eq!(item, ws::Frame::Close(Some(ws::CloseCode::Normal.into())));
|
||||||
}
|
}
|
||||||
|
72
src/test.rs
72
src/test.rs
@ -910,6 +910,7 @@ impl Drop for TestServer {
|
|||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use actix_http::httpmessage::HttpMessage;
|
use actix_http::httpmessage::HttpMessage;
|
||||||
|
use futures::FutureExt;
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
use std::time::SystemTime;
|
use std::time::SystemTime;
|
||||||
|
|
||||||
@ -1095,41 +1096,46 @@ mod tests {
|
|||||||
assert!(res.status().is_success());
|
assert!(res.status().is_success());
|
||||||
}
|
}
|
||||||
|
|
||||||
// #[actix_rt::test]
|
#[actix_rt::test]
|
||||||
// fn test_actor() {
|
async fn test_actor() {
|
||||||
// use actix::Actor;
|
use actix::Actor;
|
||||||
|
|
||||||
// struct MyActor;
|
struct MyActor;
|
||||||
|
|
||||||
// struct Num(usize);
|
struct Num(usize);
|
||||||
// impl actix::Message for Num {
|
impl actix::Message for Num {
|
||||||
// type Result = usize;
|
type Result = usize;
|
||||||
// }
|
}
|
||||||
// impl actix::Actor for MyActor {
|
impl actix::Actor for MyActor {
|
||||||
// type Context = actix::Context<Self>;
|
type Context = actix::Context<Self>;
|
||||||
// }
|
}
|
||||||
// impl actix::Handler<Num> for MyActor {
|
impl actix::Handler<Num> for MyActor {
|
||||||
// type Result = usize;
|
type Result = usize;
|
||||||
// fn handle(&mut self, msg: Num, _: &mut Self::Context) -> Self::Result {
|
fn handle(&mut self, msg: Num, _: &mut Self::Context) -> Self::Result {
|
||||||
// msg.0
|
msg.0
|
||||||
// }
|
}
|
||||||
// }
|
}
|
||||||
|
|
||||||
// let addr = run_on(|| MyActor.start());
|
let addr = MyActor.start();
|
||||||
// let mut app = init_service(App::new().service(
|
|
||||||
// web::resource("/index.html").to(move || {
|
|
||||||
// addr.send(Num(1)).from_err().and_then(|res| {
|
|
||||||
// if res == 1 {
|
|
||||||
// HttpResponse::Ok()
|
|
||||||
// } else {
|
|
||||||
// HttpResponse::BadRequest()
|
|
||||||
// }
|
|
||||||
// })
|
|
||||||
// }),
|
|
||||||
// ));
|
|
||||||
|
|
||||||
// let req = TestRequest::post().uri("/index.html").to_request();
|
let mut app = init_service(App::new().service(web::resource("/index.html").to(
|
||||||
// let res = block_fn(|| app.call(req)).unwrap();
|
move || {
|
||||||
// assert!(res.status().is_success());
|
addr.send(Num(1)).map(|res| match res {
|
||||||
// }
|
Ok(res) => {
|
||||||
|
if res == 1 {
|
||||||
|
Ok(HttpResponse::Ok())
|
||||||
|
} else {
|
||||||
|
Ok(HttpResponse::BadRequest())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Err(err) => Err(err),
|
||||||
|
})
|
||||||
|
},
|
||||||
|
)))
|
||||||
|
.await;
|
||||||
|
|
||||||
|
let req = TestRequest::post().uri("/index.html").to_request();
|
||||||
|
let res = app.call(req).await.unwrap();
|
||||||
|
assert!(res.status().is_success());
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
Loading…
Reference in New Issue
Block a user