Remove impl ServiceProvider for ServerActorBuilder

Connecting a server actor to its clients has been the main motivation
for the ServiceProvider trait which where used to let the client and the
server exchange senders, one given by the server to the client requests and
another given by the client to server responses.

With the new mechanism of RequestEnvelope, this exchange is no more
useful. The actor server is a simple MessageSink accepting requests
wrapped in an envelope with a reply-to address, i.e. a sender.
Clients only needs a sender to post their request.

The next step will be to fully deprecate the ServiceProvider trait,
in favor of the combination of two traits MessageSource + MessageSink.
Indeed, the complexity of the ServiceProvider was only used by server
actors and not the actors, such as the MQTTActor, for which the messages
received and sent don't have a request-response relationship.

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

Author: didier-wenzek

crates/common/batcher/src/driver.rs

@@ -178,7 +178,6 @@ mod tests {
     use std::time::Duration;
     use tedge_actors::test_helpers::ServiceProviderExt;
     use tedge_actors::Builder;
-    use tedge_actors::NoConfig;
     use tedge_actors::SimpleMessageBoxBuilder;
     use tokio::time::timeout;
 
@@ -256,7 +255,7 @@ mod tests {
             .build();
         let batcher = Batcher::new(config);
         let mut box_builder = SimpleMessageBoxBuilder::new("test", 1);
-        let test_box = box_builder.new_client_box(NoConfig);
+        let test_box = box_builder.new_client_box();
         let driver_box = box_builder.build();
 
         let driver = BatchDriver::new(batcher, driver_box);

crates/core/tedge_actors/src/actors.rs

@@ -77,7 +77,7 @@ pub mod tests {
     #[tokio::test]
     async fn running_an_actor_without_a_runtime() {
         let mut box_builder = SimpleMessageBoxBuilder::new("test", 16);
-        let mut client_message_box = box_builder.new_client_box(NoConfig);
+        let mut client_message_box = box_builder.new_client_box();
         let mut runtime_box = box_builder.get_signal_sender();
         let actor_message_box = box_builder.build();
         let actor = Echo {

crates/core/tedge_actors/src/lib.rs

@@ -133,7 +133,7 @@
 //! let mut actor_box_builder = SimpleMessageBoxBuilder::new("Actor", 10);
 //!
 //! // Create a test box ready then the actor box
-//! let mut test_box = actor_box_builder.new_client_box(NoConfig);
+//! let mut test_box = actor_box_builder.new_client_box();
 //! let actor_box = actor_box_builder.build();
 //!
 //! // The actor is then spawn in the background with its message box.

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

@@ -201,16 +201,6 @@ impl<S: Server + Clone> Builder<ConcurrentServerActor<S>> for ServerActorBuilder
     }
 }
 
-impl<S: Server, K> ServiceProvider<S::Request, S::Response, NoConfig> for ServerActorBuilder<S, K> {
-    fn connect_consumer(
-        &mut self,
-        config: NoConfig,
-        response_sender: DynSender<S::Response>,
-    ) -> DynSender<S::Request> {
-        self.box_builder.connect_consumer(config, response_sender)
-    }
-}
-
 impl<S: Server, K> MessageSink<RequestEnvelope<S::Request, S::Response>, NoConfig>
     for ServerActorBuilder<S, K>
 {

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

@@ -166,7 +166,6 @@ mod tests {
     use crate::Builder;
     use crate::ConcurrentServerActor;
     use crate::DynSender;
-    use crate::NoConfig;
     use crate::Runtime;
     use crate::RuntimeRequest;
     use crate::RuntimeRequestSink;
@@ -181,7 +180,7 @@ mod tests {
     #[tokio::test]
     async fn only_processes_messages_up_to_max_concurrency() {
         let mut builder = ServerMessageBoxBuilder::new("ConcurrentServerMessageBoxTest", 16);
-        let mut test_box = builder.new_client_box(NoConfig);
+        let mut test_box = builder.new_client_box();
         let message_box: ServerMessageBox<i32, i32> = builder.build();
         let mut concurrent_box = ConcurrentServerMessageBox::new(4, message_box);
 
@@ -285,7 +284,7 @@ mod tests {
             }
 
             fn client_box(&mut self) -> SimpleMessageBox<i32, i32> {
-                self.box_builder.new_client_box(NoConfig)
+                self.box_builder.new_client_box()
             }
         }
 

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

@@ -68,8 +68,8 @@
 //! let mut handle = actor.request_sender();
 //!
 //! // This handle can then be used to connect client message boxes
-//! let mut client_1 = handle.new_client_box(NoConfig);
-//! let mut client_2 = handle.new_client_box(NoConfig);
+//! let mut client_1 = handle.new_client_box();
+//! let mut client_2 = handle.new_client_box();
 //!
 //! // The actor is then spawn in the background.
 //! tokio::spawn(async move { actor.run().await } );
@@ -103,7 +103,6 @@ use crate::Message;
 use crate::MessageSink;
 use crate::NoConfig;
 use crate::Sender;
-use crate::ServiceProvider;
 use async_trait::async_trait;
 
 /// Define how a server process a request
@@ -141,28 +140,23 @@ impl<Request: Debug, Response> Debug for RequestEnvelope<Request, Response> {
 /// A request sender to some [Server]
 pub type DynRequestSender<Request, Response> = DynSender<RequestEnvelope<Request, Response>>;
 
-impl<Request: Message, Response: Message> ServiceProvider<Request, Response, NoConfig>
-    for DynRequestSender<Request, Response>
-{
-    fn connect_consumer(
-        &mut self,
-        _config: NoConfig,
-        reply_to: DynSender<Response>,
-    ) -> DynSender<Request> {
-        Box::new(RequestSender {
-            sender: self.sender_clone(),
-            reply_to,
-        })
-    }
-}
-
 /// A connector to a [Server] expecting Request and returning Response.
 pub trait Service<Request: Message, Response: Message>:
     MessageSink<RequestEnvelope<Request, Response>, NoConfig>
 {
+    /// Connect a request message box to the server box under construction
+    fn add_requester(&mut self, response_sender: DynSender<Response>) -> DynSender<Request>;
 }
 
-impl<T, Request: Message, Response: Message> Service<Request, Response> for T where
-    T: MessageSink<RequestEnvelope<Request, Response>, NoConfig>
+impl<T, Request: Message, Response: Message> Service<Request, Response> for T
+where
+    T: MessageSink<RequestEnvelope<Request, Response>, NoConfig>,
 {
+    fn add_requester(&mut self, reply_to: DynSender<Response>) -> DynSender<Request> {
+        let request_sender = RequestSender {
+            sender: self.get_sender(),
+            reply_to,
+        };
+        request_sender.into()
+    }
 }

crates/core/tedge_actors/src/test_helpers.rs

@@ -2,6 +2,8 @@
 use crate::mpsc;
 use crate::Builder;
 use crate::ChannelError;
+use crate::CloneSender;
+use crate::DynRequestSender;
 use crate::DynSender;
 use crate::Message;
 use crate::MessageReceiver;
@@ -11,10 +13,11 @@ use crate::NoConfig;
 use crate::NoMessage;
 use crate::NullSender;
 use crate::RequestEnvelope;
+use crate::RequestSender;
 use crate::RuntimeRequest;
 use crate::Sender;
+use crate::ServerMessageBoxBuilder;
 use crate::ServiceConsumer;
-use crate::ServiceProvider;
 use crate::SimpleMessageBox;
 use crate::SimpleMessageBoxBuilder;
 use crate::WrappedInput;
@@ -651,56 +654,39 @@ impl<I: MessagePlus, O: MessagePlus> MessageSink<I, NoConfig> for Probe<I, O> {
     }
 }
 
-pub trait ServiceProviderExt<I: Message, O: Message, C> {
+pub trait ServiceProviderExt<I: Message, O: Message> {
     /// Create a simple message box connected to a box under construction.
-    fn new_client_box(&mut self, config: C) -> SimpleMessageBox<O, I>;
+    fn new_client_box(&mut self) -> SimpleMessageBox<O, I>;
 }
 
-impl<I, O, C, T> ServiceProviderExt<I, O, C> for T
-where
-    I: Message,
-    O: Message,
-    C: Clone,
-    T: ServiceProvider<I, O, C>,
-{
-    fn new_client_box(&mut self, config: C) -> SimpleMessageBox<O, I> {
+impl<I: Message, O: Message> ServiceProviderExt<I, O> for DynRequestSender<I, O> {
+    fn new_client_box(&mut self) -> SimpleMessageBox<O, I> {
         let name = "client-box";
         let capacity = 16;
-        let mut client_box = ConsumerBoxBuilder::new(name, capacity, config);
-        self.add_peer(&mut client_box);
+        let mut client_box = SimpleMessageBoxBuilder::new(name, capacity);
+        let request_sender = RequestSender {
+            sender: self.sender_clone(),
+            reply_to: client_box.get_sender(),
+        };
+        client_box.set_request_sender(request_sender.into());
         client_box.build()
     }
 }
 
-struct ConsumerBoxBuilder<I, O: Debug, C> {
-    config: C,
-    box_builder: SimpleMessageBoxBuilder<O, I>,
-}
-
-impl<I: Message, O: Message, C> ConsumerBoxBuilder<I, O, C> {
-    fn new(name: &str, capacity: usize, config: C) -> Self {
-        ConsumerBoxBuilder {
-            config,
-            box_builder: SimpleMessageBoxBuilder::new(name, capacity),
-        }
-    }
-
-    fn build(self) -> SimpleMessageBox<O, I> {
-        self.box_builder.build()
+impl<I: Message, O: Message> ServiceProviderExt<I, O> for ServerMessageBoxBuilder<I, O> {
+    fn new_client_box(&mut self) -> SimpleMessageBox<O, I> {
+        self.request_sender().new_client_box()
     }
 }
 
-impl<I: Message, O: Message, C: Clone> ServiceConsumer<I, O, C> for ConsumerBoxBuilder<I, O, C> {
-    fn get_config(&self) -> C {
-        self.config.clone()
-    }
-
-    fn set_request_sender(&mut self, request_sender: DynSender<I>) {
-        self.box_builder.set_request_sender(request_sender)
-    }
-
-    fn get_response_sender(&self) -> DynSender<O> {
-        self.box_builder.get_response_sender()
+impl<I: Message, O: Message> ServiceProviderExt<I, O> for SimpleMessageBoxBuilder<I, O> {
+    fn new_client_box(&mut self) -> SimpleMessageBox<O, I> {
+        let name = "client-box";
+        let capacity = 16;
+        let mut client_box = SimpleMessageBoxBuilder::new(name, capacity);
+        self.set_request_sender(client_box.get_sender());
+        client_box.set_request_sender(self.get_sender());
+        client_box.build()
     }
 }
 
@@ -735,9 +721,7 @@ pub struct FakeServerBox<Request: Debug, Response> {
 impl<Request: Message, Response: Message> FakeServerBox<Request, Response> {
     /// Return a fake message box builder
     pub fn builder() -> FakeServerBoxBuilder<Request, Response> {
-        FakeServerBoxBuilder {
-            messages: SimpleMessageBoxBuilder::new("Fake Server", 16),
-        }
+        FakeServerBoxBuilder::default()
     }
 }
 
@@ -799,6 +783,14 @@ pub struct FakeServerBoxBuilder<Request: Debug, Response> {
     messages: SimpleMessageBoxBuilder<RequestEnvelope<Request, Response>, NoMessage>,
 }
 
+impl<Request: Message, Response: Message> Default for FakeServerBoxBuilder<Request, Response> {
+    fn default() -> Self {
+        FakeServerBoxBuilder {
+            messages: SimpleMessageBoxBuilder::new("Fake Server", 16),
+        }
+    }
+}
+
 impl<Request: Message, Response: Message> MessageSink<RequestEnvelope<Request, Response>, NoConfig>
     for FakeServerBoxBuilder<Request, Response>
 {

crates/core/tedge_actors/src/tests.rs

@@ -50,7 +50,7 @@ async fn spawn_concurrent_sleep_service(max_concurrency: usize) -> DynRequestSen
 #[tokio::test]
 async fn requests_are_served_in_turn() {
     let mut service_handle = spawn_sleep_service().await;
-    let mut client = service_handle.new_client_box(NoConfig);
+    let mut client = service_handle.new_client_box();
 
     // The requests being sent in some order
     client.send(1).await.unwrap();
@@ -67,8 +67,8 @@ async fn requests_are_served_in_turn() {
 async fn clients_can_interleave_request() {
     let mut service_handle = spawn_sleep_service().await;
 
-    let mut client_1 = service_handle.new_client_box(NoConfig);
-    let mut client_2 = service_handle.new_client_box(NoConfig);
+    let mut client_1 = service_handle.new_client_box();
+    let mut client_2 = service_handle.new_client_box();
 
     // Two clients can independently send requests
     client_1.send(1).await.unwrap();
@@ -85,8 +85,8 @@ async fn clients_can_interleave_request() {
 async fn requests_can_be_sent_concurrently() {
     let mut service_handle = spawn_concurrent_sleep_service(2).await;
 
-    let mut client_1 = service_handle.new_client_box(NoConfig);
-    let mut client_2 = service_handle.new_client_box(NoConfig);
+    let mut client_1 = service_handle.new_client_box();
+    let mut client_2 = service_handle.new_client_box();
 
     // Despite a long running request from client_1
     client_1.send(1000).await.unwrap();

crates/core/tedge_agent/src/operation_file_cache/mod.rs

@@ -30,6 +30,7 @@ use tedge_actors::RuntimeError;
 use tedge_actors::RuntimeRequest;
 use tedge_actors::RuntimeRequestSink;
 use tedge_actors::Sender;
+use tedge_actors::Service;
 use tedge_actors::ServiceProvider;
 use tedge_actors::SimpleMessageBoxBuilder;
 use tedge_api::messages::ConfigUpdateCmdPayload;
@@ -289,13 +290,13 @@ impl FileCacheActorBuilder {
         mqtt_schema: MqttSchema,
         tedge_http_host: Arc<str>,
         data_dir: DataDir,
-        downloader_actor: &mut impl ServiceProvider<IdDownloadRequest, IdDownloadResult, NoConfig>,
+        downloader_actor: &mut impl Service<IdDownloadRequest, IdDownloadResult>,
         mqtt_actor: &mut impl ServiceProvider<MqttMessage, MqttMessage, TopicFilter>,
     ) -> Self {
         let message_box = SimpleMessageBoxBuilder::new("RestartManager", 10);
 
         let download_sender =
-            downloader_actor.connect_consumer(NoConfig, message_box.get_sender().sender_clone());
+            downloader_actor.add_requester(message_box.get_sender().sender_clone());
 
         let mqtt_sender = mqtt_actor.connect_consumer(
             Self::subscriptions(&mqtt_schema),

crates/extensions/c8y_mapper_ext/src/actor.rs

@@ -298,8 +298,8 @@ impl C8yMapperBuilder {
         mqtt: &mut impl ServiceProvider<MqttMessage, MqttMessage, TopicFilter>,
         http: &mut impl Service<C8YRestRequest, C8YRestResult>,
         timer: &mut impl ServiceProvider<SyncStart, SyncComplete, NoConfig>,
-        uploader: &mut impl ServiceProvider<IdUploadRequest, IdUploadResult, NoConfig>,
-        downloader: &mut impl ServiceProvider<IdDownloadRequest, IdDownloadResult, NoConfig>,
+        uploader: &mut impl Service<IdUploadRequest, IdUploadResult>,
+        downloader: &mut impl Service<IdDownloadRequest, IdDownloadResult>,
         fs_watcher: &mut impl MessageSource<FsWatchEvent, PathBuf>,
         service_monitor: &mut impl ServiceProvider<MqttMessage, MqttMessage, TopicFilter>,
     ) -> Result<Self, FileError> {
@@ -314,10 +314,8 @@ impl C8yMapperBuilder {
         let http_proxy = C8YHttpProxy::new(http);
         let timer_sender =
             timer.connect_consumer(NoConfig, box_builder.get_sender().sender_clone());
-        let upload_sender =
-            uploader.connect_consumer(NoConfig, box_builder.get_sender().sender_clone());
-        let download_sender =
-            downloader.connect_consumer(NoConfig, box_builder.get_sender().sender_clone());
+        let upload_sender = uploader.add_requester(box_builder.get_sender().sender_clone());
+        let download_sender = downloader.add_requester(box_builder.get_sender().sender_clone());
         fs_watcher.register_peer(
             config.ops_dir.clone(),
             box_builder.get_sender().sender_clone(),