refactor(server::useragent): migrated to new connection design

server/crates/arbiter-proto/Cargo.toml

@@ -21,6 +21,7 @@ base64 = "0.22.1"
 prost-types.workspace = true
 tracing.workspace = true
 async-trait.workspace = true
+tokio-stream.workspace = true
 
 [build-dependencies]
 tonic-prost-build = "0.14.3"

server/crates/arbiter-proto/src/transport.rs

@@ -63,16 +63,29 @@ where
     extractor(msg).ok_or(Error::UnexpectedMessage)
 }
 
+#[async_trait]
+pub trait Sender<Outbound>: Send + Sync {
+    async fn send(&mut self, item: Outbound) -> Result<(), Error>;
+}
+
+#[async_trait]
+pub trait Receiver<Inbound>: Send + Sync  {
+    async fn recv(&mut self) -> Option<Inbound>;
+}
+
 /// Minimal bidirectional transport abstraction used by protocol code.
 ///
 /// `Bi<Inbound, Outbound>` models a duplex channel with:
 /// - inbound items of type `Inbound` read via [`Bi::recv`]
 /// - outbound items of type `Outbound` written via [`Bi::send`]
-#[async_trait]
-pub trait Bi<Inbound, Outbound>: Send + Sync + 'static {
-    async fn send(&mut self, item: Outbound) -> Result<(), Error>;
+pub trait Bi<Inbound, Outbound>: Sender<Outbound> + Receiver<Inbound> + Send + Sync {}
 
-    async fn recv(&mut self) -> Option<Inbound>;
+pub trait SplittableBi<Inbound, Outbound>: Bi<Inbound, Outbound> {
+    type Sender: Sender<Outbound>;
+    type Receiver: Receiver<Inbound>;
+
+    fn split(self) -> (Self::Sender, Self::Receiver);
+    fn from_parts(sender: Self::Sender, receiver: Self::Receiver) -> Self;
 }
 
 /// No-op [`Bi`] transport for tests and manual actor usage.
@@ -83,22 +96,16 @@ pub struct DummyTransport<Inbound, Outbound> {
     _marker: PhantomData<(Inbound, Outbound)>,
 }
 
-impl<Inbound, Outbound> DummyTransport<Inbound, Outbound> {
-    pub fn new() -> Self {
+impl<Inbound, Outbound> Default for DummyTransport<Inbound, Outbound> {
+    fn default() -> Self {
         Self {
             _marker: PhantomData,
         }
     }
 }
 
-impl<Inbound, Outbound> Default for DummyTransport<Inbound, Outbound> {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
 #[async_trait]
-impl<Inbound, Outbound> Bi<Inbound, Outbound> for DummyTransport<Inbound, Outbound>
+impl<Inbound, Outbound> Sender<Outbound> for DummyTransport<Inbound, Outbound>
 where
     Inbound: Send + Sync + 'static,
     Outbound: Send + Sync + 'static,
@@ -106,9 +113,25 @@ where
     async fn send(&mut self, _item: Outbound) -> Result<(), Error> {
         Ok(())
     }
+}
 
+#[async_trait]
+impl<Inbound, Outbound> Receiver<Inbound> for DummyTransport<Inbound, Outbound>
+where
+    Inbound: Send + Sync + 'static,
+    Outbound: Send + Sync + 'static,
+{
     async fn recv(&mut self) -> Option<Inbound> {
         std::future::pending::<()>().await;
         None
     }
 }
+
+impl<Inbound, Outbound> Bi<Inbound, Outbound> for DummyTransport<Inbound, Outbound>
+where
+    Inbound: Send + Sync + 'static,
+    Outbound: Send + Sync + 'static,
+{
+}
+
+pub mod grpc;

server/crates/arbiter-proto/src/transport/grpc.rs

@@ -0,0 +1,106 @@
+use async_trait::async_trait;
+use futures::StreamExt;
+use tokio::sync::mpsc;
+use tokio_stream::wrappers::ReceiverStream;
+
+use super::{Bi, Receiver, Sender};
+
+pub struct GrpcSender<Outbound> {
+    tx: mpsc::Sender<Result<Outbound, tonic::Status>>,
+}
+
+#[async_trait]
+impl<Outbound> Sender<Result<Outbound, tonic::Status>> for GrpcSender<Outbound>
+where
+    Outbound: Send + Sync + 'static,
+{
+    async fn send(&mut self, item: Result<Outbound, tonic::Status>) -> Result<(), super::Error> {
+        self.tx
+            .send(item)
+            .await
+            .map_err(|_| super::Error::ChannelClosed)
+    }
+}
+
+pub struct GrpcReceiver<Inbound> {
+    rx: tonic::Streaming<Inbound>,
+}
+#[async_trait]
+impl<Inbound> Receiver<Result<Inbound, tonic::Status>> for GrpcReceiver<Inbound>
+where
+    Inbound: Send + Sync + 'static,
+{
+    async fn recv(&mut self) -> Option<Result<Inbound, tonic::Status>> {
+        self.rx.next().await
+    }
+}
+
+pub struct GrpcBi<Inbound, Outbound> {
+    sender: GrpcSender<Outbound>,
+    receiver: GrpcReceiver<Inbound>,
+}
+
+impl<Inbound, Outbound> GrpcBi<Inbound, Outbound>
+where
+    Inbound: Send + Sync + 'static,
+    Outbound: Send + Sync + 'static,
+{
+    pub fn from_bi_stream(
+        receiver: tonic::Streaming<Inbound>,
+    ) -> (Self, ReceiverStream<Result<Outbound, tonic::Status>>) {
+        let (tx, rx) = mpsc::channel(10);
+        let sender = GrpcSender { tx };
+        let receiver = GrpcReceiver { rx: receiver };
+        let bi = GrpcBi { sender, receiver };
+        (bi, ReceiverStream::new(rx))
+    }
+}
+
+#[async_trait]
+impl<Inbound, Outbound> Sender<Result<Outbound, tonic::Status>> for GrpcBi<Inbound, Outbound>
+where
+    Inbound: Send + Sync + 'static,
+    Outbound: Send + Sync + 'static,
+{
+    async fn send(&mut self, item: Result<Outbound, tonic::Status>) -> Result<(), super::Error> {
+        self.sender.send(item).await
+    }
+}
+
+#[async_trait]
+impl<Inbound, Outbound> Receiver<Result<Inbound, tonic::Status>> for GrpcBi<Inbound, Outbound>
+where
+    Inbound: Send + Sync + 'static,
+    Outbound: Send + Sync + 'static,
+{
+    async fn recv(&mut self) -> Option<Result<Inbound, tonic::Status>> {
+        self.receiver.recv().await
+    }
+}
+
+impl<Inbound, Outbound> Bi<Result<Inbound, tonic::Status>, Result<Outbound, tonic::Status>>
+    for GrpcBi<Inbound, Outbound>
+where
+    Inbound: Send + Sync + 'static,
+    Outbound: Send + Sync + 'static,
+{
+}
+
+impl<Inbound, Outbound>
+    super::SplittableBi<Result<Inbound, tonic::Status>, Result<Outbound, tonic::Status>>
+    for GrpcBi<Inbound, Outbound>
+where
+    Inbound: Send + Sync + 'static,
+    Outbound: Send + Sync + 'static,
+{
+    type Sender = GrpcSender<Outbound>;
+    type Receiver = GrpcReceiver<Inbound>;
+
+    fn split(self) -> (Self::Sender, Self::Receiver) {
+        (self.sender, self.receiver)
+    }
+
+    fn from_parts(sender: Self::Sender, receiver: Self::Receiver) -> Self {
+        GrpcBi { sender, receiver }
+    }
+}