Merge pull request #2780 from didier-wenzek/refactor/improve-message-source-sink-naming

refactor: improve message source and sink traits

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

@@ -47,17 +47,17 @@
 //!   (possibly several receivers to handle message priorities among inputs),
 //!   and has to give clones of the associated Sender (or Senders) to its peers.
 //! - The first responsibility of a builder is to create a channel per receiver of the actor
-//!   under construction. The receiver will be given to the actor on build.
+//!   under construction. The receiver will be given to the actor when built.
 //!   The sender is owned by the builder to be cloned and given to any peer that needs to send data
 //!   to the actor under construction.
-//! - The second responsibility of the builder is to collect a Sender for each peer the actor
+//! - The second responsibility of the builder is to collect a sender for each peer the actor
 //!   under construction needs to send messages to. This is the mirror of the previous responsibility:
 //!   each builder gives to the others clones of its senders and collects senders from others.
 //! - This is why all the actor building traits
 //!   ([MessageSource], [MessageSink] and [RuntimeRequestSink])
-//!   are related to exchanges of Sender. A sink gives to a source a sender attached to its receiver.
-//! - To be precise, the actor builders exchange [DynSender] and not [Sender]. The difference is that
-//!   a [DynSender] can transform the messages sent by the source to adapt them to the sink expectations,
+//!   are related to exchanges of Sender. A sink gives to a source a sender attached to its own receiver.
+//! - To be precise, the actor builders exchange [DynSender] and not [Sender](futures::channel::mpsc::Sender).
+//!   The difference is that a [DynSender] can transform the messages sent by the source to adapt them to the sink expectations,
 //!   using an `impl From<SourceMessage> for SinkMessage`. This flexibility allows an actor to receive
 //!   messages from several independent sources (see the [fan_in_message_type](crate::fan_in_message_type) macro).
 use crate::mpsc;
@@ -95,8 +95,8 @@ 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, C>` can be connected to any other actor
-/// whose builder is a `MessageSource<M, C>` so that the sink can receive messages from that source.
+/// 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
     fn get_sender(&self) -> DynSender<M>;
@@ -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 add_input<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.
@@ -139,7 +139,7 @@ pub trait MessageSink<M: Message> {
     /// let mut sender_builder = SimpleMessageBoxBuilder::new("Send", 16);
     ///
     /// // Convert the `&str` sent by the source into an iterator of `char` as expected by the receiver.
-    /// receiver_builder.add_mapped_input(NoConfig, &mut sender_builder, |str: &'static str| str.chars() );
+    /// receiver_builder.connect_mapped_source(NoConfig, &mut sender_builder, |str: &'static str| str.chars() );
     ///
     /// let mut sender: SimpleMessageBox<NoMessage, &'static str>= sender_builder.build();
     /// let receiver: SimpleMessageBox<char, NoMessage> = receiver_builder.build();
@@ -158,8 +158,8 @@ pub trait MessageSink<M: Message> {
     /// # Ok(())
     /// # }
     /// ```
-    fn add_mapped_input<N, C, MS, MessageMapper>(
-        &mut self,
+    fn connect_mapped_source<N, C, MS, MessageMapper>(
+        &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<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 add_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/channels.rs

@@ -25,7 +25,7 @@ pub trait CloneSender<M>: Sender<M> {
 
     /// Clone a cast of this sender into a `Box<dyn Sender<M>>`
     ///
-    /// This is a workaround for https://github.com/rust-lang/rust/issues/65991
+    /// This is a workaround for <https://github.com/rust-lang/rust/issues/65991>
     fn sender(&self) -> Box<dyn Sender<M>>;
 }
 

crates/core/tedge_actors/src/converter.rs

@@ -213,10 +213,10 @@ impl<C: Converter> ConvertingActor<C> {
 ///     );
 ///
 ///     // Connect this actor as a sink of the mqtt actor, to receive input messages from it
-///     converter_builder.add_input(TopicFilter("some/mqtt/topics/#"), &mut mqtt_builder);
+///     converter_builder.connect_source(TopicFilter("some/mqtt/topics/#"), &mut mqtt_builder);
 ///
 ///     // Connect the same mqtt actor as a sink of this actor, to send output messages to it
-///     converter_builder.add_sink(NoConfig, &mut mqtt_builder);
+///     converter_builder.connect_sink(NoConfig, &mut mqtt_builder);
 ///
 ///     // Finally build the actor
 ///     converter_builder.build()
@@ -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

@@ -43,7 +43,7 @@
 //! to establish appropriate connections between the actor message boxes.
 //!
 //! ```
-//! # use tedge_actors::{Actor, Builder, ChannelError, MessageReceiver, Sender, NoConfig, ServerActor, ServerMessageBox, ServerMessageBoxBuilder, SimpleMessageBox, SimpleMessageBoxBuilder, Service, MessageSink};
+//! # use tedge_actors::{Actor, Builder, ChannelError, MessageReceiver, Sender, NoConfig, ServerActor, ServerMessageBox, ServerMessageBoxBuilder, SimpleMessageBox, SimpleMessageBoxBuilder, Service, MessageSink, MessageSource};
 //! # use crate::tedge_actors::examples::calculator_server::*;
 //! # #[tokio::main]
 //! # async fn main_test() -> Result<(),ChannelError> {
@@ -60,12 +60,14 @@
 //! // Connecting the two boxes, so the box built by the `player_box_builder`:
 //! // - receives as input, the output messages sent from the server message box
 //! // - sends output messages to the server message box as its input.
-//! server_box_builder.add_client(&mut player_1_box_builder);
+//! let request_sender = server_box_builder.connect_client(player_1_box_builder.get_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);
-//! server_box_builder.add_client(&mut player_2_box_builder);
+//! let request_sender = server_box_builder.connect_client(player_2_box_builder.get_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

@@ -4,7 +4,7 @@
 //!
 //! Actors are processing units that interact using asynchronous messages.
 //!
-//! The behavior of an [Actor](crate::Actor) is defined by:
+//! The behavior of an [Actor] is defined by:
 //! - a state owned and freely updated by the actor,
 //! - a [message box](crate::message_boxes) connected to peer actors,
 //! - input [messages](crate::Message) that the actor receives from its peers and processes in turn,
@@ -87,9 +87,9 @@
 //! ```
 //!
 //! This crate provides specific `Actor` implementations:
-//! - The [ServerActor](crate::ServerActor) wraps a [Server](crate::Server),
+//! - The [ServerActor] wraps a [Server],
 //!   to implement a request-response communication pattern with a set of connected client actors.
-//! - The [ConvertingActor](crate::ConvertingActor) wraps a [Converter](crate::Converter),
+//! - The [ConvertingActor] wraps a [Converter],
 //!   that translates each input message into a sequence of output messages.
 //!
 //! ## Testing an actor
@@ -104,9 +104,8 @@
 //! [Actor and message box builders](crate::builders) are provided to address these specificities
 //! with a generic approach without exposing the internal structure of the actors.
 //!
-//! To test the `Calculator` example we need first to create its box using a
-//! [SimpleMessageBoxBuilder](crate::SimpleMessageBoxBuilder),
-//! as this actor expects a [SimpleMessageBox](crate::SimpleMessageBox).
+//! To test the `Calculator` example we need first to create its box using a [SimpleMessageBoxBuilder],
+//! as this actor expects a [SimpleMessageBox].
 //! And then, to create a test box connected to the actor message box,
 //! we use the [ServiceProviderExt](crate::test_helpers::ServiceProviderExt) test helper extension
 //! and the [new_client_box](crate::test_helpers::ServiceProviderExt::new_client_box) method.
@@ -153,7 +152,7 @@
 //! # }
 //! ```
 //!
-//! See the [test_helpers](crate::test_helpers) module for various ways
+//! See the [test_helpers] module for various ways
 //! to observe and interact with running actors.
 //!
 //! - The primary tool to interact with an actor under test is the [SimpleMessageBoxBuilder],
@@ -172,26 +171,20 @@
 //! that define the services provided and consumed by the actors under construction.
 //!
 //! The connection builder traits work by pairs.
-//! A [MessageSink](crate::MessageSink) connects to a [MessageSource](crate::MessageSource),
+//! A [MessageSink] connects to a [MessageSource],
 //! so the messages sent by the latter will be received by the former.
 //!
 //! These traits define the types of the messages sent and received.
 //! - A sink that excepts message of type `M` can only be connected to a source of messages
 //!   that can be converted into `M` values.
-//! - Similarly a service is defined by two types of messages, the requests received by the service
-//!   and the responses sent by the service. To use a service, a consumer will have to send messages
-//!   that can be converted into the service request type and be ready to receive messages converted from
-//!   the service response type.
 //! - Note, that no contract is enforced beyond the type-compatibility of the messages sent between the actors.
 //!   A consumer of an HTTP service needs to known that a request must be sent before any response can be received;
 //!   while a consumer of an MQTT service can expect to receive messages without sending a single one.
 //!
 //! The connection builder traits also define a configuration type.
-//! - The semantics of this type is defined by the message source or the service provider.
-//!   It can be used to filter the values sent to a given sink
-//!   or to restrict the scope of the service provided to a given service consumer.
-//! - The configuration values are provided by the message sinks and the service consumers
-//!   to specify the context of their connection to a source or a service.
+//! - The semantics of this type is defined by the message source and is used to filter the values sent to a sink.
+//! - The actual configuration values are provided by the actor sinks when connecting the source
+//!   to specify the subset of messages their are interested into.
 //!
 //! Note that these traits are implemented by the actor builders, not by the actors themselves.
 //!
@@ -213,7 +206,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 +224,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/message_boxes.rs

@@ -7,11 +7,11 @@
 //! Conceptually, a message box is a receiver of input messages combined with a sender of output messages.
 //! * The receiver is connected to the senders of peer actors;
 //!   and reciprocally the sender is connected to receivers of peer actors.
-//! * The receivers are [mpsc::Receiver](crate::mpsc::Receiver) that collect messages from several sources,
+//! * The receivers are [mpsc::Receiver] that collect messages from several sources,
 //!   and deliver the messages to the actor in the order they have been received.
-//! * The senders are [DynSender](crate::DynSender) that adapt the messages sent to match constraints of the receivers.
+//! * The senders are [DynSender] that adapt the messages sent to match constraints of the receivers.
 //!
-//! A [SimpleMessageBox](crate::SimpleMessageBox) implements exactly this conceptual view:
+//! A [SimpleMessageBox] implements exactly this conceptual view:
 //!
 //! ```ascii
 //!                    input_senders: DynSender<Input> ...
@@ -79,7 +79,7 @@
 //!
 //! This crates provides several built-in message box implementations:
 //!
-//! - [SimpleMessageBox](crate::SimpleMessageBox) for actors that simply process messages in turn,
+//! - [SimpleMessageBox] for actors that simply process messages in turn,
 //! - [ServerMessageBox](crate::ServerMessageBox) for server actors that deliver a request-response service,
 //! - [ConcurrentServerMessageBox](crate::ConcurrentServerMessageBox) for server actors that process requests concurrently,
 //! - [ClientMessageBox](crate::ClientMessageBox) for client actors that use a request-response service from a server actor,

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

@@ -4,7 +4,6 @@ use crate::Builder;
 use crate::CloneSender;
 use crate::ConcurrentServerActor;
 use crate::ConcurrentServerMessageBox;
-use crate::DynRequestSender;
 use crate::DynSender;
 use crate::LoggingReceiver;
 use crate::Message;
@@ -52,7 +51,7 @@ impl<Request: Message, Response: Message> ServerMessageBoxBuilder<Request, Respo
     }
 
     /// Return a sender for the requests
-    pub fn request_sender(&self) -> DynRequestSender<Request, Response> {
+    pub fn request_sender(&self) -> DynSender<RequestEnvelope<Request, Response>> {
         self.request_sender.sender_clone()
     }
 
@@ -134,7 +133,7 @@ impl<S: Server, K> ServerActorBuilder<S, K> {
     }
 
     /// Return a sender for the requests
-    pub fn request_sender(&self) -> DynRequestSender<S::Request, S::Response> {
+    pub fn request_sender(&self) -> DynSender<RequestEnvelope<S::Request, S::Response>> {
         self.box_builder.request_sender()
     }
 }