extern crate tokio_core;
extern crate env_logger;
extern crate futures;
use std::io;
use std::net::SocketAddr;
use std::str;
use futures::{Future, Stream, Sink};
use tokio_core::net::{UdpSocket, UdpCodec};
use tokio_core::reactor::Core;
pub struct LineCodec;
impl UdpCodec for LineCodec {
type In = (SocketAddr, Vec<u8>);
type Out = (SocketAddr, Vec<u8>);
fn decode(&mut self, addr: &SocketAddr, buf: &[u8]) -> io::Result<Self::In> {
Ok((*addr, buf.to_vec()))
}
fn encode(&mut self, (addr, buf): Self::Out, into: &mut Vec<u8>) -> SocketAddr {
into.extend(buf);
addr
}
}
fn main() {
drop(env_logger::init());
let mut core = Core::new().unwrap();
let handle = core.handle();
let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
let a = UdpSocket::bind(&addr, &handle).unwrap();
let b = UdpSocket::bind(&addr, &handle).unwrap();
let b_addr = b.local_addr().unwrap();
let (a_sink, a_stream) = a.framed(LineCodec).split();
let (b_sink, b_stream) = b.framed(LineCodec).split();
let a = a_sink.send((b_addr, b"PING".to_vec())).and_then(|a_sink| {
let mut i = 0;
let a_stream = a_stream.take(4).map(move |(addr, msg)| {
i += 1;
println!("[a] recv: {}", String::from_utf8_lossy(&msg));
(addr, format!("PING {}", i).into_bytes())
});
a_sink.send_all(a_stream)
});
let b_stream = b_stream.map(|(addr, msg)| {
println!("[b] recv: {}", String::from_utf8_lossy(&msg));
(addr, b"PONG".to_vec())
});
let b = b_sink.send_all(b_stream);
handle.spawn(b.then(|_| Ok(())));
drop(core.run(a));
}