Deprecate MessageSource::register_peer

The purpose of the `register_peer` and `connect_sink` methods was
exactly the same: to connect a source to a sink. The only difference was
on their arguments, the sink being respectively passed as a sender or as
a sink. Simply adding an `impl<M> MessageSink<M> for DynSender<M>`,
makes the two methods fully equivalent.

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

Author: didier-wenzek

crates/common/batcher/src/lib.rs

@@ -74,8 +74,8 @@ impl<B: Batchable> MessageSink<BatchDriverInput<B>> for BatchingActorBuilder<B>
 }
 
 impl<B: Batchable> MessageSource<BatchDriverOutput<B>, NoConfig> for BatchingActorBuilder<B> {
-    fn register_peer(&mut self, config: NoConfig, sender: DynSender<BatchDriverOutput<B>>) {
-        self.message_box.register_peer(config, sender)
+    fn connect_sink(&mut self, config: NoConfig, peer: &impl MessageSink<BatchDriverOutput<B>>) {
+        self.message_box.connect_sink(config, peer)
     }
 }
 

crates/core/tedge_actors/src/builders.rs

@@ -95,7 +95,7 @@ pub struct NoConfig;
 /// The [Builder] of an [Actor](crate::Actor) must implement this trait
 /// for every message type that actor can receive from its peers.
 ///
-/// An actor whose builder is a `MessageSink<M` can be connected to any other actor
+/// An actor whose builder is a `MessageSink<M>` can be connected to any other actor
 /// whose builder is a `MessageSource<Into<M>, C>` so that the sink can receive messages from that source.
 pub trait MessageSink<M: Message> {
     /// Return the sender that can be used by peers to send messages to this actor
@@ -105,12 +105,12 @@ pub trait MessageSink<M: Message> {
     ///
     /// A sink might be interested only in a subset of the messages emitted by the source.
     /// This subset is defined by the config parameter.
-    fn connect_source<N, C>(&mut self, config: C, source: &mut impl MessageSource<N, C>)
+    fn connect_source<N, C>(&self, config: C, source: &mut impl MessageSource<N, C>)
     where
         N: Message,
         M: From<N>,
     {
-        source.register_peer(config, self.get_sender().sender_clone())
+        source.connect_sink(config, &self.get_sender())
     }
 
     /// Add a source of messages to the actor under construction, the messages being translated on the fly.
@@ -159,7 +159,7 @@ pub trait MessageSink<M: Message> {
     /// # }
     /// ```
     fn connect_mapped_source<N, C, MS, MessageMapper>(
-        &mut self,
+        &self,
         config: C,
         source: &mut impl MessageSource<N, C>,
         cast: MessageMapper,
@@ -169,23 +169,30 @@ pub trait MessageSink<M: Message> {
         MessageMapper: Fn(N) -> MS,
         MessageMapper: 'static + Send + Sync,
     {
-        let sender = MappingSender::new(self.get_sender(), cast);
-        source.register_peer(config, sender.into())
+        let sender: DynSender<N> = MappingSender::new(self.get_sender(), cast).into();
+        source.connect_sink(config, &sender)
+    }
+}
+
+/// A [DynSender] can be used as a [MessageSink],
+/// provided the messages expected by the sender can be built from those sent to the sink.
+impl<N: Message, M: Message + From<N>> MessageSink<N> for DynSender<M> {
+    fn get_sender(&self) -> DynSender<N> {
+        self.sender_clone()
     }
 }
 
 /// The [Builder] of an [Actor](crate::Actor) must implement this trait
 /// for every message type that actor can send to its peers.
 ///
-/// To receive messages from a `MessageSource<M, C>`, the peer must be a `MessageSink<From<M>`.
+/// To receive messages from a `MessageSource<M, C>`, the peer must be a `MessageSink<From<M>>`.
 pub trait MessageSource<M: Message, Config> {
-    /// The message will be sent to the peer using the provided `sender`
-    fn register_peer(&mut self, config: Config, sender: DynSender<M>);
-
-    /// Connect a peer actor that will consume the message produced by this actor
-    fn connect_sink(&mut self, config: Config, peer: &impl MessageSink<M>) {
-        self.register_peer(config, peer.get_sender());
-    }
+    /// Connect a peer actor that will consume the messages produced by this actor
+    ///
+    /// The messages will be sent to the peer using its sender, the `peer.get_sender()`.
+    /// A peer can subscribe to a subset of the messages produced by this source.
+    /// This subset of messages expected by the peer is defined by the `config` parameter.
+    fn connect_sink(&mut self, config: Config, peer: &impl MessageSink<M>);
 }
 
 /// The [Builder] of an [Actor](crate::Actor) must implement this trait
@@ -290,8 +297,8 @@ pub trait RuntimeRequestSink {
 /// #     }
 /// # }
 /// # impl MessageSource<MyActorOutput, NoConfig> for MyActorBuilder {
-/// #    fn register_peer(&mut self, config: NoConfig, sender: DynSender<MyActorOutput>) {
-/// #        self.messages.register_peer(config, sender)
+/// #    fn connect_sink(&mut self, config: NoConfig, peer: &impl MessageSink<MyActorOutput>) {
+/// #        self.messages.connect_sink(config, peer)
 /// #    }
 /// # }
 /// # impl MessageSink<MyActorInput> for MyActorBuilder {
@@ -330,8 +337,8 @@ pub trait RuntimeRequestSink {
 /// // Connect a test box to an actor under test
 /// let mut my_actor_builder = MyActorBuilder::new(MyActorConfig::default());
 /// let mut test_box_builder = SimpleMessageBoxBuilder::new("Test box", 16);
-/// my_actor_builder.register_peer(NoConfig, test_box_builder.get_sender());
-/// test_box_builder.register_peer(NoConfig, my_actor_builder.get_sender());
+/// my_actor_builder.connect_sink(NoConfig, &test_box_builder);
+/// my_actor_builder.connect_source(NoConfig, &mut test_box_builder);
 ///
 /// // Build the test box and run the actor
 /// let mut test_box = test_box_builder.build();
@@ -379,8 +386,8 @@ impl<I: Message, O: Message> SimpleMessageBoxBuilder<I, O> {
         config: Config,
         service: &mut (impl MessageSink<O> + MessageSource<I, Config>),
     ) {
-        service.register_peer(config, self.input_sender.sender_clone());
-        self.register_peer(NoConfig, service.get_sender());
+        service.connect_sink(config, self);
+        self.connect_sink(NoConfig, service);
     }
 
     /// Connect this client message box to the service message box
@@ -399,8 +406,8 @@ impl<I: Message, O: Message> SimpleMessageBoxBuilder<I, O> {
 
 /// A `SimpleMessageBoxBuilder<Input,Output>` is a [MessageSource] of `Output` messages ignoring the config.
 impl<I: Message, O: Message, C> MessageSource<O, C> for SimpleMessageBoxBuilder<I, O> {
-    fn register_peer(&mut self, _config: C, sender: DynSender<O>) {
-        self.output_sender = sender;
+    fn connect_sink(&mut self, _config: C, peer: &impl MessageSink<O>) {
+        self.output_sender = peer.get_sender();
     }
 }
 

crates/core/tedge_actors/src/converter.rs

@@ -259,8 +259,8 @@ impl<C: Converter> Builder<ConvertingActor<C>> for ConvertingActorBuilder<C> {
 }
 
 impl<C: Converter> MessageSource<C::Output, NoConfig> for ConvertingActorBuilder<C> {
-    fn register_peer(&mut self, config: NoConfig, sender: DynSender<C::Output>) {
-        self.message_box.register_peer(config, sender)
+    fn connect_sink(&mut self, config: NoConfig, peer: &impl MessageSink<C::Output>) {
+        self.message_box.connect_sink(config, peer)
     }
 }
 

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

@@ -61,13 +61,13 @@
 //! // - receives as input, the output messages sent from the server message box
 //! // - sends output messages to the server message box as its input.
 //! let request_sender = server_box_builder.add_requester(player_1_box_builder.get_sender());
-//! player_1_box_builder.register_peer(NoConfig, request_sender);
+//! player_1_box_builder.connect_sink(NoConfig, &request_sender);
 //!
 //! // It matters that the builder of the server box is a `ServerMessageBoxBuilder`:
 //! // as this builder accepts multiple client actors to connect to the same server.
 //! let mut player_2_box_builder = SimpleMessageBoxBuilder::new("Player 2", 1);
 //! let request_sender = server_box_builder.add_requester(player_2_box_builder.get_sender());
-//! player_2_box_builder.register_peer(NoConfig, request_sender);
+//! player_2_box_builder.connect_sink(NoConfig, &request_sender);
 //!
 //! // One can then build the message boxes
 //! let server_box: ServerMessageBox<Operation,Update> = server_box_builder.build();

crates/core/tedge_actors/src/lib.rs

@@ -213,7 +213,7 @@
 //!     ///
 //!     /// The source registers the new consumer and its sender (i.e. where to send response),
 //!     /// possibly using the configuration `config` to filter messages.
-//!     fn register_peer(&mut self, config: SomeConfig, sender: DynSender<SomeMessage>) {
+//!     fn connect_sink(&mut self, config: SomeConfig, peer: &impl MessageSink<SomeMessage>) {
 //!         todo!()
 //!     }
 //! }
@@ -231,7 +231,7 @@
 //! // can then be connected to each other.
 //! let mut producer = SomeActorBuilder::default();
 //! let mut consumer = SomeOtherActorBuilder::default();
-//! producer.register_peer(SomeConfig, consumer.get_sender());
+//! producer.connect_sink(SomeConfig, &consumer);
 //! ```
 //!
 //! ## Running actors

crates/core/tedge_actors/src/test_helpers.rs

@@ -520,8 +520,9 @@ impl<I: MessagePlus, O: MessagePlus> Probe<I, O> {
         source: &mut impl MessageSource<I, C>,
         sink: &mut impl MessageSink<O>,
     ) {
+        let input_interceptor: DynSender<I> = self.input_interceptor.clone().into();
         self.output_forwarder = sink.get_sender();
-        source.register_peer(config, self.input_interceptor.clone().into());
+        source.connect_sink(config, &input_interceptor);
     }
 
     /// Connect this probe to a service provider
@@ -638,15 +639,16 @@ where
         &'a mut self,
         probe: &'a mut Probe<Response, Request>,
     ) -> &'a mut Probe<Response, Request> {
+        let output_interceptor: DynSender<Request> = probe.output_interceptor.clone().into();
         probe.input_forwarder = self.get_sender();
-        self.register_peer(NoConfig, probe.output_interceptor.sender_clone());
+        self.connect_sink(NoConfig, &output_interceptor);
         probe
     }
 }
 
 impl<I: MessagePlus, O: MessagePlus> MessageSource<O, NoConfig> for Probe<I, O> {
-    fn register_peer(&mut self, _config: NoConfig, sender: DynSender<O>) {
-        self.output_forwarder = sender;
+    fn connect_sink(&mut self, _config: NoConfig, peer: &impl MessageSink<O>) {
+        self.output_forwarder = peer.get_sender();
     }
 }
 
@@ -666,11 +668,11 @@ impl<I: Message, O: Message> ServiceProviderExt<I, O> for DynSender<RequestEnvel
         let name = "client-box";
         let capacity = 16;
         let mut client_box = SimpleMessageBoxBuilder::new(name, capacity);
-        let request_sender = RequestSender {
+        let request_sender = Box::new(RequestSender {
             sender: self.sender_clone(),
             reply_to: client_box.get_sender(),
-        };
-        client_box.register_peer(NoConfig, request_sender.into());
+        });
+        client_box.connect_sink(NoConfig, &request_sender.sender_clone());
         client_box.build()
     }
 }
@@ -686,8 +688,8 @@ impl<I: Message, O: Message> ServiceProviderExt<I, O> for SimpleMessageBoxBuilde
         let name = "client-box";
         let capacity = 16;
         let mut client_box = SimpleMessageBoxBuilder::new(name, capacity);
-        self.register_peer(NoConfig, client_box.get_sender());
-        client_box.register_peer(NoConfig, self.get_sender());
+        self.connect_sink(NoConfig, &client_box);
+        self.connect_source(NoConfig, &mut client_box);
         client_box.build()
     }
 }

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

@@ -298,10 +298,7 @@ impl FileCacheActorBuilder {
             downloader_actor.add_requester(message_box.get_sender().sender_clone());
 
         let mqtt_sender = mqtt_actor.get_sender();
-        mqtt_actor.register_peer(
-            Self::subscriptions(&mqtt_schema),
-            message_box.get_sender().sender_clone(),
-        );
+        mqtt_actor.connect_sink(Self::subscriptions(&mqtt_schema), &message_box.get_sender());
 
         Self {
             message_box,

crates/core/tedge_agent/src/restart_manager/builder.rs

@@ -34,8 +34,8 @@ impl MessageSink<RestartCommand> for RestartManagerBuilder {
 }
 
 impl MessageSource<RestartCommand, NoConfig> for RestartManagerBuilder {
-    fn register_peer(&mut self, config: NoConfig, sender: DynSender<RestartCommand>) {
-        self.message_box.register_peer(config, sender)
+    fn connect_sink(&mut self, config: NoConfig, peer: &impl MessageSink<RestartCommand>) {
+        self.message_box.connect_sink(config, peer)
     }
 }
 

crates/core/tedge_agent/src/restart_manager/tests.rs

@@ -8,7 +8,6 @@ use tedge_actors::Actor;
 use tedge_actors::Builder;
 use tedge_actors::DynError;
 use tedge_actors::MessageReceiver;
-use tedge_actors::MessageSink;
 use tedge_actors::MessageSource;
 use tedge_actors::NoConfig;
 use tedge_actors::Sender;
@@ -120,8 +119,8 @@ async fn spawn_restart_manager(
     };
 
     let mut restart_actor_builder = RestartManagerBuilder::new(config);
-    converter_builder.register_peer(NoConfig, restart_actor_builder.get_sender());
-    restart_actor_builder.register_peer(NoConfig, converter_builder.get_sender());
+    converter_builder.connect_sink(NoConfig, &restart_actor_builder);
+    restart_actor_builder.connect_sink(NoConfig, &converter_builder);
 
     let converter_box = converter_builder.build().with_timeout(TEST_TIMEOUT_MS);
 

crates/core/tedge_agent/src/software_manager/builder.rs

@@ -34,8 +34,8 @@ impl MessageSink<SoftwareCommand> for SoftwareManagerBuilder {
 }
 
 impl MessageSource<SoftwareCommand, NoConfig> for SoftwareManagerBuilder {
-    fn register_peer(&mut self, config: NoConfig, sender: DynSender<SoftwareCommand>) {
-        self.message_box.register_peer(config, sender)
+    fn connect_sink(&mut self, config: NoConfig, peer: &impl MessageSink<SoftwareCommand>) {
+        self.message_box.connect_sink(config, peer)
     }
 }