sdk: add wait_for_ok to Options

crates/nostr-sdk/Cargo.toml

@@ -65,4 +65,11 @@ required-features = ["all-nips"]
 
 [[example]]
 name = "client-stop"
-required-features = ["all-nips"]
+required-features = ["all-nips"]
+
+[[example]]
+name = "shutdown-on-drop"
+
+[[example]]
+name = "wait-for-ok"
+required-features = ["nip19"]

crates/nostr-sdk/examples/shutdown-on-drop.rs

@@ -0,0 +1,35 @@
+// Copyright (c) 2022-2023 Yuki Kishimoto
+// Distributed under the MIT software license
+
+use std::time::Duration;
+
+use async_utility::thread;
+use nostr_sdk::prelude::*;
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    env_logger::init();
+
+    let my_keys = Keys::generate();
+
+    let opts = Options::new().shutdown_on_drop(true);
+    let client = Client::with_opts(&my_keys, opts);
+    client.add_relay("wss://relay.nostr.info", None).await?;
+    client.add_relay("wss://relay.damus.io", None).await?;
+
+    client.connect().await;
+
+    let c = client.clone();
+    thread::spawn(async move {
+        thread::sleep(Duration::from_secs(3)).await;
+        c.relays().await;
+        // First drop, dropping client...
+    });
+
+    thread::sleep(Duration::from_secs(10)).await;
+
+    // Try to publish a text note (will fail since the client is dropped)
+    client.publish_text_note("Hello world", &[]).await?;
+
+    Ok(())
+}

crates/nostr-sdk/examples/wait-for-ok.rs

@@ -0,0 +1,34 @@
+// Copyright (c) 2022-2023 Yuki Kishimoto
+// Distributed under the MIT software license
+
+use std::time::Duration;
+
+use nostr_sdk::prelude::*;
+
+const BECH32_SK: &str = "nsec1ufnus6pju578ste3v90xd5m2decpuzpql2295m3sknqcjzyys9ls0qlc85";
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    env_logger::init();
+
+    let secret_key = SecretKey::from_bech32(BECH32_SK)?;
+    let my_keys = Keys::new(secret_key);
+
+    let opts = Options::new()
+        .wait_for_ok(true)
+        .send_timeout(Some(Duration::from_secs(30)));
+    let client = Client::with_opts(&my_keys, opts);
+
+    // Paid relays that will not allow this public key to publish event
+    client.add_relay("wss://eden.nostr.land", None).await?;
+    client.add_relay("wss://nostr.wine", None).await?;
+
+    client.connect().await;
+
+    // Publish a text note
+    client
+        .publish_text_note("Hello from rust nostr-sdk", &[])
+        .await?;
+
+    Ok(())
+}

crates/nostr-sdk/src/client/mod.rs

@@ -7,8 +7,11 @@ use std::collections::HashMap;
 #[cfg(not(target_arch = "wasm32"))]
 use std::net::SocketAddr;
 use std::str::FromStr;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
 use std::time::Duration;
 
+use async_utility::thread;
 use nostr::event::builder::Error as EventBuilderError;
 use nostr::key::XOnlyPublicKey;
 #[cfg(feature = "nip46")]
@@ -112,10 +115,32 @@ pub struct Client {
     pool: RelayPool,
     keys: Keys,
     opts: Options,
+    dropped: Arc<AtomicBool>,
     #[cfg(feature = "nip46")]
     remote_signer: Option<RemoteSigner>,
 }
 
+impl Drop for Client {
+    fn drop(&mut self) {
+        if self.opts.shutdown_on_drop {
+            if self.dropped.load(Ordering::SeqCst) {
+                log::warn!("Client already dropped");
+            } else {
+                log::debug!("Dropping the Client...");
+                let _ = self
+                    .dropped
+                    .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |_| Some(true));
+                let pool = self.clone();
+                thread::spawn(async move {
+                    pool.shutdown()
+                        .await
+                        .expect("Impossible to drop the client")
+                });
+            }
+        }
+    }
+}
+
 impl Client {
     /// Create a new [`Client`]
     ///
@@ -145,6 +170,7 @@ impl Client {
             pool: RelayPool::new(opts.get_pool()),
             keys: keys.clone(),
             opts,
+            dropped: Arc::new(AtomicBool::new(false)),
             #[cfg(feature = "nip46")]
             remote_signer: None,
         }
@@ -167,6 +193,7 @@ impl Client {
             pool: RelayPool::new(opts.get_pool()),
             keys: app_keys.clone(),
             opts,
+            dropped: Arc::new(AtomicBool::new(false)),
             remote_signer: Some(remote_signer),
         }
     }
@@ -181,6 +208,11 @@ impl Client {
         self.keys.clone()
     }
 
+    /// Get [`RelayPool`]
+    pub fn pool(&self) -> RelayPool {
+        self.pool.clone()
+    }
+
     /// Get NIP46 uri
     #[cfg(feature = "nip46")]
     pub fn nostr_connect_uri(
@@ -222,7 +254,7 @@ impl Client {
 
     /// Completely shutdown [`Client`]
     pub async fn shutdown(self) -> Result<(), Error> {
-        Ok(self.pool.shutdown().await?)
+        Ok(self.pool.clone().shutdown().await?)
     }
 
     /// Clear already seen events
@@ -591,88 +623,57 @@ impl Client {
         Ok(())
     }
 
-    /// Send client message with custom wait
-    pub async fn send_msg_with_custom_wait(
-        &self,
-        msg: ClientMessage,
-        wait: Option<Duration>,
-    ) -> Result<(), Error> {
-        self.pool.send_msg(msg, wait).await?;
-        Ok(())
-    }
-
     /// Send client message to a specific relay
     pub async fn send_msg_to<S>(&self, url: S, msg: ClientMessage) -> Result<(), Error>
     where
         S: Into<String>,
     {
+        let url = Url::parse(&url.into())?;
         let wait: Option<Duration> = if self.opts.get_wait_for_send() {
             self.opts.get_send_timeout()
         } else {
             None
         };
-        self.send_msg_to_with_custom_wait(url, msg, wait).await
-    }
-
-    /// Send client message to a specific relay with custom wait
-    pub async fn send_msg_to_with_custom_wait<S>(
-        &self,
-        url: S,
-        msg: ClientMessage,
-        wait: Option<Duration>,
-    ) -> Result<(), Error>
-    where
-        S: Into<String>,
-    {
-        let url = Url::parse(&url.into())?;
-        self.pool.send_msg_to(url, msg, wait).await?;
-        Ok(())
+        Ok(self.pool.send_msg_to(url, msg, wait).await?)
     }
 
     /// Send event
     pub async fn send_event(&self, event: Event) -> Result<EventId, Error> {
         let event_id = event.id;
-        self.send_msg(ClientMessage::new_event(event)).await?;
+        if self.opts.get_wait_for_ok() {
+            let timeout: Option<Duration> = if self.opts.get_wait_for_send() {
+                self.opts.get_send_timeout()
+            } else {
+                None
+            };
+            self.pool.send_event(event, timeout).await?;
+        } else {
+            self.send_msg(ClientMessage::new_event(event)).await?;
+        }
         Ok(event_id)
     }
 
-    /// Send event with custom wait
-    pub async fn send_event_with_custom_wait(
-        &self,
-        event: Event,
-        wait: Option<Duration>,
-    ) -> Result<(), Error> {
-        self.pool
-            .send_msg(ClientMessage::new_event(event), wait)
-            .await?;
-        Ok(())
-    }
-
     /// Send event to specific relay
     pub async fn send_event_to<S>(&self, url: S, event: Event) -> Result<EventId, Error>
     where
         S: Into<String>,
     {
         let event_id = event.id;
-        self.send_msg_to(url, ClientMessage::new_event(event))
-            .await?;
+        if self.opts.get_wait_for_ok() {
+            let url = Url::parse(&url.into())?;
+            let timeout: Option<Duration> = if self.opts.get_wait_for_send() {
+                self.opts.get_send_timeout()
+            } else {
+                None
+            };
+            self.pool.send_event_to(url, event, timeout).await?;
+        } else {
+            self.send_msg_to(url, ClientMessage::new_event(event))
+                .await?;
+        }
         Ok(event_id)
     }
 
-    /// Send event to a specific relay with custom wait
-    pub async fn send_event_to_with_custom_wait<S>(
-        &self,
-        url: S,
-        event: Event,
-        wait: Option<Duration>,
-    ) -> Result<(), Error>
-    where
-        S: Into<String>,
-    {
-        self.send_msg_to_with_custom_wait(url, ClientMessage::new_event(event), wait)
-            .await
-    }
-
     async fn send_event_builder(&self, builder: EventBuilder) -> Result<EventId, Error> {
         #[cfg(feature = "nip46")]
         let event: Event = if let Some(signer) = self.remote_signer.as_ref() {

crates/nostr-sdk/src/client/options.rs

@@ -16,6 +16,8 @@ pub struct Options {
     wait_for_connection: Arc<AtomicBool>,
     /// Wait for the msg to be sent (default: true)
     wait_for_send: Arc<AtomicBool>,
+    /// Wait for `OK` relay msg (default: false)
+    wait_for_ok: Arc<AtomicBool>,
     /// Wait for the subscription msg to be sent (default: false)
     wait_for_subscription: Arc<AtomicBool>,
     /// POW difficulty for all events (default: 0)
@@ -29,6 +31,8 @@ pub struct Options {
     /// NIP46 timeout (default: 180 secs)
     #[cfg(feature = "nip46")]
     nip46_timeout: Option<Duration>,
+    /// Shutdown on [Client] drop
+    pub shutdown_on_drop: bool,
     /// Pool Options
     pool: RelayPoolOptions,
 }
@@ -38,13 +42,15 @@ impl Default for Options {
         Self {
             wait_for_connection: Arc::new(AtomicBool::new(false)),
             wait_for_send: Arc::new(AtomicBool::new(true)),
+            wait_for_ok: Arc::new(AtomicBool::new(false)),
             wait_for_subscription: Arc::new(AtomicBool::new(false)),
             difficulty: Arc::new(AtomicU8::new(0)),
             req_filters_chunk_size: Arc::new(AtomicU8::new(10)),
             timeout: None,
             send_timeout: Some(Duration::from_secs(60)),
             #[cfg(feature = "nip46")]
             nip46_timeout: Some(Duration::from_secs(180)),
+            shutdown_on_drop: false,
             pool: RelayPoolOptions::default(),
         }
     }
@@ -80,6 +86,18 @@ impl Options {
         self.wait_for_send.load(Ordering::SeqCst)
     }
 
+    /// Wait for `OK` relay msg
+    pub fn wait_for_ok(self, wait: bool) -> Self {
+        Self {
+            wait_for_ok: Arc::new(AtomicBool::new(wait)),
+            ..self
+        }
+    }
+
+    pub(crate) fn get_wait_for_ok(&self) -> bool {
+        self.wait_for_ok.load(Ordering::SeqCst)
+    }
+
     /// If set to `true`, `Client` wait that a subscription msg is sent before continue (`subscribe` and `unsubscribe` methods)
     pub fn wait_for_subscription(self, wait: bool) -> Self {
         Self {
@@ -157,6 +175,14 @@ impl Options {
         self.nip46_timeout
     }
 
+    /// Shutdown client on drop
+    pub fn shutdown_on_drop(self, value: bool) -> Self {
+        Self {
+            shutdown_on_drop: value,
+            ..self
+        }
+    }
+
     /// Set pool options
     pub fn pool(self, opts: RelayPoolOptions) -> Self {
         Self { pool: opts, ..self }
@@ -165,15 +191,4 @@ impl Options {
     pub(crate) fn get_pool(&self) -> RelayPoolOptions {
         self.pool
     }
-
-    /// Shutdown client on drop
-    pub fn shutdown_on_drop(self, value: bool) -> Self {
-        Self {
-            pool: RelayPoolOptions {
-                shutdown_on_drop: value,
-                ..self.pool
-            },
-            ..self
-        }
-    }
 }