wip

ntex-rs · Nov 17, 2023 · 8f9de78 · 8f9de78
1 parent eca6915
commit 8f9de78
Showing 1 changed file with 44 additions and 37 deletions.
diff --git a/src/io.rs b/src/io.rs
@@ -256,48 +256,55 @@ where
                             if inner.io.is_closed() {
                                 log::trace!("io has been closed, stop dispatcher");
                                 inner.st = IoDispatcherState::Stop;
-                                continue;
-                            }
-
-                            // decode incoming bytes stream
-                            match inner.io.poll_recv_decode(this.codec, cx) {
-                                Ok(decoded) => {
-                                    inner.update_timer(&decoded);
-                                    if let Some(el) = decoded.item {
-                                        DispatchItem::Item(el)
-                                    } else {
-                                        return Poll::Pending;
-                                    }
-                                }
-                                Err(RecvError::Stop) => {
-                                    log::trace!("dispatcher is instructed to stop");
-                                    inner.st = IoDispatcherState::Stop;
-                                    continue;
+                                if let Poll::Ready(IoStatusUpdate::PeerGone(err)) =
+                                    inner.io.poll_status_update(cx)
+                                {
+                                    DispatchItem::Disconnect(err)
+                                } else {
+                                    DispatchItem::Disconnect(None)
                                 }
-                                Err(RecvError::KeepAlive) => {
-                                    log::trace!("keep-alive error, stopping dispatcher");
-                                    inner.st = IoDispatcherState::Stop;
-                                    if inner.flags.contains(Flags::READ_TIMEOUT) {
-                                        DispatchItem::ReadTimeout
-                                    } else {
-                                        DispatchItem::KeepAliveTimeout
+                            } else {
+                                // decode incoming bytes stream
+                                match inner.io.poll_recv_decode(this.codec, cx) {
+                                    Ok(decoded) => {
+                                        inner.update_timer(&decoded);
+                                        if let Some(el) = decoded.item {
+                                            DispatchItem::Item(el)
+                                        } else {
+                                            return Poll::Pending;
+                                        }
                                     }
-                                }
-                                Err(RecvError::WriteBackpressure) => {
-                                    if let Err(err) = ready!(inner.io.poll_flush(cx, false)) {
+                                    Err(RecvError::Stop) => {
+                                        log::trace!("dispatcher is instructed to stop");
                                         inner.st = IoDispatcherState::Stop;
-                                        DispatchItem::Disconnect(Some(err))
-                                    } else {
                                         continue;
                                     }
-                                }
-                                Err(RecvError::Decoder(err)) => {
-                                    inner.st = IoDispatcherState::Stop;
-                                    DispatchItem::DecoderError(err)
-                                }
-                                Err(RecvError::PeerGone(err)) => {
-                                    inner.st = IoDispatcherState::Stop;
-                                    DispatchItem::Disconnect(err)
+                                    Err(RecvError::KeepAlive) => {
+                                        log::trace!("keep-alive error, stopping dispatcher");
+                                        inner.st = IoDispatcherState::Stop;
+                                        if inner.flags.contains(Flags::READ_TIMEOUT) {
+                                            DispatchItem::ReadTimeout
+                                        } else {
+                                            DispatchItem::KeepAliveTimeout
+                                        }
+                                    }
+                                    Err(RecvError::WriteBackpressure) => {
+                                        if let Err(err) = ready!(inner.io.poll_flush(cx, false))
+                                        {
+                                            inner.st = IoDispatcherState::Stop;
+                                            DispatchItem::Disconnect(Some(err))
+                                        } else {
+                                            continue;
+                                        }
+                                    }
+                                    Err(RecvError::Decoder(err)) => {
+                                        inner.st = IoDispatcherState::Stop;
+                                        DispatchItem::DecoderError(err)
+                                    }
+                                    Err(RecvError::PeerGone(err)) => {
+                                        inner.st = IoDispatcherState::Stop;
+                                        DispatchItem::Disconnect(err)
+                                    }
                                 }
                             }
                         }