feat!: Client support for incoming messages and disconnections

The original assumption was that clients would communicate with servers
exclusively using bidirectional streams initiated by the client. This
was thought to be the only way to ensure that clients did not have to be
reachable or able to serve incoming connections.

However, QUIC allows any side of a connection to initiate streams,
meaning that a client-initiated connection could still receive incoming
streams from the server. The API as it stood did not support this, since
we would not give callers of `new_client` any way of receiving messages
sent to clients in this way.

Ultimately, the only technical limitation of clients is that they are
unable to handle incoming *connections* (and thus do not need to be
reachable or have server configuration), so the `new_client` function
now returns `IncomingMessages` and `DisconnectionEvents` since these
remain applicable.

BREAKING CHANGE: The signature of `Endpoint::new_client` has changed to
return a tuple of `(Endpoint, IncomingMessages, DisconnectionEvents)`
rather than just the `Endpoint`.

src/endpoint.rs

@@ -194,13 +194,17 @@ impl<I: ConnId> Endpoint<I> {
     pub fn new_client(
         local_addr: impl Into<SocketAddr>,
         config: Config,
-    ) -> Result<Self, ClientEndpointError> {
+    ) -> Result<(Self, IncomingMessages, DisconnectionEvents), ClientEndpointError> {
         let config = InternalConfig::try_from_config(config)?;
 
-        let (endpoint, _, _) =
+        let (endpoint, _, channels) =
             Self::build_endpoint(local_addr.into(), config, quinn::Endpoint::builder())?;
 
-        Ok(endpoint)
+        Ok((
+            endpoint,
+            IncomingMessages(channels.message.1),
+            DisconnectionEvents(channels.disconnection.1),
+        ))
     }
 
     // A private helper for initialising an endpoint.

src/tests/common.rs

@@ -673,13 +673,14 @@ async fn client() -> Result<()> {
     use crate::{Config, Endpoint};
 
     let (server, _, mut server_messages, _, _) = new_endpoint().await?;
-    let client = Endpoint::<[u8; 32]>::new_client(
-        local_addr(),
-        Config {
-            min_retry_duration: Some(Duration::from_millis(500)),
-            ..Default::default()
-        },
-    )?;
+    let (client, mut client_messages, mut client_disconnections) =
+        Endpoint::<[u8; 32]>::new_client(
+            local_addr(),
+            Config {
+                min_retry_duration: Some(Duration::from_millis(500)),
+                ..Default::default()
+            },
+        )?;
 
     client
         .send_message(b"hello"[..].into(), &server.public_addr(), 0)
@@ -692,5 +693,22 @@ async fn client() -> Result<()> {
     assert_eq!(sender, client.public_addr());
     assert_eq!(&message[..], b"hello");
 
+    server
+        .send_message(b"world"[..].into(), &client.public_addr(), 0)
+        .await?;
+    let (sender, message) = client_messages
+        .next()
+        .await
+        .ok_or_else(|| eyre!("Did not receive expected message"))?;
+    assert_eq!(sender, server.public_addr());
+    assert_eq!(&message[..], b"world");
+
+    server.disconnect_from(&client.public_addr()).await;
+    let disconnector = client_disconnections
+        .next()
+        .await
+        .ok_or_else(|| eyre!("Did not receive expected disconnection"))?;
+    assert_eq!(disconnector, server.public_addr());
+
     Ok(())
 }