refactor transports to start themselves

1 files changed, 65 insertions, 8 deletions

diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs
index d33e323..836eca9 100644
--- a/src/transport/mycelium/mod.rs
+++ b/src/transport/mycelium/mod.rs
@@ -91,6 +91,70 @@ pub(crate) struct MyceliumTransport {
     local_id: String,
 }
 
+/// The transport as `settings` configure it (`transport::start`): its daemon supervised
+/// (started, which can take a while) or one running elsewhere. None if `stop` came
+/// first.
+pub(crate) async fn start(
+    settings: &crate::config::Settings,
+    mut stop: tokio::sync::watch::Receiver<bool>,
+) -> Result<Option<MyceliumTransport>> {
+    let spawn_config = MyceliumSpawnConfig {
+        rest_url_opt: settings.mycelium_rest_url(),
+        key_file_opt: settings.mycelium_key_file.clone(),
+        network_key_file: settings.mycelium_network_key_file.clone(),
+        network_name: settings.mycelium_network_name.clone(),
+        tcp_port: settings.mycelium_tcp_listen_port,
+        peers: settings.mycelium_peers.clone(),
+    };
+    let setup = tokio::select! {
+        setup = MyceliumSetup::bootstrap(spawn_config) => setup?,
+        _ = stop.wait_for(|stop| *stop) => return Ok(None),
+    };
+    // Without a password it may carry (the option is secret).
+    let mut shown_url = setup.rest_url.clone();
+    let _ = shown_url.set_password(None);
+    tracing::debug!(url = shown_url.to_string(), "Using Mycelium REST API");
+    tokio::select! {
+        transport = MyceliumTransport::connect(setup) => Ok(Some(transport?)),
+        _ = stop.wait_for(|stop| *stop) => Ok(None),
+    }
+}
+
+/// How long the daemon's API may take to come up: a daemon just started listens a
+/// moment later (`admin_info`).
+const API_READY_WITHIN: Duration = Duration::from_secs(30);
+
+/// The daemon's identity, from its API at `url` (`/api/v1/admin`), once the API answers:
+/// a daemon we just started doesn't listen yet, so refused connections are tried again
+/// (50 ms, doubling up to a second) for `API_READY_WITHIN`; anything else is an error.
+async fn admin_info(client: &reqwest::Client, url: Url) -> Result<MyceliumAdminInfo> {
+    let deadline = tokio::time::Instant::now() + API_READY_WITHIN;
+    let mut wait = Duration::from_millis(50);
+    loop {
+        match client.get(url.clone()).send().await {
+            Ok(response) => {
+                return response
+                    .error_for_status()?
+                    .json()
+                    .await
+                    .context("Querying local Mycelium identity");
+            }
+            Err(error) if error.is_connect() && tokio::time::Instant::now() + wait < deadline => {
+                tracing::debug!(url = %url, error = %error, "Mycelium's API doesn't answer yet; trying again");
+                tokio::time::sleep(wait).await;
+                wait = (wait * 2).min(Duration::from_secs(1));
+            }
+            Err(error) => {
+                return Err(error).with_context(|| {
+                    format!(
+                        "Querying local Mycelium identity (no answer within {API_READY_WITHIN:?})"
+                    )
+                });
+            }
+        }
+    }
+}
+
 impl MyceliumTransport {
     pub(crate) async fn connect(setup: MyceliumSetup) -> Result<Self> {
         let rest_url = setup.rest_url.clone();
@@ -102,14 +166,7 @@ impl MyceliumTransport {
             .timeout(Duration::from_secs(MYCELIUM_POLL_TIMEOUT_SECS + 30))
             .build()?;
 
-        let admin: MyceliumAdminInfo = client
-            .get(rest_url.join("/api/v1/admin")?)
-            .send()
-            .await?
-            .error_for_status()?
-            .json()
-            .await
-            .context("Querying local Mycelium identity")?;
+        let admin = admin_info(&client, rest_url.join("/api/v1/admin")?).await?;
 
         let transport = Self {
             setup: tokio::sync::Mutex::new(setup),