actix/actix-extras
1
0
Fork 0
mirror of https://github.com/actix/actix-extras.git synced 2024-11-24 07:53:00 +01:00
Code Issues Releases Wiki Activity

rename payload

Browse Source
This commit is contained in:
Nikolay Kim 2018-01-10 20:08:13 -08:00
parent e0faf3f69c
commit aed90ed458
4 changed files with 13 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/context.rs
View File

@ -23,7 +23,7 @@ pub trait ActorHttpContext: 'static {
#[derive(Debug)] #[derive(Debug)]
pub enum Frame { pub enum Frame {
Payload(Option<Binary>), Chunk(Option<Binary>),
Drain(oneshot::Sender<()>), Drain(oneshot::Sender<()>),
} }
@ -121,7 +121,7 @@ impl<A, S> HttpContext<A, S> where A: Actor<Context=Self> {
#[inline] #[inline]
pub fn write<B: Into<Binary>>(&mut self, data: B) { pub fn write<B: Into<Binary>>(&mut self, data: B) {
if !self.disconnected { if !self.disconnected {
self.stream.push_back(Frame::Payload(Some(data.into()))); self.stream.push_back(Frame::Chunk(Some(data.into())));
} else { } else {
warn!("Trying to write to disconnected response"); warn!("Trying to write to disconnected response");
} }
@ -130,7 +130,7 @@ impl<A, S> HttpContext<A, S> where A: Actor<Context=Self> {
/// Indicate end of streamimng payload. Also this method calls `Self::close`. /// Indicate end of streamimng payload. Also this method calls `Self::close`.
#[inline] #[inline]
pub fn write_eof(&mut self) { pub fn write_eof(&mut self) {
self.stream.push_back(Frame::Payload(None)); self.stream.push_back(Frame::Chunk(None));
} }
/// Returns drain future /// Returns drain future

22
src/pipeline.rs
View File

@ -209,8 +209,7 @@ struct StartMiddlewares<S, H> {
impl<S: 'static, H: PipelineHandler<S>> StartMiddlewares<S, H> { impl<S: 'static, H: PipelineHandler<S>> StartMiddlewares<S, H> {
fn init(info: &mut PipelineInfo<S>, handler: Rc<RefCell<H>>) -> PipelineState<S, H> fn init(info: &mut PipelineInfo<S>, handler: Rc<RefCell<H>>) -> PipelineState<S, H> {
{
// execute middlewares, we need this stage because middlewares could be non-async // execute middlewares, we need this stage because middlewares could be non-async
// and we can move to next state immidietly // and we can move to next state immidietly
let len = info.mws.len(); let len = info.mws.len();
@ -247,8 +246,7 @@ impl<S: 'static, H: PipelineHandler<S>> StartMiddlewares<S, H> {
} }
} }
fn poll(&mut self, info: &mut PipelineInfo<S>) -> Option<PipelineState<S, H>> fn poll(&mut self, info: &mut PipelineInfo<S>) -> Option<PipelineState<S, H>> {
{
let len = info.mws.len(); let len = info.mws.len();
'outer: loop { 'outer: loop {
match self.fut.as_mut().unwrap().poll() { match self.fut.as_mut().unwrap().poll() {
@ -296,8 +294,7 @@ struct WaitingResponse<S, H> {
impl<S: 'static, H> WaitingResponse<S, H> { impl<S: 'static, H> WaitingResponse<S, H> {
#[inline] #[inline]
fn init(info: &mut PipelineInfo<S>, reply: Reply) -> PipelineState<S, H> fn init(info: &mut PipelineInfo<S>, reply: Reply) -> PipelineState<S, H> {
{
match reply.into() { match reply.into() {
ReplyItem::Message(resp) => ReplyItem::Message(resp) =>
RunMiddlewares::init(info, resp), RunMiddlewares::init(info, resp),
@ -307,8 +304,7 @@ impl<S: 'static, H> WaitingResponse<S, H> {
} }
} }
fn poll(&mut self, info: &mut PipelineInfo<S>) -> Option<PipelineState<S, H>> fn poll(&mut self, info: &mut PipelineInfo<S>) -> Option<PipelineState<S, H>> {
{
match self.fut.poll() { match self.fut.poll() {
Ok(Async::NotReady) => None, Ok(Async::NotReady) => None,
Ok(Async::Ready(response)) => Ok(Async::Ready(response)) =>
@ -329,8 +325,7 @@ struct RunMiddlewares<S, H> {
impl<S: 'static, H> RunMiddlewares<S, H> { impl<S: 'static, H> RunMiddlewares<S, H> {
fn init(info: &mut PipelineInfo<S>, mut resp: HttpResponse) -> PipelineState<S, H> fn init(info: &mut PipelineInfo<S>, mut resp: HttpResponse) -> PipelineState<S, H> {
{
if info.count == 0 { if info.count == 0 {
return ProcessResponse::init(resp); return ProcessResponse::init(resp);
} }
@ -440,8 +435,7 @@ enum IOState {
impl<S: 'static, H> ProcessResponse<S, H> { impl<S: 'static, H> ProcessResponse<S, H> {
#[inline] #[inline]
fn init(resp: HttpResponse) -> PipelineState<S, H> fn init(resp: HttpResponse) -> PipelineState<S, H> {
{
PipelineState::Response( PipelineState::Response(
ProcessResponse{ resp: resp, ProcessResponse{ resp: resp,
iostate: IOState::Response, iostate: IOState::Response,
@ -513,7 +507,7 @@ impl<S: 'static, H> ProcessResponse<S, H> {
match ctx.poll() { match ctx.poll() {
Ok(Async::Ready(Some(frame))) => { Ok(Async::Ready(Some(frame))) => {
match frame { match frame {
Frame::Payload(None) => { Frame::Chunk(None) => {
info.context = Some(ctx); info.context = Some(ctx);
self.iostate = IOState::Done; self.iostate = IOState::Done;
if let Err(err) = io.write_eof() { if let Err(err) = io.write_eof() {
@ -523,7 +517,7 @@ impl<S: 'static, H> ProcessResponse<S, H> {
} }
break break
}, },
Frame::Payload(Some(chunk)) => { Frame::Chunk(Some(chunk)) => {
self.iostate = IOState::Actor(ctx); self.iostate = IOState::Actor(ctx);
match io.write(chunk.as_ref()) { match io.write(chunk.as_ref()) {
Err(err) => { Err(err) => {

2
src/ws/context.rs
View File

@ -107,7 +107,7 @@ impl<A, S> WebsocketContext<A, S> where A: Actor<Context=Self> {
#[inline] #[inline]
fn write<B: Into<Binary>>(&mut self, data: B) { fn write<B: Into<Binary>>(&mut self, data: B) {
if !self.disconnected { if !self.disconnected {
self.stream.push_back(ContextFrame::Payload(Some(data.into()))); self.stream.push_back(ContextFrame::Chunk(Some(data.into())));
} else { } else {
warn!("Trying to write to disconnected response"); warn!("Trying to write to disconnected response");
} }

4
src/ws/frame.rs
View File

@ -224,9 +224,7 @@ impl Frame {
} }
/// Write a frame out to a buffer /// Write a frame out to a buffer
pub fn format<W>(&mut self, w: &mut W) -> Result<(), Error> pub fn format<W: Write>(&mut self, w: &mut W) -> Result<(), Error> {
where W: Write
{
let mut one = 0u8; let mut one = 0u8;
let code: u8 = self.opcode.into(); let code: u8 = self.opcode.into();
if self.finished { if self.finished {