feat: stream error handling

1 files changed, 32 insertions, 8 deletions

diff --git a/filamento/src/logic/mod.rs b/filamento/src/logic/mod.rs
index 638f682..365a0df 100644
--- a/filamento/src/logic/mod.rs
+++ b/filamento/src/logic/mod.rs
@@ -3,8 +3,9 @@ use std::{collections::HashMap, sync::Arc};
 use lampada::{Logic, error::ReadError};
 use stanza::client::Stanza;
 use tokio::sync::{Mutex, mpsc, oneshot};
+use tracing::{error, info, warn};
 
-use crate::{Command, UpdateMessage, db::Db};
+use crate::{Client, Command, UpdateMessage, db::Db, error::Error};
 
 mod abort;
 mod connect;
@@ -16,6 +17,7 @@ mod process_stanza;
 
 #[derive(Clone)]
 pub struct ClientLogic {
+    client: Client,
     db: Db,
     pending: Arc<Mutex<HashMap<String, oneshot::Sender<Result<Stanza, ReadError>>>>>,
     update_sender: mpsc::Sender<UpdateMessage>,
@@ -23,6 +25,7 @@ pub struct ClientLogic {
 
 impl ClientLogic {
     pub fn new(
+        client: Client,
         db: Db,
         pending: Arc<Mutex<HashMap<String, oneshot::Sender<Result<Stanza, ReadError>>>>>,
         update_sender: mpsc::Sender<UpdateMessage>,
@@ -31,9 +34,14 @@ impl ClientLogic {
             db,
             pending,
             update_sender,
+            client,
         }
     }
 
+    pub fn client(&self) -> &Client {
+        &self.client
+    }
+
     pub fn db(&self) -> &Db {
         &self.db
     }
@@ -45,6 +53,23 @@ impl ClientLogic {
     pub fn update_sender(&self) -> &mpsc::Sender<UpdateMessage> {
         &self.update_sender
     }
+
+    pub async fn handle_unsupported(&self, stanza: impl Into<Stanza>) {
+        let stanza: Stanza = stanza.into();
+        warn!("received unsupported stanza: {:?}", stanza);
+        self.handle_update(UpdateMessage::Unsupported(stanza)).await;
+    }
+
+    pub async fn handle_update(&self, update: UpdateMessage) {
+        // TODO: impl fmt
+        info!("{:?}", update);
+        self.update_sender().send(update).await;
+    }
+
+    pub async fn handle_error(&self, e: Error) {
+        error!("{}", e);
+        self.handle_update(UpdateMessage::Error(e)).await;
+    }
 }
 
 impl Logic for ClientLogic {
@@ -63,13 +88,8 @@ impl Logic for ClientLogic {
         disconnect::handle_disconnect(self, connection).await;
     }
 
-    async fn handle_stanza(
-        self,
-        stanza: ::stanza::client::Stanza,
-        connection: lampada::Connected,
-        supervisor: lampada::SupervisorSender,
-    ) {
-        process_stanza::handle_stanza(self, stanza, connection, supervisor).await;
+    async fn handle_stanza(self, stanza: ::stanza::client::Stanza, connection: lampada::Connected) {
+        process_stanza::handle_stanza(self, stanza, connection).await;
     }
 
     async fn handle_online(self, command: Self::Cmd, connection: lampada::Connected) {
@@ -87,4 +107,8 @@ impl Logic for ClientLogic {
     async fn handle_connection_error(self, error: lampada::error::ConnectionError) {
         connection_error::handle_connection_error(self, error).await;
     }
+
+    async fn handle_stream_error(self, stream_error: stanza::stream::Error) {
+        self.handle_error(Error::Stream(stream_error)).await;
+    }
 }