Added PubSub client init success/failure handling and channel messages.

dmackdev · Sep 21, 2023 · d17c70e · d17c70e
1 parent 0f5ccfb
commit d17c70e
Show file tree

Hide file tree

Showing 3 changed files with 36 additions and 13 deletions.
diff --git a/pubsubman/src/app.rs b/pubsubman/src/app.rs
@@ -42,7 +42,9 @@ impl App {
         let (back_tx, back_rx) = tokio::sync::mpsc::channel(10);
 
         std::thread::spawn(|| {
-            Backend::new(back_tx, front_rx, emulator_project_id).init();
+            if let Ok(mut backend) = Backend::new(back_tx, front_rx, emulator_project_id) {
+                backend.init();
+            };
         });
 
         refresh_topics(&front_tx, None);
@@ -67,6 +69,7 @@ impl App {
     fn handle_backend_message(&mut self) {
         match self.back_rx.try_recv() {
             Ok(message) => match message {
+                BackendMessage::ClientInitialised => {}
                 BackendMessage::TopicsUpdated(topic_names) => {
                     self.topic_names = topic_names;
                     refresh_topics(&self.front_tx, Some(5000));

diff --git a/pubsubman_backend/src/lib.rs b/pubsubman_backend/src/lib.rs
@@ -1,4 +1,4 @@
-use std::sync::Arc;
+use std::{error::Error, sync::Arc};
 
 use futures_util::StreamExt;
 use google_cloud_gax::conn::Environment;
@@ -34,20 +34,38 @@ impl Backend {
         back_tx: Sender<BackendMessage>,
         front_rx: Receiver<FrontendMessage>,
         emulator_project_id: Option<String>,
-    ) -> Self {
+    ) -> Result<Self, Box<dyn Error>> {
         let rt = Builder::new_multi_thread()
             .worker_threads(4)
             .enable_all()
             .build()
             .unwrap();
 
-        let client = rt.block_on(async { create_client(emulator_project_id).await });
-
-        Self {
-            back_tx,
-            front_rx,
-            client: Arc::new(client),
-            rt,
+        match rt.block_on(async { create_client(emulator_project_id).await }) {
+            Ok(client) => {
+                let back_tx_clone = back_tx.clone();
+                rt.spawn(async move {
+                    back_tx_clone
+                        .send(BackendMessage::ClientInitialised)
+                        .await
+                        .unwrap();
+                });
+                Ok(Self {
+                    back_tx,
+                    front_rx,
+                    client: Arc::new(client),
+                    rt,
+                })
+            }
+            Err(err) => {
+                rt.spawn(async move {
+                    back_tx
+                        .send(BackendMessage::Error(BackendError::ClientInitFailed))
+                        .await
+                        .unwrap();
+                });
+                Err(err)
+            }
         }
     }
 
@@ -217,14 +235,14 @@ impl Backend {
     }
 }
 
-async fn create_client(emulator_project_id: Option<String>) -> Client {
-    let mut config = ClientConfig::default().with_auth().await.unwrap();
+async fn create_client(emulator_project_id: Option<String>) -> Result<Client, Box<dyn Error>> {
+    let mut config = ClientConfig::default().with_auth().await?;
 
     if let (Environment::Emulator(_), Some(emulator_project_id)) =
         (&config.environment, emulator_project_id)
     {
         config.project_id = Some(emulator_project_id);
     }
 
-    Client::new(config).await.unwrap()
+    Ok(Client::new(config).await?)
 }
diff --git a/pubsubman_backend/src/message.rs b/pubsubman_backend/src/message.rs
@@ -13,6 +13,7 @@ pub enum FrontendMessage {
 
 #[derive(Debug)]
 pub enum BackendMessage {
+    ClientInitialised,
     TopicsUpdated(Vec<TopicName>),
     SubscriptionCreated(TopicName, SubscriptionName),
     MessageReceived(TopicName, PubsubMessage),
@@ -22,6 +23,7 @@ pub enum BackendMessage {
 
 #[derive(Debug)]
 pub enum BackendError {
+    ClientInitFailed,
     GetTopicsFailed,
     CreateSubscriptionFailed(TopicName),
     StreamMessagesFailed(TopicName, SubscriptionName),