extern crate actix;
use futures::sync::oneshot;
use futures::sync::oneshot::Sender;
use futures::{Async, Future, Poll};
use smallvec::SmallVec;
use std::marker::PhantomData;
use self::actix::dev::{
AsyncContextParts, ContextFut, ContextParts, Envelope, Mailbox, ToEnvelope,
};
use self::actix::fut::ActorFuture;
use self::actix::{
Actor, ActorContext, ActorState, Addr, AsyncContext, Handler, Message, SpawnHandle,
};
use body::{Binary, Body};
use error::{Error, ErrorInternalServerError};
use httprequest::HttpRequest;
pub trait ActorHttpContext: 'static {
fn disconnected(&mut self);
fn poll(&mut self) -> Poll<Option<SmallVec<[Frame; 4]>>, Error>;
}
#[derive(Debug)]
pub enum Frame {
Chunk(Option<Binary>),
Drain(oneshot::Sender<()>),
}
impl Frame {
pub fn len(&self) -> usize {
match *self {
Frame::Chunk(Some(ref bin)) => bin.len(),
_ => 0,
}
}
}
pub struct HttpContext<A, S = ()>
where
A: Actor<Context = HttpContext<A, S>>,
{
inner: ContextParts<A>,
stream: Option<SmallVec<[Frame; 4]>>,
request: HttpRequest<S>,
disconnected: bool,
}
impl<A, S> ActorContext for HttpContext<A, S>
where
A: Actor<Context = Self>,
{
fn stop(&mut self) {
self.inner.stop();
}
fn terminate(&mut self) {
self.inner.terminate()
}
fn state(&self) -> ActorState {
self.inner.state()
}
}
impl<A, S> AsyncContext<A> for HttpContext<A, S>
where
A: Actor<Context = Self>,
{
#[inline]
fn spawn<F>(&mut self, fut: F) -> SpawnHandle
where
F: ActorFuture<Item = (), Error = (), Actor = A> + 'static,
{
self.inner.spawn(fut)
}
#[inline]
fn wait<F>(&mut self, fut: F)
where
F: ActorFuture<Item = (), Error = (), Actor = A> + 'static,
{
self.inner.wait(fut)
}
#[doc(hidden)]
#[inline]
fn waiting(&self) -> bool {
self.inner.waiting()
|| self.inner.state() == ActorState::Stopping
|| self.inner.state() == ActorState::Stopped
}
#[inline]
fn cancel_future(&mut self, handle: SpawnHandle) -> bool {
self.inner.cancel_future(handle)
}
#[inline]
fn address(&self) -> Addr<A> {
self.inner.address()
}
}
impl<A, S: 'static> HttpContext<A, S>
where
A: Actor<Context = Self>,
{
#[inline]
pub fn create(req: HttpRequest<S>, actor: A) -> Body {
let mb = Mailbox::default();
let ctx = HttpContext {
inner: ContextParts::new(mb.sender_producer()),
stream: None,
request: req,
disconnected: false,
};
Body::Actor(Box::new(HttpContextFut::new(ctx, actor, mb)))
}
pub fn with_factory<F>(req: HttpRequest<S>, f: F) -> Body
where
F: FnOnce(&mut Self) -> A + 'static,
{
let mb = Mailbox::default();
let mut ctx = HttpContext {
inner: ContextParts::new(mb.sender_producer()),
stream: None,
request: req,
disconnected: false,
};
let act = f(&mut ctx);
Body::Actor(Box::new(HttpContextFut::new(ctx, act, mb)))
}
}
impl<A, S> HttpContext<A, S>
where
A: Actor<Context = Self>,
{
#[inline]
pub fn state(&self) -> &S {
self.request.state()
}
#[inline]
pub fn request(&mut self) -> &mut HttpRequest<S> {
&mut self.request
}
#[inline]
pub fn write<B: Into<Binary>>(&mut self, data: B) {
if !self.disconnected {
self.add_frame(Frame::Chunk(Some(data.into())));
} else {
warn!("Trying to write to disconnected response");
}
}
#[inline]
pub fn write_eof(&mut self) {
self.add_frame(Frame::Chunk(None));
}
pub fn drain(&mut self) -> Drain<A> {
let (tx, rx) = oneshot::channel();
self.add_frame(Frame::Drain(tx));
Drain::new(rx)
}
#[inline]
pub fn connected(&self) -> bool {
!self.disconnected
}
#[inline]
fn add_frame(&mut self, frame: Frame) {
if self.stream.is_none() {
self.stream = Some(SmallVec::new());
}
if let Some(s) = self.stream.as_mut() {
s.push(frame)
}
}
pub fn handle(&self) -> SpawnHandle {
self.inner.curr_handle()
}
}
impl<A, S> AsyncContextParts<A> for HttpContext<A, S>
where
A: Actor<Context = Self>,
{
fn parts(&mut self) -> &mut ContextParts<A> {
&mut self.inner
}
}
struct HttpContextFut<A, S>
where
A: Actor<Context = HttpContext<A, S>>,
{
fut: ContextFut<A, HttpContext<A, S>>,
}
impl<A, S> HttpContextFut<A, S>
where
A: Actor<Context = HttpContext<A, S>>,
{
fn new(ctx: HttpContext<A, S>, act: A, mailbox: Mailbox<A>) -> Self {
let fut = ContextFut::new(ctx, act, mailbox);
HttpContextFut { fut }
}
}
impl<A, S> ActorHttpContext for HttpContextFut<A, S>
where
A: Actor<Context = HttpContext<A, S>>,
S: 'static,
{
#[inline]
fn disconnected(&mut self) {
self.fut.ctx().disconnected = true;
self.fut.ctx().stop();
}
fn poll(&mut self) -> Poll<Option<SmallVec<[Frame; 4]>>, Error> {
if self.fut.alive() {
match self.fut.poll() {
Ok(Async::NotReady) | Ok(Async::Ready(())) => (),
Err(_) => return Err(ErrorInternalServerError("error")),
}
}
if let Some(data) = self.fut.ctx().stream.take() {
Ok(Async::Ready(Some(data)))
} else if self.fut.alive() {
Ok(Async::NotReady)
} else {
Ok(Async::Ready(None))
}
}
}
impl<A, M, S> ToEnvelope<A, M> for HttpContext<A, S>
where
A: Actor<Context = HttpContext<A, S>> + Handler<M>,
M: Message + Send + 'static,
M::Result: Send,
{
fn pack(msg: M, tx: Option<Sender<M::Result>>) -> Envelope<A> {
Envelope::new(msg, tx)
}
}
pub struct Drain<A> {
fut: oneshot::Receiver<()>,
_a: PhantomData<A>,
}
impl<A> Drain<A> {
pub fn new(fut: oneshot::Receiver<()>) -> Self {
Drain {
fut,
_a: PhantomData,
}
}
}
impl<A: Actor> ActorFuture for Drain<A> {
type Item = ();
type Error = ();
type Actor = A;
#[inline]
fn poll(
&mut self, _: &mut A, _: &mut <Self::Actor as Actor>::Context,
) -> Poll<Self::Item, Self::Error> {
self.fut.poll().map_err(|_| ())
}
}