update example

.travis.yml

@@ -60,7 +60,7 @@ script:
       cd examples/template_tera && cargo check && cd ../..
       cd examples/diesel && cargo check && cd ../..
       cd examples/tls && cargo check && cd ../..
-      # cd examples/websocket-chat && cargo check && cd ../..
+      cd examples/websocket-chat && cargo check && cd ../..
       cd examples/websocket && cargo check && cd ../..
     fi
   - |

examples/websocket-chat/Cargo.toml

@@ -25,5 +25,5 @@ serde = "1.0"
 serde_json = "1.0"
 serde_derive = "1.0"
 
-actix = "^0.4.2"
+actix = { git="https://github.com/actix/actix.git" }
 actix-web = { path="../../" }

examples/websocket-chat/src/client.rs

@@ -12,6 +12,7 @@ use std::{io, net, process, thread};
 use std::str::FromStr;
 use std::time::Duration;
 use futures::Future;
+use tokio_io::AsyncRead;
 use tokio_core::net::TcpStream;
 use actix::prelude::*;
 
@@ -26,7 +27,9 @@ fn main() {
     Arbiter::handle().spawn(
         TcpStream::connect(&addr, Arbiter::handle())
             .and_then(|stream| {
-                let addr: SyncAddress<_> = ChatClient.framed(stream, codec::ClientChatCodec);
+                let addr: SyncAddress<_> = ChatClient::create_with(
+                    stream.framed(codec::ClientChatCodec),
+                    |_, framed| ChatClient{framed: framed});
 
                 // start console loop
                 thread::spawn(move|| {
@@ -54,20 +57,22 @@ fn main() {
 }
 
 
-struct ChatClient;
+struct ChatClient {
+    framed: FramedCell<TcpStream, codec::ClientChatCodec>,
+}
 
 #[derive(Message)]
 struct ClientCommand(String);
 
 impl Actor for ChatClient {
-    type Context = FramedContext<Self>;
+    type Context = Context<Self>;
 
-    fn started(&mut self, ctx: &mut FramedContext<Self>) {
+    fn started(&mut self, ctx: &mut Context<Self>) {
         // start heartbeats otherwise server will disconnect after 10 seconds
         self.hb(ctx)
     }
 
-    fn stopping(&mut self, _: &mut FramedContext<Self>) -> bool {
+    fn stopping(&mut self, _: &mut Context<Self>) -> bool {
         println!("Disconnected");
 
         // Stop application on disconnect
@@ -78,11 +83,10 @@ impl Actor for ChatClient {
 }
 
 impl ChatClient {
-    fn hb(&self, ctx: &mut FramedContext<Self>) {
+    fn hb(&self, ctx: &mut Context<Self>) {
         ctx.run_later(Duration::new(1, 0), |act, ctx| {
-            if ctx.send(codec::ChatRequest::Ping).is_ok() {
+            act.framed.send(codec::ChatRequest::Ping);
-                act.hb(ctx);
+            act.hb(ctx);
-            }
         });
     }
 }
@@ -91,7 +95,7 @@ impl ChatClient {
 impl Handler<ClientCommand> for ChatClient {
     type Result = ();
 
-    fn handle(&mut self, msg: ClientCommand, ctx: &mut FramedContext<Self>) {
+    fn handle(&mut self, msg: ClientCommand, _: &mut Context<Self>) {
         let m = msg.0.trim();
         if m.is_empty() {
             return
@@ -102,11 +106,11 @@ impl Handler<ClientCommand> for ChatClient {
             let v: Vec<&str> = m.splitn(2, ' ').collect();
             match v[0] {
                 "/list" => {
-                    let _ = ctx.send(codec::ChatRequest::List);
+                    let _ = self.framed.send(codec::ChatRequest::List);
                 },
                 "/join" => {
                     if v.len() == 2 {
-                        let _ = ctx.send(codec::ChatRequest::Join(v[1].to_owned()));
+                        let _ = self.framed.send(codec::ChatRequest::Join(v[1].to_owned()));
                     } else {
                         println!("!!! room name is required");
                     }
@@ -114,18 +118,16 @@ impl Handler<ClientCommand> for ChatClient {
                 _ => println!("!!! unknown command"),
             }
         } else {
-            let _ = ctx.send(codec::ChatRequest::Message(m.to_owned()));
+            let _ = self.framed.send(codec::ChatRequest::Message(m.to_owned()));
         }
     }
 }
 
 /// Server communication
 
-impl FramedActor for ChatClient {
+impl FramedActor<TcpStream, codec::ClientChatCodec> for ChatClient {
-    type Io = TcpStream;
-    type Codec = codec::ClientChatCodec;
 
-    fn handle(&mut self, msg: io::Result<codec::ChatResponse>, ctx: &mut FramedContext<Self>) {
+    fn handle(&mut self, msg: io::Result<codec::ChatResponse>, ctx: &mut Context<Self>) {
         match msg {
             Err(_) => ctx.stop(),
             Ok(msg) => match msg {

examples/websocket-chat/src/main.rs

@@ -62,9 +62,9 @@ impl Actor for WsChatSession {
         // before processing any other events.
         // HttpContext::state() is instance of WsChatSessionState, state is shared across all
         // routes within application
-        let subs = ctx.sync_subscriber();
+        let addr: SyncAddress<_> = ctx.address();
         ctx.state().addr.call(
-            self, server::Connect{addr: subs}).then(
+            self, server::Connect{addr: addr.into_subscriber()}).then(
             |res, act, ctx| {
                 match res {
                     Ok(Ok(res)) => act.id = res,

examples/websocket-chat/src/session.rs

@@ -4,6 +4,7 @@ use std::{io, net};
 use std::str::FromStr;
 use std::time::{Instant, Duration};
 use futures::Stream;
+use tokio_io::AsyncRead;
 use tokio_core::net::{TcpStream, TcpListener};
 
 use actix::prelude::*;
@@ -26,12 +27,14 @@ pub struct ChatSession {
     hb: Instant,
     /// joined room
     room: String,
+    /// Framed wrapper
+    framed: FramedCell<TcpStream, ChatCodec>,
 }
 
 impl Actor for ChatSession {
     /// For tcp communication we are going to use `FramedContext`.
     /// It is convenient wrapper around `Framed` object from `tokio_io`
-    type Context = FramedContext<Self>;
+    type Context = Context<Self>;
 
     fn started(&mut self, ctx: &mut Self::Context) {
         // we'll start heartbeat process on session start.
@@ -41,7 +44,7 @@ impl Actor for ChatSession {
         // future within context, but context waits until this future resolves
         // before processing any other events.
         let addr: SyncAddress<_> = ctx.address();
-        self.addr.call(self, server::Connect{addr: addr.subscriber()})
+        self.addr.call(self, server::Connect{addr: addr.into_subscriber()})
             .then(|res, act, ctx| {
                 match res {
                     Ok(Ok(res)) => act.id = res,
@@ -59,23 +62,21 @@ impl Actor for ChatSession {
     }
 }
 
-/// To use `FramedContext` we have to define Io type and Codec
+/// To use `Framed` we have to define Io type and Codec
-impl FramedActor for ChatSession {
+impl FramedActor<TcpStream, ChatCodec> for ChatSession {
-    type Io = TcpStream;
-    type Codec= ChatCodec;
 
     /// This is main event loop for client requests
-    fn handle(&mut self, msg: io::Result<ChatRequest>, ctx: &mut FramedContext<Self>) {
+    fn handle(&mut self, msg: io::Result<ChatRequest>, ctx: &mut Context<Self>) {
         match msg {
             Err(_) => ctx.stop(),
             Ok(msg) => match msg {
                 ChatRequest::List => {
                     // Send ListRooms message to chat server and wait for response
                     println!("List rooms");
-                    self.addr.call(self, server::ListRooms).then(|res, _, ctx| {
+                    self.addr.call(self, server::ListRooms).then(|res, act, ctx| {
                         match res {
                             Ok(Ok(rooms)) => {
-                                let _ = ctx.send(ChatResponse::Rooms(rooms));
+                                let _ = act.framed.send(ChatResponse::Rooms(rooms));
                             },
                         _ => println!("Something is wrong"),
                         }
@@ -88,7 +89,7 @@ impl FramedActor for ChatSession {
                     println!("Join to room: {}", name);
                     self.room = name.clone();
                     self.addr.send(server::Join{id: self.id, name: name.clone()});
-                    let _ = ctx.send(ChatResponse::Joined(name));
+                    let _ = self.framed.send(ChatResponse::Joined(name));
                 },
                 ChatRequest::Message(message) => {
                     // send message to chat server
@@ -110,23 +111,25 @@ impl FramedActor for ChatSession {
 impl Handler<Message> for ChatSession {
     type Result = ();
 
-    fn handle(&mut self, msg: Message, ctx: &mut FramedContext<Self>) {
+    fn handle(&mut self, msg: Message, ctx: &mut Context<Self>) {
         // send message to peer
-        let _ = ctx.send(ChatResponse::Message(msg.0));
+        let _ = self.framed.send(ChatResponse::Message(msg.0));
     }
 }
 
 /// Helper methods
 impl ChatSession {
 
-    pub fn new(addr: SyncAddress<ChatServer>) -> ChatSession {
+    pub fn new(addr: SyncAddress<ChatServer>,
-        ChatSession {id: 0, addr: addr, hb: Instant::now(), room: "Main".to_owned()}
+               framed: FramedCell<TcpStream, ChatCodec>) -> ChatSession {
+        ChatSession {id: 0, addr: addr, hb: Instant::now(),
+                     room: "Main".to_owned(), framed: framed}
     }
     
     /// helper method that sends ping to client every second.
     ///
     /// also this method check heartbeats from client
-    fn hb(&self, ctx: &mut FramedContext<Self>) {
+    fn hb(&self, ctx: &mut Context<Self>) {
         ctx.run_later(Duration::new(1, 0), |act, ctx| {
             // check client heartbeats
             if Instant::now().duration_since(act.hb) > Duration::new(10, 0) {
@@ -140,10 +143,9 @@ impl ChatSession {
                 ctx.stop();
             }
 
-            if ctx.send(ChatResponse::Ping).is_ok() {
+            act.framed.send(ChatResponse::Ping);
-                // if we can not send message to sink, sink is closed (disconnected)
+            // if we can not send message to sink, sink is closed (disconnected)
-                act.hb(ctx);
+            act.hb(ctx);
-            }
         });
     }
 }
@@ -192,6 +194,8 @@ impl Handler<TcpConnect> for TcpServer {
         // For each incoming connection we create `ChatSession` actor
         // with out chat server address.
         let server = self.chat.clone();
-        let _: () = ChatSession::new(server).framed(msg.0, ChatCodec);
+        let _: () = ChatSession::create_with(msg.0.framed(ChatCodec), |_, framed| {
+            ChatSession::new(server, framed)
+        });
     }
 }

examples/websocket/Cargo.toml

@@ -15,6 +15,6 @@ path = "src/client.rs"
 env_logger = "*"
 futures = "0.1"
 tokio-core = "0.1"
-#actix = "^0.4.2"
+#actix = "^0.4.6"
 actix = { git = "https://github.com/actix/actix.git" }
 actix-web = { path="../../" }

src/server/srv.rs

@@ -477,8 +477,10 @@ impl<T, A, H, U, V> HttpServer<WrapperStream<T>, A, H, U>
         // start server
         let signals = self.subscribe_to_signals();
         let addr: SyncAddress<_> = HttpServer::create(move |ctx| {
-            ctx.add_stream(stream.map(
+            ctx.add_message_stream(
-                move |(t, _)| Conn{io: WrapperStream::new(t), peer: None, http2: false}));
+                stream
+                    .map_err(|_| ())
+                    .map(move |(t, _)| Conn{io: WrapperStream::new(t), peer: None, http2: false}));
             self
         });
         signals.map(|signals| signals.send(
@@ -542,6 +544,22 @@ impl<T, A, H, U, V> Handler<io::Result<Conn<T>>> for HttpServer<T, A, H, U>
     }
 }
 
+impl<T, A, H, U, V> Handler<Conn<T>> for HttpServer<T, A, H, U>
+    where T: IoStream,
+          H: HttpHandler + 'static,
+          U: IntoIterator<Item=V> + 'static,
+          V: IntoHttpHandler<Handler=H>,
+          A: 'static,
+{
+    type Result = ();
+
+    fn handle(&mut self, msg: Conn<T>, _: &mut Context<Self>) -> Self::Result {
+        Arbiter::handle().spawn(
+            HttpChannel::new(
+                Rc::clone(self.h.as_ref().unwrap()), msg.io, msg.peer, msg.http2));
+    }
+}
+
 impl<T, A, H, U, V> Handler<PauseServer> for HttpServer<T, A, H, U>
     where T: IoStream,
           H: HttpHandler + 'static,