Merge branch 'main' of github.com:holochain/holochain-client-rust

Cargo.toml

@@ -9,32 +9,33 @@ name = "holochain_client"
 readme = "README.md"
 repository = "https://github.com/holochain/holochain-client-rust"
 resolver = "2"
-version = "0.4.8"
+version = "0.4.10"
 
 [workspace]
 members = ["fixture/zomes/foo"]
 
 [workspace.dependencies]
-holochain_zome_types = "0.2.6"
+holochain_zome_types = "0.2.7"
 
 [dependencies]
 again = "0.1"
 anyhow = "1.0"
 ed25519-dalek = { version = "2.1", features = ["rand_core"] }
+event-emitter-rs = "0.1"
 holo_hash = { version = "0.2.6", features = ["encoding"] }
-holochain_conductor_api = "0.2.6"
+holochain_conductor_api = "0.2.7"
 holochain_serialized_bytes = "0.0.53"
-holochain_state = "0.2.6"
-holochain_types = "0.2.6"
-holochain_websocket = "0.2.6"
+holochain_state = "0.2.7"
+holochain_types = "0.2.7"
+holochain_websocket = "0.2.7"
 holochain_zome_types = { workspace = true }
 serde = ">=1.0.0, <=1.0.166"
 url = "2.2"
 rand = "0.8"
 async-trait = "0.1"
 parking_lot = "0.12.1"
+tokio = { version = "1.36", features = ["rt"] }
 
 [dev-dependencies]
 arbitrary = "1.2"
-holochain = { version = "0.2.6", features = ["test_utils"] }
-tokio = { version = "1.3", features = ["full"] }
+holochain = { version = "0.2.7", features = ["test_utils"] }

src/admin_websocket.rs

@@ -1,22 +1,22 @@
-use std::sync::Arc;
-
-use anyhow::{Context, Result};
+use crate::error::{ConductorApiError, ConductorApiResult};
+use anyhow::Result;
 use holo_hash::DnaHash;
 use holochain_conductor_api::{AdminRequest, AdminResponse, AppInfo, AppStatusFilter, StorageInfo};
 use holochain_types::{
     dna::AgentPubKey,
-    prelude::{CellId, DeleteCloneCellPayload, InstallAppPayload},
+    prelude::{CellId, DeleteCloneCellPayload, InstallAppPayload, UpdateCoordinatorsPayload},
+};
+use holochain_websocket::{connect, WebsocketConfig, WebsocketSender};
+use holochain_zome_types::{
+    capability::GrantedFunctions,
+    prelude::{DnaDef, GrantZomeCallCapabilityPayload, Record},
 };
-use holochain_websocket::{connect, WebsocketConfig, WebsocketReceiver, WebsocketSender};
-use holochain_zome_types::{DnaDef, GrantZomeCallCapabilityPayload, GrantedFunctions, Record};
 use serde::{Deserialize, Serialize};
+use std::{net::ToSocketAddrs, sync::Arc};
 use url::Url;
 
-use crate::error::{ConductorApiError, ConductorApiResult};
-
 pub struct AdminWebsocket {
     tx: WebsocketSender,
-    rx: WebsocketReceiver,
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
@@ -33,21 +33,36 @@ pub struct AuthorizeSigningCredentialsPayload {
 
 impl AdminWebsocket {
     pub async fn connect(admin_url: String) -> Result<Self> {
-        let url = Url::parse(&admin_url).context("invalid ws:// URL")?;
-        let websocket_config = Arc::new(WebsocketConfig::default());
-        let (tx, rx) = again::retry(|| {
+        let url = Url::parse(&admin_url)?;
+        let host = url
+            .host_str()
+            .expect("websocket url does not have valid host part");
+        let port = url.port().expect("websocket url does not have valid port");
+        let admin_addr = format!("{}:{}", host, port);
+        let addr = admin_addr
+            .to_socket_addrs()?
+            .find(|addr| addr.is_ipv4())
+            .expect("no valid ipv4 websocket addresses found");
+
+        // app installation takes > 2 min on CI at the moment, hence the high
+        // request timeout
+        let websocket_config = WebsocketConfig {
+            default_request_timeout: std::time::Duration::from_secs(180),
+            ..Default::default()
+        };
+        let websocket_config = Arc::new(websocket_config);
+
+        let (tx, mut rx) = again::retry(|| {
             let websocket_config = Arc::clone(&websocket_config);
-            connect(url.clone().into(), websocket_config)
+            connect(websocket_config, addr)
         })
         .await?;
 
-        Ok(Self { tx, rx })
-    }
+        // WebsocketReceiver needs to be polled in order to receive responses
+        // from remote to sender requests.
+        tokio::task::spawn(async move { while rx.recv::<AdminResponse>().await.is_ok() {} });
 
-    pub fn close(&mut self) {
-        if let Some(h) = self.rx.take_handle() {
-            h.close()
-        }
+        Ok(Self { tx })
     }
 
     pub async fn generate_agent_pub_key(&mut self) -> ConductorApiResult<AgentPubKey> {
@@ -183,6 +198,18 @@ impl AdminWebsocket {
         }
     }
 
+    pub async fn update_coordinators(
+        &mut self,
+        update_coordinators_payload: UpdateCoordinatorsPayload,
+    ) -> ConductorApiResult<()> {
+        let msg = AdminRequest::UpdateCoordinators(Box::new(update_coordinators_payload));
+        let response = self.send(msg).await?;
+        match response {
+            AdminResponse::CoordinatorsUpdated => Ok(()),
+            _ => unreachable!("Unexpected response {:?}", response),
+        }
+    }
+
     pub async fn graft_records(
         &mut self,
         cell_id: CellId,

src/app_agent_websocket.rs

@@ -1,18 +1,21 @@
-use std::{ops::DerefMut, sync::Arc};
-
 use crate::{
     signing::{sign_zome_call, AgentSigner},
     AppWebsocket, ConductorApiError, ConductorApiResult,
 };
 use anyhow::{anyhow, Result};
 use holo_hash::AgentPubKey;
-use holochain_conductor_api::{AppInfo, CellInfo, ClonedCell, ProvisionedCell};
+use holochain_conductor_api::{AppInfo, CellInfo, ProvisionedCell};
 use holochain_state::nonce::fresh_nonce;
-use holochain_types::{app::InstalledAppId, prelude::CloneId};
-use holochain_zome_types::prelude::{
-    CellId, ExternIO, FunctionName, RoleName, Timestamp, ZomeCallUnsigned, ZomeName,
+use holochain_types::{
+    app::InstalledAppId,
+    prelude::{CloneId, Signal},
+};
+use holochain_zome_types::{
+    clone::ClonedCell,
+    prelude::{CellId, ExternIO, FunctionName, RoleName, Timestamp, ZomeCallUnsigned, ZomeName},
 };
 use std::ops::Deref;
+use std::{ops::DerefMut, sync::Arc};
 
 #[derive(Clone)]
 pub struct AppAgentWebsocket {
@@ -51,6 +54,33 @@ impl AppAgentWebsocket {
         })
     }
 
+    pub async fn on_signal<F: Fn(Signal) + 'static + Sync + Send>(
+        &mut self,
+        handler: F,
+    ) -> Result<String> {
+        let app_info = self.app_info.clone();
+        self.app_ws
+            .on_signal(move |signal| {
+                if let Signal::App {
+                    cell_id,
+                    zome_name: _,
+                    signal: _,
+                } = signal.clone()
+                {
+                    if app_info.cell_info.values().any(|cells| {
+                        cells.iter().any(|cell_info| match cell_info {
+                            CellInfo::Provisioned(cell) => cell.cell_id.eq(&cell_id),
+                            CellInfo::Cloned(cell) => cell.cell_id.eq(&cell_id),
+                            _ => false,
+                        })
+                    }) {
+                        handler(signal);
+                    }
+                }
+            })
+            .await
+    }
+
     pub async fn call_zome(
         &mut self,
         target: ZomeCallTarget,

src/app_websocket.rs

@@ -1,36 +1,76 @@
-use std::sync::Arc;
-
-use anyhow::{Context, Result};
-use holochain_conductor_api::{
-    AppInfo, AppRequest, AppResponse, ClonedCell, NetworkInfo, ZomeCall,
-};
+use crate::error::{ConductorApiError, ConductorApiResult};
+use anyhow::Result;
+use event_emitter_rs::EventEmitter;
+use holochain_conductor_api::{AppInfo, AppRequest, AppResponse, NetworkInfo, ZomeCall};
 use holochain_types::{
     app::InstalledAppId,
     prelude::{
         CreateCloneCellPayload, DisableCloneCellPayload, EnableCloneCellPayload, ExternIO,
         NetworkInfoRequestPayload,
     },
+    signal::Signal,
 };
 use holochain_websocket::{connect, WebsocketConfig, WebsocketSender};
+use holochain_zome_types::clone::ClonedCell;
+use std::{net::ToSocketAddrs, sync::Arc};
+use tokio::sync::Mutex;
 use url::Url;
 
-use crate::error::{ConductorApiError, ConductorApiResult};
-
 #[derive(Clone)]
 pub struct AppWebsocket {
     tx: WebsocketSender,
+    event_emitter: Arc<Mutex<EventEmitter>>,
 }
 
 impl AppWebsocket {
     pub async fn connect(app_url: String) -> Result<Self> {
-        let url = Url::parse(&app_url).context("invalid ws:// URL")?;
+        let url = Url::parse(&app_url)?;
+        let host = url
+            .host_str()
+            .expect("websocket url does not have valid host part");
+        let port = url.port().expect("websocket url does not have valid port");
+        let app_addr = format!("{}:{}", host, port);
+        let addr = app_addr
+            .to_socket_addrs()?
+            .find(|addr| addr.is_ipv4())
+            .expect("no valid ipv4 websocket addresses found");
+
         let websocket_config = Arc::new(WebsocketConfig::default());
-        let (tx, _rx) = again::retry(|| {
+        let (tx, mut rx) = again::retry(|| {
             let websocket_config = Arc::clone(&websocket_config);
-            connect(url.clone().into(), websocket_config)
+            connect(websocket_config, addr)
         })
         .await?;
-        Ok(Self { tx })
+
+        let event_emitter = EventEmitter::new();
+        let mutex = Arc::new(Mutex::new(event_emitter));
+
+        tokio::task::spawn({
+            let mutex = mutex.clone();
+            async move {
+                while let Ok(msg) = rx.recv::<AppResponse>().await {
+                    if let holochain_websocket::ReceiveMessage::Signal(signal_bytes) = msg {
+                        let mut event_emitter = mutex.lock().await;
+                        let signal = Signal::try_from_vec(signal_bytes).expect("Malformed signal");
+                        event_emitter.emit("signal", signal);
+                    }
+                }
+            }
+        });
+
+        Ok(Self {
+            tx,
+            event_emitter: mutex,
+        })
+    }
+
+    pub async fn on_signal<F: Fn(Signal) + 'static + Sync + Send>(
+        &mut self,
+        handler: F,
+    ) -> Result<String> {
+        let mut event_emitter = self.event_emitter.lock().await;
+        let id = event_emitter.on("signal", handler);
+        Ok(id)
     }
 
     pub async fn app_info(

src/error.rs

@@ -1,6 +1,6 @@
 use holochain_conductor_api::ExternalApiWireError;
 use holochain_state::prelude::DatabaseError;
-use holochain_websocket::WebsocketError;
+use holochain_websocket::Error as WebsocketError;
 
 #[derive(Debug)]
 pub enum ConductorApiError {