actix tcp server

use bytes::BytesMut;
use futures::Stream;
use std::str::FromStr;
use actix::prelude::*;
use actix::io::{FramedWrite, WriteHandler};
use tokio_codec::FramedRead;
use tokio_tcp::{TcpListener, TcpStream};
use tokio_io::{AsyncRead, codec::{Decoder, Encoder}, io::WriteHalf};

struct Server;
struct Client(FramedWrite<WriteHalf<TcpStream>, HttpCodec>);

#[derive(Message)]
struct Incoming(pub TcpStream);

impl Actor for Server {
    type Context = Context<Self>;
}

impl Handler<Incoming> for Server {
    type Result = ();

    fn handle(&mut self, msg: Incoming, _: &mut Context<Self>) {
        Client::create(move |ctx| {
            let (r, w) = msg.0.split();
            Client::add_stream(FramedRead::new(r, HttpCodec), ctx);
            Client(FramedWrite::new(w, HttpCodec, ctx))
        });
    }
}

impl Actor for Client {
    type Context = Context<Self>;
}

impl WriteHandler<std::io::Error> for Client {}
impl StreamHandler<HttpPacket, std::io::Error> for Client {
    fn handle(&mut self, _: HttpPacket, _: &mut Self::Context) {
        self.0.write(HttpResponse);
    }
}

#[derive(Message)]
struct HttpPacket;
#[derive(Message)]
struct HttpResponse;
struct HttpCodec;

impl Decoder for HttpCodec {
    type Item = HttpPacket;
    type Error = std::io::Error;
    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
        Ok(twoway::find_bytes(src, b"\r\n\r\n").map(|offset| {
            src.advance(offset + b"\r\n\r\n".len());
            HttpPacket
        }))
    }
}

impl Encoder for HttpCodec {
    type Item = HttpResponse;
    type Error = std::io::Error;
    fn encode(&mut self, _: Self::Item, dst: &mut BytesMut) -> Result<(), Self::Error> {
        dst.extend_from_slice(b"HTTP/1.1 200 Ok\r\nContent-Length: 11\r\n\r\nHello World");
        Ok(())
    }
}

fn main() -> std::io::Result<()> {
    actix::System::run(|| {
        let addr = std::net::SocketAddr::from_str("0.0.0.0:12345").unwrap();
        let listener = TcpListener::bind(&addr).unwrap();
        Server::create(|ctx| {
            ctx.add_message_stream(listener
                .incoming()
                .map_err(|_| ())
                .map(|stream| Incoming(stream))
            );
            Server
        });
        actix::spawn(tokio_signal::ctrl_c()
            .flatten_stream()
            .for_each(|_| {
                System::current().stop();
                Ok(())
            })
            .map_err(|_| ())
        );
        println!("Listening on :12345");
    })
}