Always store open frames to internal buffer. (#319)

mplex/Cargo.toml

@@ -4,19 +4,11 @@ version = "0.1.0"
 authors = ["Parity Technologies <admin@parity.io>"]
 
 [dependencies]
-arrayvec = "0.4.6"
 bytes = "0.4.5"
-circular-buffer = { path = "../circular-buffer" }
-error-chain = "0.11.0"
 fnv = "1.0"
 futures = "0.1"
-futures-mutex = { git = "https://github.com/paritytech/futures-mutex" }
 libp2p-core = { path = "../core" }
-log = "0.4"
-num-bigint = { version = "0.1.40", default-features = false }
-num-traits = "0.1.40"
 parking_lot = "0.4.8"
-rand = "0.3.17"
 tokio-codec = "0.1"
 tokio-io = "0.1"
 varint = { path = "../varint-rs" }

mplex/src/codec.rs

@@ -49,6 +49,14 @@ impl Elem {
             Elem::Reset { substream_id, .. } => substream_id,
         }
     }
+
+    pub fn is_open_msg(&self) -> bool {
+        if let Elem::Open { .. } = self {
+            true
+        } else {
+            false
+        }
+    }
 }
 
 pub struct Codec {

mplex/src/lib.rs

@@ -23,7 +23,6 @@ extern crate fnv;
 #[macro_use]
 extern crate futures;
 extern crate libp2p_core as core;
-extern crate log;
 extern crate parking_lot;
 extern crate tokio_codec;
 extern crate tokio_io;
@@ -154,7 +153,7 @@ where C: AsyncRead + AsyncWrite,
                     return Err(IoError::new(IoErrorKind::InvalidData, "reached maximum buffer length"));
                 }
 
-                if inner.opened_substreams.contains(&elem.substream_id()) {
+                if inner.opened_substreams.contains(&elem.substream_id()) || elem.is_open_msg() {
                     inner.buffer.push(elem);
                     for task in inner.to_notify.drain(..) {
                         task.notify();