Merge main and fix up

Author: rklaehn

src/transport/boxed.rs

@@ -421,9 +421,9 @@ impl<S: Service> BoxableConnection<S::Res, S::Req> for super::tokio_mpsc::Connec
         Box::new(self.clone())
     }
 
-    fn open_bi_boxed(&self) -> OpenFuture<S::Res, S::Req> {
+    fn open_boxed(&self) -> OpenFuture<S::Res, S::Req> {
         let f = Box::pin(async move {
-            let (send, recv) = super::Connection::open_bi(self).await?;
+            let (send, recv) = super::Connection::open(self).await?;
             // return the boxed streams
             anyhow::Ok((
                 SendSink::direct_tokio(send.0),
@@ -442,7 +442,7 @@ impl<S: Service> BoxableServerEndpoint<S::Req, S::Res> for super::tokio_mpsc::Se
 
     fn accept_bi_boxed(&self) -> AcceptFuture<S::Req, S::Res> {
         let f = async move {
-            let (send, recv) = super::ServerEndpoint::accept_bi(self).await?;
+            let (send, recv) = super::ServerEndpoint::accept(self).await?;
             let send = send.sink_map_err(anyhow::Error::from);
             let recv = recv.map_err(anyhow::Error::from);
             anyhow::Ok((SendSink::boxed(send), RecvStream::boxed(recv)))

src/transport/tokio_mpsc.rs

@@ -137,7 +137,7 @@ impl<S: Service> ConnectionCommon<S::Req, S::Res> for ServerEndpoint<S> {
 }
 
 impl<S: Service> transport::ServerEndpoint<S::Req, S::Res> for ServerEndpoint<S> {
-    async fn accept_bi(&self) -> Result<(Self::SendSink, Self::RecvStream), AcceptBiError> {
+    async fn accept(&self) -> Result<(Self::SendSink, Self::RecvStream), AcceptBiError> {
         let (send, recv) = self
             .stream
             .lock()
@@ -167,7 +167,7 @@ impl<S: Service> ConnectionCommon<S::Res, S::Req> for Connection<S> {
 }
 
 impl<S: Service> transport::Connection<S::Res, S::Req> for Connection<S> {
-    async fn open_bi(&self) -> result::Result<Socket<S::Res, S::Req>, self::OpenBiError> {
+    async fn open(&self) -> result::Result<Socket<S::Res, S::Req>, self::OpenBiError> {
         let (local_send, remote_recv) = mpsc::channel::<S::Req>(128);
         let (remote_send, local_recv) = mpsc::channel::<S::Res>(128);
         let remote_chan = (

tests/tokio_mpsc.rs

@@ -9,7 +9,7 @@ use quic_rpc::{
 };
 
 #[tokio::test]
-async fn async_channel_channel_bench() -> anyhow::Result<()> {
+async fn tokio_mpsc_channel_bench() -> anyhow::Result<()> {
     tracing_subscriber::fmt::try_init().ok();
     let (server, client) = tokio_mpsc::connection::<ComputeService>(1);
 
@@ -26,7 +26,7 @@ async fn async_channel_channel_bench() -> anyhow::Result<()> {
 }
 
 #[tokio::test]
-async fn async_channel_channel_mapped_bench() -> anyhow::Result<()> {
+async fn tokio_mpsc_channel_mapped_bench() -> anyhow::Result<()> {
     use derive_more::{From, TryInto};
     use serde::{Deserialize, Serialize};
 
@@ -67,7 +67,7 @@ async fn async_channel_channel_mapped_bench() -> anyhow::Result<()> {
         tokio::task::spawn(async move {
             let service = ComputeService;
             loop {
-                let (req, chan) = server.accept().await?;
+                let (req, chan) = server.accept().await?.read_first().await?;
                 let service = service.clone();
                 tokio::spawn(async move {
                     let req: OuterRequest = req;