reactor
/
JKHKQKJV
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor

main
_ 2025-02-23 19:30:26 -06:00
parent 4d018413f8
commit 425503d66d
1 changed files with 22 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
src/client.rs
View File

@ -42,36 +42,32 @@ impl App {
fn step(&mut self, cx: &mut Context<'_>) -> Result<()> { fn step(&mut self, cx: &mut Context<'_>) -> Result<()> {
let mut stream = pin!(&mut self.stream); let mut stream = pin!(&mut self.stream);
match <_ as futures_sink::Sink<Bytes>>::poll_ready(stream.as_mut(), cx) { if let Poll::Ready(result) =
Poll::Pending => {} <_ as futures_sink::Sink<Bytes>>::poll_ready(stream.as_mut(), cx)
Poll::Ready(result) => { {
result?; result?;
if let Some(frame) = self.client.poll_send() { if let Some(frame) = self.client.poll_send() {
stream stream
.as_mut() .as_mut()
.start_send(frame) .start_send(frame)
.context("stream.start_send")?; .context("stream.start_send")?;
tracing::debug!("Started send"); tracing::debug!("Started send");
} }
match <_ as futures_sink::Sink<Bytes>>::poll_flush(stream.as_mut(), cx) { match <_ as futures_sink::Sink<Bytes>>::poll_flush(stream.as_mut(), cx) {
Poll::Pending => {} Poll::Pending => {}
Poll::Ready(result) => { Poll::Ready(result) => {
result.context("poll_flush")?; result.context("poll_flush")?;
}
} }
} }
} }
match stream.as_mut().poll_next(cx) { if let Poll::Ready(frame_opt) = stream.as_mut().poll_next(cx) {
Poll::Pending => {} let frame = frame_opt.context("Server closed cxn")?;
Poll::Ready(frame_opt) => { cx.waker().wake_by_ref();
let frame = frame_opt.context("Server closed cxn")?; let frame = frame.context("network framing decode")?;
cx.waker().wake_by_ref(); self.client
let frame = frame.context("network framing decode")?; .handle_frame(frame.into())
self.client .context("client.handle_frame")?;
.handle_frame(frame.into())
.context("client.handle_frame")?;
}
} }
if self.timer.poll_tick(cx).is_ready() { if self.timer.poll_tick(cx).is_ready() {