lots of stuff

in particular, parse link info correctly

netlink-ip/Cargo.toml

@@ -3,8 +3,15 @@ name = "netlink-ip"
 version = "0.1.0"
 authors = ["Corentin Henry <[email protected]>"]
 
+[dependencies]
+bytes = "*"
+tokio-core = "*"
+futures = "0.1"
+log = "*"
+lazy_static = "*"
+
 [dependencies.netlink-sys]
 version = "*"
 default-features = false
-features = ["rtnl_support"]
+features = ["rtnl_support", "tokio_support"]
 path = "../netlink-sys"

netlink-ip/src/connection/connection.rs

@@ -0,0 +1,140 @@
+use std::collections::HashMap;
+use std::io;
+
+use futures::sync::mpsc::{UnboundedReceiver, UnboundedSender};
+use futures::{Async, Future, Poll, Sink, Stream};
+
+use netlink_sys::rtnl::Message;
+use netlink_sys::{NetlinkCodec, NetlinkFramed, Protocol, SocketAddr, TokioSocket};
+
+type RequestsRx = UnboundedReceiver<(UnboundedSender<Message>, Message)>;
+
+lazy_static! {
+    static ref KERNEL_PORT: SocketAddr = SocketAddr::new(0, 0);
+}
+
+pub struct Connection {
+    socket: NetlinkFramed<NetlinkCodec<Message>>,
+    sequence_id: u32,
+    pending_requests: HashMap<u32, UnboundedSender<Message>>,
+    requests_rx: RequestsRx,
+    shutting_down: bool,
+}
+
+impl Connection {
+    pub(crate) fn new(requests_rx: RequestsRx) -> io::Result<Self> {
+        let socket = TokioSocket::new(Protocol::Route)?;
+        socket.connect(&SocketAddr::new(0, 0))?;
+        Ok(Connection {
+            socket: NetlinkFramed::new(socket, NetlinkCodec::<Message>::new()),
+            sequence_id: 0,
+            pending_requests: HashMap::new(),
+            requests_rx,
+            shutting_down: false,
+        })
+    }
+
+    fn send_request(&mut self, mut message: Message) -> u32 {
+        self.sequence_id += 1;
+        message.set_sequence_number(self.sequence_id);
+        message.finalize();
+        // FIXME: in futures 0.2, use poll_ready before reading from pending_responses, and
+        // don't panic here.
+        self.socket.start_send((message, *KERNEL_PORT)).unwrap();
+        self.sequence_id
+    }
+
+    fn handle_message(&mut self, message: Message) {
+        let seq = message.sequence_number();
+        let mut close_chan = false;
+
+        if let Some(tx) = self.pending_requests.get_mut(&seq) {
+            if ! message.flags().has_multipart() {
+                trace!("not a multipart message");
+                close_chan = true;
+            }
+
+            if message.is_done() {
+                trace!("received end of dump message");
+                close_chan = true;
+            } else if message.is_noop() {
+                trace!("ignoring NOOP");
+            } else if message.is_error() {
+                //
+                // FIXME: handle ack! They are special errors!
+                //
+                trace!("forwarding error message and closing channel with handle");
+                // If send returns an Err, its because the other side has been dropped, so it
+                // does not really matter.
+                let _ = UnboundedSender::unbounded_send(tx, message);
+                close_chan = true;
+            } else if message.is_overrun() {
+                // FIXME: we should obviously NOT panic here but I'm not sure what we
+                // should do. Can we increase the buffer size now? Should we leave that the
+                // users?
+                panic!("overrun: receive buffer is full");
+            }
+        } else {
+            // FIXME: we should check whether it's an Overrun error maybe?
+            trace!(
+                "unknown sequence number {}, ignoring the message",
+                message.sequence_number()
+            );
+        }
+
+        if close_chan {
+            let _ = self.pending_requests.remove(&seq);
+        }
+    }
+
+    fn shutdown(&mut self) {
+        self.requests_rx.close();
+        self.shutting_down = true;
+    }
+}
+
+impl Future for Connection {
+    type Item = ();
+    type Error = io::Error;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        trace!("polling socket");
+
+        while let Async::Ready(msg) = self.socket.poll().unwrap() {
+            if let Some((msg, _addr)) = msg {
+                trace!("message received: {:?}", msg);
+                self.handle_message(msg);
+            } else {
+                trace!("socket closed");
+                // XXX: check if there's something else to do?
+                return Ok(Async::Ready(()));
+            }
+        }
+
+        if let Async::NotReady = self.socket.poll_complete().unwrap() {
+            // We do not poll the requests channel if the sink is full to create backpressure. It's
+            // ok not to poll because as soon as the sink makes progress, this future will be
+            // called.
+            return Ok(Async::NotReady);
+        }
+
+        if self.shutting_down {
+            // If we're shutting down, we don't accept any more request
+            return Ok(Async::NotReady);
+        }
+
+        trace!("polling requests channel");
+        while let Async::Ready(request) = self.requests_rx.poll().unwrap() {
+            if let Some((tx_channel, msg)) = request {
+                trace!("request received");
+                let seq = self.send_request(msg);
+                self.pending_requests.insert(seq, tx_channel);
+            } else {
+                trace!("requests channel is closed");
+                self.shutdown();
+            }
+        }
+
+        Ok(Async::NotReady)
+    }
+}

netlink-ip/src/connection/handle.rs

@@ -0,0 +1,21 @@
+use std::io;
+use futures::sync::mpsc::{unbounded, UnboundedReceiver, UnboundedSender};
+use netlink_sys::rtnl::Message;
+
+type RequestsTx = UnboundedSender<(UnboundedSender<Message>, Message)>;
+
+pub struct ConnectionHandle {
+    requests_tx: UnboundedSender<(UnboundedSender<Message>, Message)>,
+}
+
+impl ConnectionHandle {
+    pub(crate) fn new(requests_tx: RequestsTx) -> Self {
+        ConnectionHandle { requests_tx }
+    }
+
+    pub fn request<S>(&mut self, message: Message) -> io::Result<UnboundedReceiver<Message>> {
+        let (tx, rx) = unbounded::<Message>();
+        UnboundedSender::unbounded_send(&self.requests_tx, (tx, message)).unwrap();
+        Ok(rx)
+    }
+}

netlink-ip/src/connection/mod.rs

@@ -0,0 +1,18 @@
+mod connection;
+mod handle;
+
+pub use self::connection::*;
+pub use self::handle::*;
+
+use futures::sync::mpsc::{unbounded, UnboundedSender};
+use futures::Future;
+use netlink_sys::rtnl::Message;
+use std::io;
+use tokio_core::reactor::Handle;
+
+pub fn new_connection(handle: &Handle) -> io::Result<ConnectionHandle> {
+    let (tx, rx) = unbounded::<(UnboundedSender<Message>, Message)>();
+    let connection = Connection::new(rx)?;
+    handle.spawn(connection.map_err(|_| ()));
+    Ok(ConnectionHandle::new(tx))
+}

netlink-ip/src/lib.rs

@@ -0,0 +1,12 @@
+#[macro_use]
+extern crate log;
+#[macro_use]
+extern crate lazy_static;
+extern crate bytes;
+extern crate futures;
+extern crate netlink_sys;
+extern crate tokio_core;
+
+mod connection;
+
+pub use connection::*;

netlink-ip/src/link.rs

@@ -0,0 +1,10 @@
+use netlink_sys::packet::rtnl::{LinkNla};
+
+pub struct Link {
+    index: u32,
+    attributes: Vec<LinkNla>,
+}
+
+// impl Link {
+//     fn new
+// }

netlink-sys/Cargo.toml

@@ -2,6 +2,7 @@
 authors = ["Corentin Henry <[email protected]>"]
 name = "netlink-sys"
 version = "0.1.0"
+
 [dependencies]
 byteorder = "1.2.3"
 bytes = "0.4.8"

netlink-sys/examples/dump_links.rs

@@ -1,8 +1,8 @@
-extern crate netlink;
+extern crate netlink_sys;
 
-use netlink::constants::{NLM_F_DUMP, NLM_F_REQUEST};
-use netlink::rtnl::{LinkFlags, LinkHeader, LinkLayerType, LinkMessage, Message, RtnlMessage};
-use netlink::{NetlinkFlags, Protocol, Socket, SocketAddr};
+use netlink_sys::constants::{NLM_F_DUMP, NLM_F_REQUEST};
+use netlink_sys::rtnl::{LinkFlags, LinkHeader, LinkLayerType, LinkMessage, Message, RtnlMessage};
+use netlink_sys::{NetlinkFlags, Protocol, Socket, SocketAddr};
 
 fn main() {
     let mut socket = Socket::new(Protocol::Route).unwrap();

netlink-sys/examples/dump_links_tokio.rs

@@ -1,10 +1,10 @@
 extern crate futures;
-extern crate netlink;
+extern crate netlink_sys;
 
 use futures::{Future, Sink, Stream};
-use netlink::constants::{NLM_F_DUMP, NLM_F_REQUEST};
-use netlink::rtnl::{LinkFlags, LinkHeader, LinkLayerType, LinkMessage, Message, RtnlMessage};
-use netlink::{NetlinkCodec, NetlinkFlags, NetlinkFramed, Protocol, SocketAddr, TokioSocket};
+use netlink_sys::constants::{NLM_F_DUMP, NLM_F_REQUEST};
+use netlink_sys::rtnl::{LinkFlags, LinkHeader, LinkLayerType, LinkMessage, Message, RtnlMessage};
+use netlink_sys::{NetlinkCodec, NetlinkFlags, NetlinkFramed, Protocol, SocketAddr, TokioSocket};
 
 fn main() {
     let mut socket = TokioSocket::new(Protocol::Route).unwrap();

netlink-sys/src/bindgen_constants.rs

@@ -86,52 +86,52 @@ pub const IFLA_INET6_ADDR_GEN_MODE: int = 8;
 // pub const IN6_ADDR_GEN_MODE_NONE: int = 1;
 // pub const IN6_ADDR_GEN_MODE_STABLE_PRIVACY: int = 2;
 // pub const IN6_ADDR_GEN_MODE_RANDOM: int = 3;
-//
-// pub const IFLA_BR_UNSPEC: int = 0;
-// pub const IFLA_BR_FORWARD_DELAY: int = 1;
-// pub const IFLA_BR_HELLO_TIME: int = 2;
-// pub const IFLA_BR_MAX_AGE: int = 3;
-// pub const IFLA_BR_AGEING_TIME: int = 4;
-// pub const IFLA_BR_STP_STATE: int = 5;
-// pub const IFLA_BR_PRIORITY: int = 6;
-// pub const IFLA_BR_VLAN_FILTERING: int = 7;
-// pub const IFLA_BR_VLAN_PROTOCOL: int = 8;
-// pub const IFLA_BR_GROUP_FWD_MASK: int = 9;
-// pub const IFLA_BR_ROOT_ID: int = 10;
-// pub const IFLA_BR_BRIDGE_ID: int = 11;
-// pub const IFLA_BR_ROOT_PORT: int = 12;
-// pub const IFLA_BR_ROOT_PATH_COST: int = 13;
-// pub const IFLA_BR_TOPOLOGY_CHANGE: int = 14;
-// pub const IFLA_BR_TOPOLOGY_CHANGE_DETECTED: int = 15;
-// pub const IFLA_BR_HELLO_TIMER: int = 16;
-// pub const IFLA_BR_TCN_TIMER: int = 17;
-// pub const IFLA_BR_TOPOLOGY_CHANGE_TIMER: int = 18;
-// pub const IFLA_BR_GC_TIMER: int = 19;
-// pub const IFLA_BR_GROUP_ADDR: int = 20;
-// pub const IFLA_BR_FDB_FLUSH: int = 21;
-// pub const IFLA_BR_MCAST_ROUTER: int = 22;
-// pub const IFLA_BR_MCAST_SNOOPING: int = 23;
-// pub const IFLA_BR_MCAST_QUERY_USE_IFADDR: int = 24;
-// pub const IFLA_BR_MCAST_QUERIER: int = 25;
-// pub const IFLA_BR_MCAST_HASH_ELASTICITY: int = 26;
-// pub const IFLA_BR_MCAST_HASH_MAX: int = 27;
-// pub const IFLA_BR_MCAST_LAST_MEMBER_CNT: int = 28;
-// pub const IFLA_BR_MCAST_STARTUP_QUERY_CNT: int = 29;
-// pub const IFLA_BR_MCAST_LAST_MEMBER_INTVL: int = 30;
-// pub const IFLA_BR_MCAST_MEMBERSHIP_INTVL: int = 31;
-// pub const IFLA_BR_MCAST_QUERIER_INTVL: int = 32;
-// pub const IFLA_BR_MCAST_QUERY_INTVL: int = 33;
-// pub const IFLA_BR_MCAST_QUERY_RESPONSE_INTVL: int = 34;
-// pub const IFLA_BR_MCAST_STARTUP_QUERY_INTVL: int = 35;
-// pub const IFLA_BR_NF_CALL_IPTABLES: int = 36;
-// pub const IFLA_BR_NF_CALL_IP6TABLES: int = 37;
-// pub const IFLA_BR_NF_CALL_ARPTABLES: int = 38;
-// pub const IFLA_BR_VLAN_DEFAULT_PVID: int = 39;
-// pub const IFLA_BR_PAD: int = 40;
-// pub const IFLA_BR_VLAN_STATS_ENABLED: int = 41;
-// pub const IFLA_BR_MCAST_STATS_ENABLED: int = 42;
-// pub const IFLA_BR_MCAST_IGMP_VERSION: int = 43;
-// pub const IFLA_BR_MCAST_MLD_VERSION: int = 44;
+
+pub const IFLA_BR_UNSPEC: int = 0;
+pub const IFLA_BR_FORWARD_DELAY: int = 1;
+pub const IFLA_BR_HELLO_TIME: int = 2;
+pub const IFLA_BR_MAX_AGE: int = 3;
+pub const IFLA_BR_AGEING_TIME: int = 4;
+pub const IFLA_BR_STP_STATE: int = 5;
+pub const IFLA_BR_PRIORITY: int = 6;
+pub const IFLA_BR_VLAN_FILTERING: int = 7;
+pub const IFLA_BR_VLAN_PROTOCOL: int = 8;
+pub const IFLA_BR_GROUP_FWD_MASK: int = 9;
+pub const IFLA_BR_ROOT_ID: int = 10;
+pub const IFLA_BR_BRIDGE_ID: int = 11;
+pub const IFLA_BR_ROOT_PORT: int = 12;
+pub const IFLA_BR_ROOT_PATH_COST: int = 13;
+pub const IFLA_BR_TOPOLOGY_CHANGE: int = 14;
+pub const IFLA_BR_TOPOLOGY_CHANGE_DETECTED: int = 15;
+pub const IFLA_BR_HELLO_TIMER: int = 16;
+pub const IFLA_BR_TCN_TIMER: int = 17;
+pub const IFLA_BR_TOPOLOGY_CHANGE_TIMER: int = 18;
+pub const IFLA_BR_GC_TIMER: int = 19;
+pub const IFLA_BR_GROUP_ADDR: int = 20;
+pub const IFLA_BR_FDB_FLUSH: int = 21;
+pub const IFLA_BR_MCAST_ROUTER: int = 22;
+pub const IFLA_BR_MCAST_SNOOPING: int = 23;
+pub const IFLA_BR_MCAST_QUERY_USE_IFADDR: int = 24;
+pub const IFLA_BR_MCAST_QUERIER: int = 25;
+pub const IFLA_BR_MCAST_HASH_ELASTICITY: int = 26;
+pub const IFLA_BR_MCAST_HASH_MAX: int = 27;
+pub const IFLA_BR_MCAST_LAST_MEMBER_CNT: int = 28;
+pub const IFLA_BR_MCAST_STARTUP_QUERY_CNT: int = 29;
+pub const IFLA_BR_MCAST_LAST_MEMBER_INTVL: int = 30;
+pub const IFLA_BR_MCAST_MEMBERSHIP_INTVL: int = 31;
+pub const IFLA_BR_MCAST_QUERIER_INTVL: int = 32;
+pub const IFLA_BR_MCAST_QUERY_INTVL: int = 33;
+pub const IFLA_BR_MCAST_QUERY_RESPONSE_INTVL: int = 34;
+pub const IFLA_BR_MCAST_STARTUP_QUERY_INTVL: int = 35;
+pub const IFLA_BR_NF_CALL_IPTABLES: int = 36;
+pub const IFLA_BR_NF_CALL_IP6TABLES: int = 37;
+pub const IFLA_BR_NF_CALL_ARPTABLES: int = 38;
+pub const IFLA_BR_VLAN_DEFAULT_PVID: int = 39;
+pub const IFLA_BR_PAD: int = 40;
+pub const IFLA_BR_VLAN_STATS_ENABLED: int = 41;
+pub const IFLA_BR_MCAST_STATS_ENABLED: int = 42;
+pub const IFLA_BR_MCAST_IGMP_VERSION: int = 43;
+pub const IFLA_BR_MCAST_MLD_VERSION: int = 44;
 //
 // pub const BRIDGE_MODE_UNSPEC: int = 0;
 // pub const BRIDGE_MODE_HAIRPIN: int = 1;
@@ -169,21 +169,21 @@ pub const IFLA_INET6_ADDR_GEN_MODE: int = 8;
 // pub const IFLA_BRPORT_BCAST_FLOOD: int = 30;
 // pub const IFLA_BRPORT_GROUP_FWD_MASK: int = 31;
 // pub const IFLA_BRPORT_NEIGH_SUPPRESS: int = 32;
-//
-// pub const IFLA_INFO_UNSPEC: int = 0;
-// pub const IFLA_INFO_KIND: int = 1;
-// pub const IFLA_INFO_DATA: int = 2;
-// pub const IFLA_INFO_XSTATS: int = 3;
-// pub const IFLA_INFO_SLAVE_KIND: int = 4;
-// pub const IFLA_INFO_SLAVE_DATA: int = 5;
-//
-// pub const IFLA_VLAN_UNSPEC: int = 0;
-// pub const IFLA_VLAN_ID: int = 1;
-// pub const IFLA_VLAN_FLAGS: int = 2;
-// pub const IFLA_VLAN_EGRESS_QOS: int = 3;
-// pub const IFLA_VLAN_INGRESS_QOS: int = 4;
-// pub const IFLA_VLAN_PROTOCOL: int = 5;
-//
+
+pub const IFLA_INFO_UNSPEC: int = 0;
+pub const IFLA_INFO_KIND: int = 1;
+pub const IFLA_INFO_DATA: int = 2;
+pub const IFLA_INFO_XSTATS: int = 3;
+pub const IFLA_INFO_SLAVE_KIND: int = 4;
+pub const IFLA_INFO_SLAVE_DATA: int = 5;
+
+pub const IFLA_VLAN_UNSPEC: int = 0;
+pub const IFLA_VLAN_ID: int = 1;
+pub const IFLA_VLAN_FLAGS: int = 2;
+pub const IFLA_VLAN_EGRESS_QOS: int = 3;
+pub const IFLA_VLAN_INGRESS_QOS: int = 4;
+pub const IFLA_VLAN_PROTOCOL: int = 5;
+
 // pub const IFLA_VLAN_QOS_UNSPEC: int = 0;
 // pub const IFLA_VLAN_QOS_MAPPING: int = 1;
 //