merge: feat-lints into main

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

@@ -54,10 +54,9 @@
 //!   as a closed outbound channel.
 //! - [`Bi::recv`] returns `None` when the underlying transport closes.
 //! - Message translation is intentionally out of scope for this module.
-
-use std::marker::PhantomData;
-
 use async_trait::async_trait;
+use kameo::{error::Infallible, prelude::*};
+use std::marker::PhantomData;
 
 /// Errors returned by transport adapters implementing [`Bi`].
 #[derive(thiserror::Error, Debug)]
@@ -106,6 +105,36 @@ pub trait Receiver<Inbound>: Send + Sync {
 /// any built-in correlation mechanism between inbound and outbound items.
 pub trait Bi<Inbound, Outbound>: Sender<Outbound> + Receiver<Inbound> + Send + Sync {}
 
+#[async_trait]
+impl<T, Outbound> Sender<Outbound> for &mut T
+where
+    T: Sender<Outbound> + ?Sized,
+    Outbound: Send + 'static,
+{
+    async fn send(&mut self, item: Outbound) -> Result<(), Error> {
+        (**self).send(item).await
+    }
+}
+
+#[async_trait]
+impl<T, Inbound> Receiver<Inbound> for &mut T
+where
+    T: Receiver<Inbound> + ?Sized,
+    Inbound: Send + 'static,
+{
+    async fn recv(&mut self) -> Option<Inbound> {
+        (**self).recv().await
+    }
+}
+
+impl<T, Inbound, Outbound> Bi<Inbound, Outbound> for &mut T
+where
+    T: Bi<Inbound, Outbound> + ?Sized,
+    Inbound: Send + 'static,
+    Outbound: Send + 'static,
+{
+}
+
 pub trait SplittableBi<Inbound, Outbound>: Bi<Inbound, Outbound> {
     type Sender: Sender<Outbound>;
     type Receiver: Receiver<Inbound>;
@@ -161,3 +190,29 @@ where
 }
 
 pub mod grpc;
+
+#[derive(thiserror::Error, Debug)]
+pub enum ForwardError<I> {
+    #[error("Transport error: {0}")]
+    Transport(#[from] Error),
+    #[error("Actor delivery error: {0}")]
+    Actor(SendError<I>),
+}
+
+pub async fn forward_to_actor<Transport, Inbound, Outbound, Handler>(
+    transport: &mut Transport,
+    actor: &ActorRef<Handler>,
+) -> Result<(), ForwardError<Inbound>>
+where
+    Transport: Bi<Inbound, <Outbound as Reply>::Ok>,
+    Handler: Actor + Message<Inbound, Reply = Outbound>,
+    Inbound: Send + 'static,
+    Outbound: Send + 'static + Reply<Error = Infallible>, // `Infallible` to enforce contract that `Outbound` carries handler-level error
+{
+    while let Some(request) = transport.recv().await {
+        let resp = actor.ask(request).await.map_err(ForwardError::Actor)?;
+        transport.send(resp).await?
+    }
+
+    Err(Error::ChannelClosed.into())
+}

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

@@ -1,10 +1,10 @@
+use super::{Bi, Receiver, Sender};
+
 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>>,
 }

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

@@ -1,7 +1,6 @@
-use std::fmt::Display;
-
 use base64::{Engine as _, prelude::BASE64_URL_SAFE};
 use rustls_pki_types::CertificateDer;
+use std::fmt::Display;
 
 const ARBITER_URL_SCHEME: &str = "arbiter";
 const CERT_QUERY_KEY: &str = "cert";