Changed connection to be immutable and limit number of messages on buffer to one

Author: Nathan Cohen

src/connection.rs

@@ -89,17 +89,23 @@ where
         } = self;
         let mut socket = Pin::new(socket);
 
-        while !protocol.outgoing_messages.is_empty() {
+        if !protocol.outgoing_messages.is_empty() {
             trace!(
                 "found outgoing message to send checking if socket is ready"
             );
-            if let Poll::Ready(Err(e)) = Pin::as_mut(&mut socket).poll_ready(cx)
-            {
-                // Sink errors are usually not recoverable. The socket
-                // probably shut down.
-                warn!("netlink socket shut down: {:?}", e);
-                self.socket_closed = true;
-                return;
+            match Pin::as_mut(&mut socket).poll_ready(cx) {
+                Poll::Ready(Err(e)) => {
+                    // Sink errors are usually not recoverable. The socket
+                    // probably shut down.
+                    warn!("netlink socket shut down: {:?}", e);
+                    self.socket_closed = true;
+                    return;
+                }
+                Poll::Pending => {
+                    trace!("poll is not ready, returning");
+                    return;
+                }
+                Poll::Ready(Ok(_)) => {}
             }
 
             let (mut message, addr) =

src/handle.rs

@@ -35,7 +35,7 @@ where
     ///   message, the stream is
     /// closed
     pub fn request(
-        &mut self,
+        &self,
         message: NetlinkMessage<T>,
         destination: SocketAddr,
     ) -> Result<impl Stream<Item = NetlinkMessage<T>>, Error<T>> {
@@ -59,7 +59,7 @@ where
     }
 
     pub fn notify(
-        &mut self,
+        &self,
         message: NetlinkMessage<T>,
         destination: SocketAddr,
     ) -> Result<(), Error<T>> {