Define Connection::closed helper to await connection termination

quinn-proto/src/connection/assembler.rs

@@ -218,7 +218,7 @@ impl Assembler {
 }
 
 /// A chunk of data from the receive stream
-#[derive(Debug, PartialEq)]
+#[derive(Debug, PartialEq, Eq)]
 pub struct Chunk {
     /// The offset in the stream
     pub offset: u64,

quinn-proto/src/crypto/rustls.rs

@@ -354,7 +354,7 @@ impl crypto::PacketKey for PacketKey {
         payload: &mut BytesMut,
     ) -> Result<(), CryptoError> {
         let plain = self
-            .decrypt_in_place(packet, &*header, payload.as_mut())
+            .decrypt_in_place(packet, header, payload.as_mut())
             .map_err(|_| CryptoError)?;
         let plain_len = plain.len();
         payload.truncate(plain_len);

quinn-udp/src/unix.rs

@@ -212,7 +212,7 @@ fn send(
         unsafe { MaybeUninit::uninit().assume_init() };
     for (i, transmit) in transmits.iter().enumerate().take(BATCH_SIZE) {
         let dst_addr = unsafe {
-            std::ptr::write(
+            ptr::write(
                 addrs[i].as_mut_ptr(),
                 socket2::SockAddr::from(transmit.destination),
             );

quinn/src/connection.rs

@@ -366,6 +366,30 @@ impl Connection {
         }
     }
 
+    /// Wait for the connection to be closed for any reason
+    ///
+    /// Despite the return type's name, closed connections are often not an error condition at the
+    /// application layer. Cases that might be routine include [`ConnectionError::LocallyClosed`]
+    /// and [`ConnectionError::ApplicationClosed`].
+    pub async fn closed(&self) -> ConnectionError {
+        let closed;
+        {
+            let conn = self.0.lock("closed");
+            closed = conn.closed.clone();
+            // Construct the future while the lock is held to ensure we can't miss a wakeup if
+            // the `Notify` is signaled immediately after we release the lock. `await` it after
+            // the lock guard is out of scope.
+            closed.notified()
+        }
+        .await;
+        self.0
+            .lock("closed")
+            .error
+            .as_ref()
+            .expect("closed without an error")
+            .clone()
+    }
+
     /// Close the connection immediately.
     ///
     /// Pending operations will fail immediately with [`ConnectionError::LocallyClosed`]. Delivery
@@ -731,6 +755,7 @@ impl ConnectionRef {
             datagram_reader: None,
             finishing: FxHashMap::default(),
             stopped: FxHashMap::default(),
+            closed: Arc::new(Notify::new()),
             error: None,
             ref_count: 0,
             udp_state,
@@ -792,6 +817,7 @@ pub struct ConnectionInner {
     datagram_reader: Option<Waker>,
     pub(crate) finishing: FxHashMap<StreamId, oneshot::Sender<Option<WriteError>>>,
     pub(crate) stopped: FxHashMap<StreamId, Waker>,
+    closed: Arc<Notify>,
     /// Always set to Some before the connection becomes drained
     pub(crate) error: Option<ConnectionError>,
     /// Number of live handles that can be used to initiate or handle I/O; excludes the driver
@@ -1022,6 +1048,7 @@ impl ConnectionInner {
         for (_, waker) in self.stopped.drain() {
             waker.wake();
         }
+        self.closed.notify_waiters();
     }
 
     fn close(&mut self, error_code: VarInt, reason: Bytes) {