Add a config parameter to MessageSink::add_input

This is first step toward removing the Source type parameter to
MessageSink. Indeed, this type parameter and the get_config method
prevents a sink to receive messages from several sources.

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

Author: didier-wenzek

crates/core/tedge_actors/src/builders.rs

@@ -109,12 +109,12 @@ pub trait MessageSink<M: Message, Config> {
     fn get_sender(&self) -> DynSender<M>;
 
     /// Add a source of messages to the actor under construction
-    fn add_input<N>(&mut self, source: &mut impl MessageSource<N, Config>)
+    fn add_input<N, C>(&mut self, config: C, source: &mut impl MessageSource<N, C>)
     where
         N: Message,
         M: From<N>,
     {
-        source.register_peer(self.get_config(), self.get_sender().sender_clone())
+        source.register_peer(config, self.get_sender().sender_clone())
     }
 
     /// Add a source of messages to the actor under construction, the messages being translated on the fly.
@@ -129,6 +129,7 @@ pub trait MessageSink<M: Message, Config> {
     /// # use tedge_actors::MessageReceiver;
     /// # use tedge_actors::MessageSink;
     /// # use tedge_actors::NoMessage;
+    /// # use tedge_actors::NoConfig;
     /// # use tedge_actors::Sender;
     /// # use tedge_actors::SimpleMessageBox;
     /// # use tedge_actors::SimpleMessageBoxBuilder;
@@ -142,7 +143,7 @@ pub trait MessageSink<M: Message, Config> {
     /// 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(&mut sender_builder, |str: &'static str| str.chars() );
+    /// receiver_builder.add_mapped_input(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();
@@ -161,9 +162,10 @@ pub trait MessageSink<M: Message, Config> {
     /// # Ok(())
     /// # }
     /// ```
-    fn add_mapped_input<N, MS, MessageMapper>(
+    fn add_mapped_input<N, C, MS, MessageMapper>(
         &mut self,
-        source: &mut impl MessageSource<N, Config>,
+        config: C,
+        source: &mut impl MessageSource<N, C>,
         cast: MessageMapper,
     ) where
         N: Message,
@@ -172,7 +174,7 @@ pub trait MessageSink<M: Message, Config> {
         MessageMapper: 'static + Send + Sync,
     {
         let sender = MappingSender::new(self.get_sender(), cast);
-        source.register_peer(self.get_config(), sender.into())
+        source.register_peer(config, sender.into())
     }
 }
 

crates/core/tedge_actors/src/converter.rs

@@ -217,7 +217,7 @@ impl<C: Converter> ConvertingActor<C> {
 ///         TopicFilter("some/mqtt/topics/#"));
 ///
 ///     // Connect this actor as a sink of the mqtt actor, to receive input messages from it
-///     converter_builder.add_input(&mut mqtt_builder);
+///     converter_builder.add_input(converter_builder.get_config(), &mut mqtt_builder);
 ///
 ///     // Connect the same mqtt actor as a sink of this actor, to send output messages to it
 ///     converter_builder.add_sink(&mut mqtt_builder);
@@ -272,13 +272,11 @@ impl<C: Converter, Config> MessageSource<C::Output, NoConfig>
     }
 }
 
-impl<C: Converter, Config: Clone, SourceConfig> MessageSink<C::Input, SourceConfig>
+impl<C: Converter, Config: Clone> MessageSink<C::Input, Config>
     for ConvertingActorBuilder<C, Config>
-where
-    SourceConfig: From<Config>,
 {
-    fn get_config(&self) -> SourceConfig {
-        self.config.clone().into()
+    fn get_config(&self) -> Config {
+        self.config.clone()
     }
 
     fn get_sender(&self) -> DynSender<C::Input> {

crates/core/tedge_agent/src/agent.rs

@@ -432,10 +432,14 @@ pub fn create_tedge_to_te_converter(
     .try_into()?;
 
     // Tedge to Te converter
-    let mut tedge_converter_actor =
-        ConvertingActor::builder("TedgetoTeConverter", tedge_to_te_converter, subscriptions);
+    let mut tedge_converter_actor: ConvertingActorBuilder<TedgetoTeConverter, TopicFilter> =
+        ConvertingActor::builder(
+            "TedgetoTeConverter",
+            tedge_to_te_converter,
+            subscriptions.clone(),
+        );
 
-    tedge_converter_actor.add_input(mqtt_actor_builder);
+    tedge_converter_actor.add_input(subscriptions, mqtt_actor_builder);
     tedge_converter_actor.add_sink(mqtt_actor_builder);
 
     Ok(tedge_converter_actor)

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

@@ -44,7 +44,7 @@ impl TEdgeComponent for AwsMapper {
             get_topic_filter(&tedge_config),
         );
 
-        aws_converting_actor.add_input(&mut mqtt_actor);
+        aws_converting_actor.add_input(aws_converting_actor.get_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

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

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

@@ -8,6 +8,7 @@ use mqtt_channel::Topic;
 use mqtt_channel::TopicFilter;
 use std::path::Path;
 use tedge_actors::MessageSink;
+use tedge_actors::NoConfig;
 use tedge_config::TEdgeConfig;
 
 const COLLECTD_MAPPER_NAME: &str = "tedge-mapper-collectd";
@@ -47,8 +48,8 @@ impl TEdgeComponent for CollectdMapper {
         let mut collectd_actor = CollectdActorBuilder::new(input_topic);
 
         collectd_actor.add_input(&mut mqtt_actor);
-        batching_actor.add_input(&mut collectd_actor);
-        mqtt_actor.add_mapped_input(&mut batching_actor, move |batch| {
+        batching_actor.add_input(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

@@ -46,7 +46,7 @@ impl OldAgentAdapter {
             OldAgentAdapter,
             old_and_new_command_topics(),
         );
-        builder.add_input(mqtt);
+        builder.add_input(builder.get_config(), mqtt);
         builder.add_sink(mqtt);
         builder
     }