fix "leak" in Blob::recv_from (#1198)

* fix "leak" in Blob::recv_from

fixes #1199

src/packet.rs

@@ -441,6 +441,18 @@ impl Blob {
         self.meta.size = new_size;
         self.set_data_size(new_size as u64).unwrap();
     }
+
+    pub fn recv_blob(socket: &UdpSocket, r: &SharedBlob) -> io::Result<()> {
+        let mut p = r.write().expect("'r' write lock in pub fn recv_from");
+        trace!("receiving on {}", socket.local_addr().unwrap());
+
+        let (nrecv, from) = socket.recv_from(&mut p.data)?;
+        p.meta.size = nrecv;
+        p.meta.set_addr(&from);
+        trace!("got {} bytes from {}", nrecv, from);
+        Ok(())
+    }
+
     pub fn recv_from(re: &BlobRecycler, socket: &UdpSocket) -> Result<SharedBlobs> {
         let mut v = Vec::new();
         //DOCUMENTED SIDE-EFFECT
@@ -452,29 +464,22 @@ impl Blob {
         socket.set_nonblocking(false)?;
         for i in 0..NUM_BLOBS {
             let r = re.allocate();
-            {
+
-                let mut p = r.write().expect("'r' write lock in pub fn recv_from");
+            match Blob::recv_blob(socket, &r) {
-                trace!("receiving on {}", socket.local_addr().unwrap());
+                Err(_) if i > 0 => {
-                match socket.recv_from(&mut p.data) {
+                    trace!("got {:?} messages on {}", i, socket.local_addr().unwrap());
-                    Err(_) if i > 0 => {
+                    break;
-                        trace!("got {:?} messages on {}", i, socket.local_addr().unwrap());
-                        break;
-                    }
-                    Err(e) => {
-                        if e.kind() != io::ErrorKind::WouldBlock {
-                            info!("recv_from err {:?}", e);
-                        }
-                        return Err(Error::IO(e));
-                    }
-                    Ok((nrecv, from)) => {
-                        p.meta.size = nrecv;
-                        p.meta.set_addr(&from);
-                        trace!("got {} bytes from {}", nrecv, from);
-                        if i == 0 {
-                            socket.set_nonblocking(true)?;
-                        }
-                    }
                 }
+                Err(e) => {
+                    if e.kind() != io::ErrorKind::WouldBlock {
+                        info!("recv_from err {:?}", e);
+                    }
+                    re.recycle(r, "Blob::recv_from");
+                    return Err(Error::IO(e));
+                }
+                Ok(()) => if i == 0 {
+                    socket.set_nonblocking(true)?;
+                },
             }
             v.push(r);
         }