v2.3.0: add crash with local multiplier and pushed crash event

1 files changed, 57 insertions, 1 deletions

diff --git a/src/bin/client/tls.rs b/src/bin/client/tls.rs
index 8c052f7..b46f52b 100644
--- a/src/bin/client/tls.rs
+++ b/src/bin/client/tls.rs
@@ -64,6 +64,7 @@ impl TlsConnector {
                 sock,
                 tls: conn,
                 lines: LineBuffer::new(),
+                read_timeout: Some(std::time::Duration::from_secs(10)),
             })),
         })
     }
@@ -150,6 +151,7 @@ struct Session {
     sock: TcpStream,
     tls: rustls::ClientConnection,
     lines: LineBuffer,
+    read_timeout: Option<std::time::Duration>,
 }
 
 impl Session {
@@ -202,7 +204,61 @@ impl TlsStream {
         &self,
         duration: Option<std::time::Duration>,
     ) -> std::io::Result<()> {
-        self.session.lock().sock.set_read_timeout(duration)
+        let mut s = self.session.lock();
+        s.read_timeout = duration;
+        s.sock.set_read_timeout(duration)
+    }
+
+    /// Return one complete line if the socket already has one, without
+    /// blocking. Used to pick up server pushes between requests.
+    pub fn read_available_line(&mut self) -> std::io::Result<Option<String>> {
+        let mut s = self.session.lock();
+        if let Some(line) = s.lines.next_line() {
+            return Ok(Some(line));
+        }
+
+        // `set_read_timeout(Some(0))` is rejected by std, so poll with the
+        // shortest timeout the socket accepts instead of a nonblocking mode.
+        s.sock
+            .set_read_timeout(Some(std::time::Duration::from_millis(1)))?;
+        let drained = (|| -> std::io::Result<()> {
+            loop {
+                let mut buf = [0u8; 4096];
+                match s.tls.reader().read(&mut buf) {
+                    Ok(0) => return Ok(()),
+                    Ok(n) => s.lines.push(&buf[..n]),
+                    // The reader only drains plaintext rustls has already
+                    // decrypted; `drive` moves bytes off the socket.
+                    Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
+                        match s.drive() {
+                            Ok(()) => continue,
+                            Err(error)
+                                if matches!(
+                                    error.kind(),
+                                    std::io::ErrorKind::WouldBlock
+                                        | std::io::ErrorKind::TimedOut
+                                ) =>
+                            {
+                                return Ok(());
+                            }
+                            Err(error) => return Err(error),
+                        }
+                    }
+                    Err(error)
+                        if matches!(
+                            error.kind(),
+                            std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut
+                        ) =>
+                    {
+                        return Ok(());
+                    }
+                    Err(error) => return Err(error),
+                }
+            }
+        })();
+        s.sock.set_read_timeout(s.read_timeout)?;
+        drained?;
+        Ok(s.lines.next_line())
     }
 }