messaging.rs

extern crate futures;
extern crate tokio_core;
extern crate tokio_uds;

use std::io;
use std::io::{Error, Write};
use std::io::ErrorKind;
use std::io::Read;
use std::net::SocketAddr;
use futures::Future;
use futures::stream::{AndThen, Stream};
use futures::sink::Sink;
use tokio_core::io::{copy, Codec, EasyBuf, Io};
use tokio_core::reactor::Core;
use tokio_core::net::TcpListener;

struct StringCodec {
}

impl StringCodec {
    fn new() -> StringCodec {
        StringCodec {}
    }
}

impl Codec for StringCodec {
    type In = String;
    type Out = String;

    fn decode(&mut self, buf: &mut EasyBuf) -> Result<Option<Self::In>, Error> {
        let len = buf.as_slice().len();

        println!("Len: {}", len);

        if (len == 0) {
            Ok(None)
        } else {
            Ok(Some(String::from_utf8_lossy(&buf.drain_to(len).as_slice()[..(len - 2)]).into_owned()))
        }
    }

    fn encode(&mut self, msg: Self::Out, buf: &mut Vec<u8>) -> io::Result<()> {
        buf.write_all(msg.as_bytes())
    }
}

fn main() {
    let mut l = Core::new().unwrap();
    let handle = l.handle();
    let addr = format!("127.0.0.1:{}", 8080);
    let addr = addr.parse::<SocketAddr>().unwrap();
    let tcp_socket = TcpListener::bind(&addr, &handle).unwrap();

    let done = tcp_socket
        .incoming()
        .for_each(move |(socket, addr)| {
            println!("Received connection from: {}", addr);

            let (sink, stream)= socket.framed(StringCodec::new()).split();
            let conn = stream
                .forward(sink.with(|req: String| {
                  let res: Result<String, Error> = Ok(format!("Ok ({} bytes)\n", req.len()));
                  res
                }))
                .then(|_| { Ok(()) });

            handle.spawn(conn);

            Ok(())
        });

    l.run(done);

    println!("Listening on: {}", addr);
}