Deprecate keyed message boxes

Instead of using an array of clients identified by their indexes,
the client address is attached to the requests.

Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>

Author: didier-wenzek

crates/core/tedge_actors/src/message_boxes.rs

@@ -235,7 +235,7 @@ impl<Output> LoggingSender<Output> {
 }
 
 #[async_trait]
-impl<Output: Message + Debug> Sender<Output> for LoggingSender<Output> {
+impl<Output: Message> Sender<Output> for LoggingSender<Output> {
     async fn send(&mut self, message: Output) -> Result<(), ChannelError> {
         log_message_sent(&self.name, &message);
         self.sender.send(message).await

crates/core/tedge_actors/src/servers/actors.rs

@@ -1,6 +1,7 @@
 use crate::Actor;
 use crate::ConcurrentServerMessageBox;
 use crate::MessageReceiver;
+use crate::RequestEnvelope;
 use crate::RuntimeError;
 use crate::RuntimeRequest;
 use crate::Sender;
@@ -13,12 +14,12 @@ use async_trait::async_trait;
 /// Requests are processed in turn, leading either to a response or an error.
 pub struct ServerActor<S: Server> {
     server: S,
-    messages: ServerMessageBox<S::Request, S::Response>,
+    requests: ServerMessageBox<S::Request, S::Response>,
 }
 
 impl<S: Server> ServerActor<S> {
-    pub fn new(server: S, messages: ServerMessageBox<S::Request, S::Response>) -> Self {
-        ServerActor { server, messages }
+    pub fn new(server: S, requests: ServerMessageBox<S::Request, S::Response>) -> Self {
+        ServerActor { server, requests }
     }
 }
 
@@ -30,12 +31,17 @@ impl<S: Server> Actor for ServerActor<S> {
 
     async fn run(mut self) -> Result<(), RuntimeError> {
         let server = &mut self.server;
-        while let Some((client_id, request)) = self.messages.recv().await {
+        while let Some(RequestEnvelope {
+            request,
+            mut reply_to,
+        }) = self.requests.recv().await
+        {
             tokio::select! {
-                result = server.handle(request) => {
-                    self.messages.send((client_id, result)).await?
+                response = server.handle(request) => {
+                    // Ignore errors on send: the requester is simply no more expecting a response
+                    let _ = reply_to.send(response).await;
                 }
-                Some(RuntimeRequest::Shutdown) = self.messages.recv_signal() => {
+                Some(RuntimeRequest::Shutdown) = self.requests.recv_signal() => {
                     break;
                 }
             }
@@ -67,16 +73,21 @@ impl<S: Server + Clone> Actor for ConcurrentServerActor<S> {
     }
 
     async fn run(mut self) -> Result<(), RuntimeError> {
-        while let Some((client_id, request)) = self.messages.recv().await {
+        while let Some(RequestEnvelope {
+            request,
+            mut reply_to,
+        }) = self.messages.next_request().await
+        {
             // Spawn the request
             let mut server = self.server.clone();
-            let pending_result = tokio::spawn(async move {
+            let request_handler = tokio::spawn(async move {
                 let result = server.handle(request).await;
-                (client_id, result)
+                // Ignore errors on send: the requester is simply no more expecting a response
+                let _ = reply_to.send(result).await;
             });
 
             // Send the response back to the client
-            self.messages.send_response_once_done(pending_result)
+            self.messages.register_request_handler(request_handler)
         }
 
         Ok(())

crates/core/tedge_actors/src/servers/builders.rs

@@ -1,36 +1,33 @@
 use crate::mpsc;
 use crate::Actor;
 use crate::Builder;
-use crate::ClientId;
 use crate::CloneSender;
 use crate::ConcurrentServerActor;
 use crate::ConcurrentServerMessageBox;
+use crate::DynRequestSender;
 use crate::DynSender;
-use crate::KeyedSender;
 use crate::LoggingReceiver;
-use crate::LoggingSender;
 use crate::Message;
+use crate::MessageSink;
 use crate::NoConfig;
+use crate::RequestEnvelope;
+use crate::RequestSender;
 use crate::RuntimeError;
 use crate::RuntimeRequest;
 use crate::RuntimeRequestSink;
-use crate::SenderVec;
 use crate::Server;
 use crate::ServerActor;
 use crate::ServerMessageBox;
 use crate::ServiceProvider;
-use crate::SimpleMessageBox;
 use std::convert::Infallible;
 use std::fmt::Debug;
 
 /// A message box builder for request-response services
 pub struct ServerMessageBoxBuilder<Request: Debug, Response> {
-    service_name: String,
     max_concurrency: usize,
-    request_sender: mpsc::Sender<(ClientId, Request)>,
-    input_receiver: LoggingReceiver<(ClientId, Request)>,
+    request_sender: mpsc::Sender<RequestEnvelope<Request, Response>>,
+    request_receiver: LoggingReceiver<RequestEnvelope<Request, Response>>,
     signal_sender: mpsc::Sender<RuntimeRequest>,
-    clients: Vec<DynSender<Response>>,
 }
 
 impl<Request: Message, Response: Message> ServerMessageBoxBuilder<Request, Response> {
@@ -39,16 +36,14 @@ impl<Request: Message, Response: Message> ServerMessageBoxBuilder<Request, Respo
         let max_concurrency = 1;
         let (request_sender, request_receiver) = mpsc::channel(capacity);
         let (signal_sender, signal_receiver) = mpsc::channel(4);
-        let input_receiver =
+        let request_receiver =
             LoggingReceiver::new(server_name.to_string(), request_receiver, signal_receiver);
 
         ServerMessageBoxBuilder {
-            service_name: server_name.to_string(),
             max_concurrency,
             request_sender,
-            input_receiver,
+            request_receiver,
             signal_sender,
-            clients: vec![],
         }
     }
 
@@ -59,19 +54,19 @@ impl<Request: Message, Response: Message> ServerMessageBoxBuilder<Request, Respo
         }
     }
 
+    /// Return a sender for the requests
+    pub fn request_sender(&self) -> DynRequestSender<Request, Response> {
+        self.request_sender.sender_clone()
+    }
+
     /// Build a message box ready to be used by the server actor
     fn build_server(self) -> ServerMessageBox<Request, Response> {
-        let response_sender = SenderVec::new_sender(self.clients);
-        let logging_sender = LoggingSender::new(self.service_name.clone(), response_sender);
-
-        SimpleMessageBox::new(self.input_receiver, logging_sender)
+        self.request_receiver
     }
 
     /// Build a message box aimed to concurrently serve requests
     fn build_concurrent(self) -> ConcurrentServerMessageBox<Request, Response> {
-        let max_concurrency = self.max_concurrency;
-        let clients = self.build_server();
-        ConcurrentServerMessageBox::new(max_concurrency, clients)
+        ConcurrentServerMessageBox::new(self.max_concurrency, self.request_receiver)
     }
 }
 
@@ -84,15 +79,23 @@ impl<Req: Message, Res: Message> RuntimeRequestSink for ServerMessageBoxBuilder<
 impl<Req: Message, Res: Message> ServiceProvider<Req, Res, NoConfig>
     for ServerMessageBoxBuilder<Req, Res>
 {
-    fn connect_consumer(
-        &mut self,
-        _config: NoConfig,
-        response_sender: DynSender<Res>,
-    ) -> DynSender<Req> {
-        let client_id = self.clients.len();
-        let request_sender = KeyedSender::new_sender(client_id, self.request_sender.clone());
-        self.clients.push(response_sender);
-        request_sender
+    fn connect_consumer(&mut self, _config: NoConfig, reply_to: DynSender<Res>) -> DynSender<Req> {
+        Box::new(RequestSender {
+            sender: self.request_sender.sender_clone(),
+            reply_to,
+        })
+    }
+}
+
+impl<Req: Message, Res: Message> MessageSink<RequestEnvelope<Req, Res>, NoConfig>
+    for ServerMessageBoxBuilder<Req, Res>
+{
+    fn get_config(&self) -> NoConfig {
+        NoConfig
+    }
+
+    fn get_sender(&self) -> DynSender<RequestEnvelope<Req, Res>> {
+        self.request_sender().sender_clone()
     }
 }
 
@@ -138,8 +141,7 @@ pub struct ServerActorBuilder<S: Server, K> {
 
 impl<S: Server, K> ServerActorBuilder<S, K> {
     pub fn new(server: S, config: &ServerConfig, kind: K) -> Self {
-        let service_name = server.name().to_string();
-        let box_builder = ServerMessageBoxBuilder::new(&service_name, config.capacity)
+        let box_builder = ServerMessageBoxBuilder::new(server.name(), config.capacity)
             .with_max_concurrency(config.max_concurrency);
 
         ServerActorBuilder {
@@ -148,6 +150,11 @@ impl<S: Server, K> ServerActorBuilder<S, K> {
             box_builder,
         }
     }
+
+    /// Return a sender for the requests
+    pub fn request_sender(&self) -> DynRequestSender<S::Request, S::Response> {
+        self.box_builder.request_sender()
+    }
 }
 
 impl<S: Server> ServerActorBuilder<S, Sequential> {
@@ -204,6 +211,18 @@ impl<S: Server, K> ServiceProvider<S::Request, S::Response, NoConfig> for Server
     }
 }
 
+impl<S: Server, K> MessageSink<RequestEnvelope<S::Request, S::Response>, NoConfig>
+    for ServerActorBuilder<S, K>
+{
+    fn get_config(&self) -> NoConfig {
+        self.box_builder.get_config()
+    }
+
+    fn get_sender(&self) -> DynSender<RequestEnvelope<S::Request, S::Response>> {
+        self.box_builder.get_sender()
+    }
+}
+
 impl<S: Server, K> RuntimeRequestSink for ServerActorBuilder<S, K> {
     fn get_signal_sender(&self) -> DynSender<RuntimeRequest> {
         self.box_builder.get_signal_sender()