Rename MessageSink::add_input to connect_source

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

Author: didier-wenzek

crates/core/tedge_actors/src/builders.rs

@@ -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,7 +105,7 @@ 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>(&mut self, config: C, source: &mut impl MessageSource<N, C>)
     where
         N: Message,
         M: From<N>,

crates/core/tedge_actors/src/converter.rs

@@ -213,7 +213,7 @@ 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);

crates/core/tedge_agent/src/agent.rs

@@ -436,7 +436,7 @@ pub fn create_tedge_to_te_converter(
     let mut tedge_converter_actor =
         ConvertingActor::builder("TedgetoTeConverter", tedge_to_te_converter);
 
-    tedge_converter_actor.add_input(subscriptions, mqtt_actor_builder);
+    tedge_converter_actor.connect_source(subscriptions, mqtt_actor_builder);
     tedge_converter_actor.add_sink(NoConfig, mqtt_actor_builder);
 
     Ok(tedge_converter_actor)

crates/core/tedge_mapper/src/aws/mapper.rs

@@ -40,7 +40,7 @@ impl TEdgeComponent for AwsMapper {
         );
         let mut aws_converting_actor = ConvertingActor::builder("AwsConverter", aws_converter);
 
-        aws_converting_actor.add_input(get_topic_filter(&tedge_config), &mut mqtt_actor);
+        aws_converting_actor.connect_source(get_topic_filter(&tedge_config), &mut mqtt_actor);
         aws_converting_actor.register_peer(NoConfig, mqtt_actor.get_sender());
 
         runtime.spawn(aws_converting_actor).await?;

crates/core/tedge_mapper/src/az/mapper.rs

@@ -38,7 +38,7 @@ impl TEdgeComponent for AzureMapper {
             tedge_config.az.mapper.timestamp_format,
         );
         let mut az_converting_actor = ConvertingActor::builder("AzConverter", az_converter);
-        az_converting_actor.add_input(get_topic_filter(&tedge_config), &mut mqtt_actor);
+        az_converting_actor.connect_source(get_topic_filter(&tedge_config), &mut mqtt_actor);
 
         az_converting_actor.register_peer(NoConfig, mqtt_actor.get_sender());
 

crates/core/tedge_mapper/src/collectd/mapper.rs

@@ -48,7 +48,7 @@ impl TEdgeComponent for CollectdMapper {
         let mut collectd_actor = CollectdActorBuilder::new(input_topic);
 
         collectd_actor.add_input(&mut mqtt_actor);
-        batching_actor.add_input(NoConfig, &mut collectd_actor);
+        batching_actor.connect_source(NoConfig, &mut collectd_actor);
         mqtt_actor.add_mapped_input(NoConfig, &mut batching_actor, move |batch| {
             collectd_ext::converter::batch_into_mqtt_messages(&output_topic, batch).into_iter()
         });

crates/extensions/c8y_mapper_ext/src/compatibility_adapter.rs

@@ -42,7 +42,7 @@ impl OldAgentAdapter {
         mqtt: &mut (impl MessageSource<MqttMessage, TopicFilter> + MessageSink<MqttMessage>),
     ) -> ConvertingActorBuilder<OldAgentAdapter> {
         let mut builder = ConvertingActor::builder("OldAgentAdapter", OldAgentAdapter);
-        builder.add_input(old_and_new_command_topics(), mqtt);
+        builder.connect_source(old_and_new_command_topics(), mqtt);
         builder.add_sink(NoConfig, mqtt);
         builder
     }