server: 🌽 listen_channel for in-memory connections

this introduces a `Server::listen_channel` interface. this is like the
existing `listen_tcp` or `listen_unix` methods, but instead applies to
an in-memory channel of `(read, write)` tuples.

this is motivated by work like penumbra-zone/penumbra#3588. to summarize
the motivation, it would be nice if we had a way to run a tower-abci
service in cargo tests, without needing to bind to an actual port, or
create a unix-domain socket within a temporary directory.

src/v037/server.rs

@@ -184,6 +184,30 @@ where
             }
         }
     }
+
+    pub async fn listen_channel<R, W>(self, mut channel: tokio::sync::mpsc::Receiver<(R, W)>)
+    where
+        R: AsyncReadExt + std::marker::Unpin + Send + 'static,
+        W: AsyncWriteExt + std::marker::Unpin + Send + 'static,
+    {
+        tracing::info!("ABCI server starting on tokio channel");
+
+        match channel.recv().await {
+            Some((read, write)) => {
+                tracing::debug!("accepted new connection");
+                let conn = Connection {
+                    consensus: self.consensus.clone(),
+                    mempool: self.mempool.clone(),
+                    info: self.info.clone(),
+                    snapshot: self.snapshot.clone(),
+                };
+                tokio::spawn(async move { conn.run(read, write).await.unwrap() });
+            }
+            None => {
+                tracing::trace!("channel closed");
+            }
+        }
+    }
 }
 
 struct Connection<C, M, I, S> {