switch to actix master

Cargo.toml

@@ -50,7 +50,8 @@ flate2-rust = ["flate2/rust_backend"]
 features = ["tls", "alpn", "session", "brotli", "flate2-c"]
 
 [dependencies]
-actix = "0.6.1"
+# actix = "0.6.1"
+actix = { git="https://github.com/actix/actix.git" }
 
 base64 = "0.9"
 bitflags = "1.0"

src/client/connector.rs

@@ -6,7 +6,7 @@ use std::{fmt, io, mem, time};
 use actix::resolver::{Connect as ResolveConnect, Resolver, ResolverError};
 use actix::{
     fut, Actor, ActorContext, ActorFuture, ActorResponse, Addr, AsyncContext, Context,
-    ContextFutureSpawner, Handler, Message, Recipient, StreamHandler, Supervised,
+    ContextFutureSpawner, Handler, Message, Recipient, StreamHandler2, Supervised,
     SystemService, WrapFuture,
 };
 
@@ -220,7 +220,7 @@ impl Actor for ClientConnector {
             self.resolver = Some(Resolver::from_registry())
         }
         self.collect_periodic(ctx);
-        ctx.add_stream(self.acq_rx.take().unwrap());
+        ctx.add_stream2(self.acq_rx.take().unwrap());
         ctx.spawn(Maintenance);
     }
 }
@@ -767,7 +767,7 @@ impl Handler<Connect> for ClientConnector {
     }
 }
 
-impl StreamHandler<AcquiredConnOperation, ()> for ClientConnector {
+impl StreamHandler2<AcquiredConnOperation, ()> for ClientConnector {
     fn handle(
         &mut self, msg: Result<Option<AcquiredConnOperation>, ()>,
         ctx: &mut Context<Self>,

src/httpresponse.rs

@@ -972,7 +972,7 @@ impl InnerHttpResponse {
     }
 }
 
-/// Internal use only! unsafe
+/// Internal use only!
 pub(crate) struct HttpResponsePool(VecDeque<Box<InnerHttpResponse>>);
 
 thread_local!(static POOL: Rc<UnsafeCell<HttpResponsePool>> = HttpResponsePool::pool());

src/router.rs

@@ -309,7 +309,7 @@ impl Resource {
         params.set_tail(len as u16);
         for (idx, segment) in segments.iter().enumerate() {
             // reason: Router is part of App, which is unique per thread
-            // app is alive during whole life of tthread
+            // app is alive during whole life of a thread
             let name = unsafe { &*(names[idx].as_str() as *const _) };
             params.add(name, *segment);
         }
@@ -378,7 +378,7 @@ impl Resource {
         params.set_tail(tail_len as u16);
         for (idx, segment) in segments.iter().enumerate() {
             // reason: Router is part of App, which is unique per thread
-            // app is alive during whole life of tthread
+            // app is alive during whole life of a thread
             let name = unsafe { &*(names[idx].as_str() as *const _) };
             params.add(name, *segment);
         }

src/server/srv.rs

@@ -5,7 +5,7 @@ use std::{io, net, thread};
 
 use actix::{
     fut, signal, Actor, ActorFuture, Addr, Arbiter, AsyncContext, Context, Handler,
-    Response, StreamHandler, System, WrapFuture,
+    Response, StreamHandler2, System, WrapFuture,
 };
 
 use futures::sync::mpsc;
@@ -449,7 +449,7 @@ impl<H: IntoHttpHandler> HttpServer<H> {
             // start http server actor
             let signals = self.subscribe_to_signals();
             let addr = Actor::create(move |ctx| {
-                ctx.add_stream(rx);
+                ctx.add_stream2(rx);
                 self
             });
             if let Some(signals) = signals {
@@ -611,7 +611,7 @@ impl<H: IntoHttpHandler> Handler<signal::Signal> for HttpServer<H> {
 }
 
 /// Commands from accept threads
-impl<H: IntoHttpHandler> StreamHandler<ServerCommand, ()> for HttpServer<H> {
+impl<H: IntoHttpHandler> StreamHandler2<ServerCommand, ()> for HttpServer<H> {
     fn handle(&mut self, msg: Result<Option<ServerCommand>, ()>, _: &mut Context<Self>) {
         if let Ok(Some(ServerCommand::WorkerDied(idx, socks))) = msg {
             let mut found = false;

tests/test_ws.rs

@@ -23,16 +23,13 @@ impl Actor for Ws {
 }
 
 impl StreamHandler<ws::Message, ws::ProtocolError> for Ws {
-    fn handle(
+    fn handle(&mut self, msg: ws::Message, ctx: &mut Self::Context) {
-        &mut self, msg: Result<Option<ws::Message>, ws::ProtocolError>,
-        ctx: &mut Self::Context,
-    ) {
         match msg {
-            Ok(Some(ws::Message::Ping(msg))) => ctx.pong(&msg),
+            ws::Message::Ping(msg) => ctx.pong(&msg),
-            Ok(Some(ws::Message::Text(text))) => ctx.text(text),
+            ws::Message::Text(text) => ctx.text(text),
-            Ok(Some(ws::Message::Binary(bin))) => ctx.binary(bin),
+            ws::Message::Binary(bin) => ctx.binary(bin),
-            Ok(Some(ws::Message::Close(reason))) => ctx.close(reason),
+            ws::Message::Close(reason) => ctx.close(reason),
-            _ => ctx.stop(),
+            _ => (),
         }
     }
 }
@@ -156,16 +153,13 @@ impl Ws2 {
 }
 
 impl StreamHandler<ws::Message, ws::ProtocolError> for Ws2 {
-    fn handle(
+    fn handle(&mut self, msg: ws::Message, ctx: &mut Self::Context) {
-        &mut self, msg: Result<Option<ws::Message>, ws::ProtocolError>,
-        ctx: &mut Self::Context,
-    ) {
         match msg {
-            Ok(Some(ws::Message::Ping(msg))) => ctx.pong(&msg),
+            ws::Message::Ping(msg) => ctx.pong(&msg),
-            Ok(Some(ws::Message::Text(text))) => ctx.text(text),
+            ws::Message::Text(text) => ctx.text(text),
-            Ok(Some(ws::Message::Binary(bin))) => ctx.binary(bin),
+            ws::Message::Binary(bin) => ctx.binary(bin),
-            Ok(Some(ws::Message::Close(reason))) => ctx.close(reason),
+            ws::Message::Close(reason) => ctx.close(reason),
-            _ => ctx.stop(),
+            _ => (),
         }
     }
 }