Assert that stream state is released

This patch adds a debug level assertion checking that state associated
with streams is successfully released. This is done by ensuring that no
stream state remains when the connection inner state is dropped. This
does not happen until all handles to the connection drops, which should
result in all streams being dropped.

This also pulls in a fix for one bug that is exposed with this change.
The fix is first proposed as part of #261.

src/proto/streams/prioritize.rs

@@ -613,21 +613,22 @@ impl Prioritize {
                     trace!("pop_frame; stream={:?}; stream.state={:?}",
                         stream.id, stream.state);
 
+                    // It's possible that this stream, besides having data to send,
+                    // is also queued to send a reset, and thus is already in the queue
+                    // to wait for "some time" after a reset.
+                    //
+                    // To be safe, we just always ask the stream.
+                    let is_pending_reset = stream.is_pending_reset_expiration();
+
                     // If the stream receives a RESET from the peer, it may have
                     // had data buffered to be sent, but all the frames are cleared
                     // in clear_queue(). Instead of doing O(N) traversal through queue
                     // to remove, lets just ignore peer_reset streams here.
                     if stream.state.is_peer_reset() {
+                        counts.transition_after(stream, is_pending_reset);
                         continue;
                     }
 
-                    // It's possible that this stream, besides having data to send,
-                    // is also queued to send a reset, and thus is already in the queue
-                    // to wait for "some time" after a reset.
-                    //
-                    // To be safe, we just always ask the stream.
-                    let is_pending_reset = stream.is_pending_reset_expiration();
-
                     trace!(" --> stream={:?}; is_pending_reset={:?};",
                         stream.id, is_pending_reset);
 

src/proto/streams/store.rs

@@ -85,6 +85,10 @@ impl Store {
         }
     }
 
+    pub fn is_empty(&self) -> bool {
+        self.slab.is_empty()
+    }
+
     pub fn find_mut(&mut self, id: &StreamId) -> Option<Ptr> {
         let key = match self.ids.get(id) {
             Some(key) => *key,

src/proto/streams/streams.rs

@@ -1099,3 +1099,13 @@ impl Actions {
         }
     }
 }
+
+impl Drop for Inner {
+    fn drop(&mut self) {
+        use std::thread;
+
+        if !thread::panicking() {
+            debug_assert!(self.store.is_empty());
+        }
+    }
+}