FIX: Message format

Author: BryanFRD

src/datas/mod.rs

@@ -1,15 +1,12 @@
 use serde_json::{json, Value};
+use tokio::sync::broadcast;
 
 use crate::events::Event;
 
 pub mod docker;
 
-pub fn create_event_dto(event: Event, data: Value) -> String {
-  let mut event_json = serde_json::to_value(event).unwrap_or_else(|_| json!({}));
-  if let Value::Object(ref mut map) = event_json {
-    map.insert("data".to_string(), data);
-  }
-  event_json.to_string()
+pub fn create_event_dto(event: Event) -> String {
+  serde_json::to_string(&event).unwrap_or_else(|_| "".to_string())
 }
 
 pub trait EventDTO {
@@ -20,4 +17,24 @@ impl EventDTO for i8 {
   fn to_json(&self) -> Value {
     json!(*self)
   }
+}
+
+pub trait SendEvent {
+  async fn send_event(&mut self, event: Event);
+}
+
+impl SendEvent for wtransport::SendStream {
+  async fn send_event(&mut self, event: Event) {
+    if let Err(error) = self.write_all(create_event_dto(event).as_bytes()).await {
+      log::error!("Failed to send event: {:?}", error);
+    }
+  }
+}
+
+impl SendEvent for broadcast::Sender<String> {
+  async fn send_event(&mut self, event: Event) {
+    if let Err(error) = self.send(create_event_dto(event)) {
+      log::error!("Failed to send event: {:?}", error);
+    }
+  }
 }

src/errors/error_status.rs

@@ -3,12 +3,17 @@ use serde::{Deserialize, Serialize};
 #[derive(Serialize, Deserialize, Debug)]
 #[serde(tag = "type")]
 pub enum DockerEvent {
-  DockerStatus,
+  DockerStatus { data: DockerStatusData },
   DockerContainersRestart { data: DockerContainersRestartData }
 }
 
+#[derive(Serialize, Deserialize, Debug)]
+pub struct DockerStatusData {
+  pub status: Option<i8>
+}
+
 #[derive(Serialize, Deserialize, Debug)]
 pub struct DockerContainersRestartData {
   #[serde(rename = "containerId")]
-  pub container_id: String
+  pub container_id: Option<String>
 }

src/errors/mod.rs

@@ -1,6 +1,5 @@
 mod events;
 mod datas;
-mod errors;
 mod webtransport;
 mod services;
 

src/events/docker.rs

@@ -5,18 +5,25 @@ use futures::StreamExt;
 use serde_json::json;
 use tokio::sync::broadcast;
 
+use crate::{datas::SendEvent, events::{docker::{DockerEvent, DockerStatusData}, Event}};
+
 const INTERVAL: Duration = Duration::from_secs(10);
 
 pub fn get_docker_client() -> Result<Docker, Error> {
     Docker::connect_with_socket_defaults()
 }
 
-pub async fn listen_docker_events(tx: broadcast::Sender<String>) {
+pub async fn listen_docker_events(mut tx: broadcast::Sender<String>) {
     let docker = loop {
         match get_docker_client() {
             Ok(client) => break client,
             Err(error) => {
                 log::error!("Failed to connect to Docker, retrying in {:?}: {:?}", INTERVAL, error);
+                tx.send_event(Event::Docker(DockerEvent::DockerStatus {
+                    data: DockerStatusData {
+                        status: Some(0)
+                    }
+                })).await;
                 sleep(INTERVAL);
             }
         }
@@ -29,7 +36,6 @@ pub async fn listen_docker_events(tx: broadcast::Sender<String>) {
         match event {
             Ok(event) => {
                 let event_str = json!(event).to_string();
-                log::info!("Docker event: [{:?}:{:?}]", event.action, event.typ);
 
                 match tx.send(event_str){
                     Ok(_) => {},

src/main.rs

@@ -1,11 +1,18 @@
-use crate::{datas::{create_event_dto, EventDTO}, events::{docker::DockerEvent, Event}, services::docker};
+use crate::{datas::SendEvent, events::{docker::{DockerContainersRestartData, DockerEvent, DockerStatusData}, Event}, services::docker};
 
 pub async fn handle_message(send_stream: &mut wtransport::SendStream, event: &DockerEvent) {
     match event {
-        DockerEvent::DockerStatus => {
-            if let Err(error) = send_stream.write_all(create_event_dto(Event::Docker(DockerEvent::DockerStatus), docker::ping().await.to_json()).as_bytes()).await {
-                log::error!("Failed to send event: {:?}", error);
-            };
+        DockerEvent::DockerStatus { .. } => {
+            send_stream.send_event(Event::Docker(DockerEvent::DockerStatus {
+                data: DockerStatusData {
+                    status: Some(docker::ping().await)
+                }
+            })).await;
+            log::info!("{:?}", Event::Docker(DockerEvent::DockerContainersRestart {
+                data: DockerContainersRestartData {
+                    container_id: Some("container_id".to_string()),
+                }
+            }));
         },
         DockerEvent::DockerContainersRestart { data } => {
             log::info!("Restarting containers: {:?}", data);

src/services/docker.rs

@@ -3,10 +3,10 @@ use std::sync::Arc;
 use tokio::sync::{broadcast, Mutex};
 use wtransport::{endpoint::IncomingSession, Connection, Endpoint, Identity, ServerConfig};
 use crate::services;
-use crate::events::{Event, system::SystemEvent};
+use crate::events::Event;
 
-mod system;
-mod docker;
+pub mod system;
+pub mod docker;
 
 pub async fn start_webtransport() -> Result<(), Box<dyn Error + Send + Sync>> {
     let identity = match Identity::load_pemfiles("localhost.pem", "localhost-key.pem").await {
@@ -30,10 +30,9 @@ pub async fn start_webtransport() -> Result<(), Box<dyn Error + Send + Sync>> {
         }
     };
 
-    log::info!("Listening on port 4433");
-    
     let (tx, _rx) = broadcast::channel::<String>(100);
 
+    //TODO handle errors
     tokio::spawn(services::docker::listen_docker_events(tx.clone()));
 
     loop {