WIP

Signed-off-by: Nicolas Buffon <nicolas.buffon@orange.com>

Author: nbuffon

rust/src/client/application/pipeline.rs

@@ -362,9 +362,9 @@ where
             for topic in topic_list.iter() {
                 match topic {
                     info_topic if info_topic.to_string().contains(Information::TYPE) => {
-                        router.add_route(info_topic.clone(), deserialize::<Information>);
+                        router.add_route(info_topic.clone(), deserialize::<T, Information>);
                     }
-                    _ => router.add_route(topic.clone(), deserialize::<Exchange>),
+                    _ => router.add_route(topic.clone(), deserialize::<T, Exchange>),
                 }
             }
 
@@ -424,15 +424,15 @@ where
     )
 }
 
-fn deserialize<T>(publish: rumqttc::v5::mqttbytes::v5::Publish) -> Option<BoxedReception>
+fn deserialize<T, P>(publish: rumqttc::v5::mqttbytes::v5::Publish) -> Option<Packet<T, P>>
 where
-    T: DeserializeOwned + Payload + 'static + Send,
+    T: Topic + 'static,
+    P: DeserializeOwned + Payload + 'static + Send,
 {
-    // Incoming publish from the broker
     match String::from_utf8(publish.payload.to_vec()) {
         Ok(message) => {
             let message_str = message.as_str();
-            match serde_json::from_str::<T>(message_str) {
+            match serde_json::from_str::<P>(message_str) {
                 Ok(message) => {
                     trace!("message parsed");
                     return Some((Box::new(message), publish.properties.unwrap_or_default()));

rust/src/transport/mqtt/mqtt_router.rs

@@ -18,30 +18,32 @@ use rumqttc::v5::{Event, Incoming};
 use crate::transport::mqtt::topic::Topic;
 use std::any::{type_name, Any};
 use std::str::from_utf8;
+use crate::transport::packet::Packet;
+use crate::transport::payload::Payload;
 
-pub type BoxedReception = (Box<dyn Any + 'static + Send>, PublishProperties);
+// pub type BoxedReception = (Box<dyn Any + 'static + Send>, PublishProperties);
 
-type BoxedCallback = Box<dyn Fn(Publish) -> Option<BoxedReception>>;
+// type BoxedCallback = Box<dyn Fn(Publish) -> Option<Packet<dyn Topic, dyn Payload>>>;
 
 #[cfg(feature = "telemetry")]
 use crate::transport::telemetry::get_reception_mqtt_span;
 
 #[derive(Default)]
-pub struct MqttRouter {
-    route_map: HashMap<String, BoxedCallback>,
+pub struct MqttRouter<T: Topic> {
+    route_map: HashMap<String, Box<dyn Fn(Publish) -> Option<Packet<T, dyn Any + Send + 'static>>>>,
 }
 
-impl MqttRouter {
-    pub fn add_route<T, C>(&mut self, topic: T, callback: C)
+impl<T> MqttRouter<T> {
+    pub fn add_route<P, C>(&mut self, topic: T, callback: C)
     where
-        T: Topic,
-        C: Fn(Publish) -> Option<BoxedReception> + 'static,
+        P: Payload,
+        C: Fn(Publish) -> Option<Packet<T, P>> + 'static,
     {
         self.route_map.insert(topic.as_route(), Box::new(callback));
         info!("Registered route for topic: {}", topic.as_route());
     }
 
-    pub fn handle_event<T: Topic>(&mut self, event: Event) -> Option<(T, BoxedReception)> {
+    pub fn handle_event<P: Payload>(&mut self, event: Event) -> Option<Packet<T, P>> {
         match event {
             Event::Incoming(incoming) => match incoming {
                 Incoming::Publish(publish) => {