bet

crates/net/eth-wire/src/multiplex.rs

@@ -19,6 +19,7 @@ use std::{
 use crate::{
     capability::{SharedCapabilities, SharedCapability, UnsupportedCapabilityError},
     errors::{EthStreamError, P2PStreamError},
+    p2pstream::DisconnectP2P,
     CanDisconnect, Capability, DisconnectReason, EthStream, P2PStream, Status, UnauthedEthStream,
 };
 use bytes::{Bytes, BytesMut};
@@ -464,6 +465,29 @@ where
                 return Poll::Ready(Some(msg))
             }
 
+            let mut conn_ready = true;
+
+            match this.inner.conn.poll_ready_unpin(cx) {
+                Poll::Ready(Ok(())) => {
+                    if let Some(msg) = this.inner.out_buffer.pop_front() {
+                        if let Err(err) = this.inner.conn.start_send_unpin(msg) {
+                            return Poll::Ready(Some(Err(err.into())))
+                        }
+                    }
+                }
+                Poll::Ready(Err(err)) => {
+                    if let Err(disconnect_err) =
+                        this.inner.conn.start_disconnect(DisconnectReason::DisconnectRequested)
+                    {
+                        return Poll::Ready(Some(Err(disconnect_err.into())))
+                    }
+                    return Poll::Ready(Some(Err(err.into())))
+                }
+                Poll::Pending => {
+                    conn_ready = false;
+                }
+            }
+
             // advance primary out
             loop {
                 match this.primary.from_primary.poll_next_unpin(cx) {
@@ -541,7 +565,7 @@ where
                 }
             }
 
-            if !delegated && this.inner.out_buffer.is_empty() {
+            if !conn_ready || (!delegated && this.inner.out_buffer.is_empty()) {
                 return Poll::Pending
             }
         }