codec.rs

extern crate tokio;
extern crate dhcp_proto;

use std::io;
use tokio::codec::{Decoder, Encoder};
use bytes::{BytesMut};

use dhcp_proto::{DHCPMessage, nom, parse_dhcp};
use dhcp_proto::builder::DHCPMessageBuilder;

#[derive(Default)]
pub struct DHCPCodec {
    decode_need_message_bytes: usize,
}

impl DHCPCodec {
    pub fn new() -> DHCPCodec {
        DHCPCodec { decode_need_message_bytes: 0}
    }
}

impl Decoder for DHCPCodec {
    type Item = DHCPMessage;
    type Error = io::Error;

    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
        if self.decode_need_message_bytes > src.len() {
            return Ok(None);
        }
        let (message, msg_len) = match parse_dhcp(src) {
            Ok((remaining, message)) => (message, src.len() - remaining.len()),
            Err(nom::Err::Incomplete(nom::Needed::Size(min))) => {
                println!("incomplete, {}", min);
                self.decode_need_message_bytes = min;
                return Ok(None);
            },
            Err(nom::Err::Incomplete(_)) => {
                println!("incomplete");
                return Ok(None);
            }
            Err(err) => {
                println!("error");
                return Err(io::Error::new(
                    io::ErrorKind::Other,
                    format!("{} during parsing of {:?}", err, src),
                ));
            }
        };
        src.split_to(msg_len);
        self.decode_need_message_bytes = 0;
        Ok(Some(message))
    }
}

impl Encoder for DHCPCodec {
    type Item = DHCPMessageBuilder;
    type Error = io::Error;

    fn encode(&mut self, item: Self::Item, dst: &mut BytesMut) -> Result<(), Self::Error> {
        item.write(dst);
        Ok(())
    }
}

main.rs

use tokio::net::{UdpFramed, UdpSocket};
use dhcp_tokio::DHCPCodec;
use tokio::prelude::*;

fn main() {
    let addr = "0.0.0.0:6767".parse().unwrap();
    let socket = UdpSocket::bind(&addr).unwrap();
    println!("Listening on: {}", socket.local_addr().unwrap());

    let (sink, stream) =
        UdpFramed::new(socket, DHCPCodec::new()).split();
    
    let run_loop = stream.for_each(|message| {
        println!("received {:#?}", message.0);
        println!("from {:?}", message.1);
        Ok(())
    }).map_err(|err| {
        println!("connection error = {:?}", err);
    });

    println!("Started");
    tokio::run({
        run_loop
    });
    println!("Stopped");
}