Refactor/typestate connections

.gitignore

@@ -3,6 +3,7 @@
 **/.DS_Store
 **/node_modules
 **/*.node
+**/dist
 wrappers/ios/vcx/vcx.framework/**
 wrappers/ios/vcx/vcx.framework.dSYM/**
 wrappers/ios_legacy/vcx/vcx.framework/**

agents/node/vcxagent-core/package.json

@@ -46,7 +46,7 @@
     "test:integration:out-of-band": "jest --forceExit --env=node --runInBand test/out-of-band.spec.js",
     "test:integration:nonmediated-endpoint": "jest --forceExit --env=node --runInBand test/nonmediated-endpoint.spec.js",
     "test:integration:nonmediated-connection": "jest --forceExit --env=node --runInBand test/nonmediated-connection.spec.js"
- },
+  },
   "dependencies": {
     "axios": "^0.27.2",
     "ffi-napi": "^4.0.3",

agents/node/vcxagent-core/src/services/service-nonmediated-connections.js

@@ -47,7 +47,7 @@ module.exports.createServiceNonmediatedConnections = function createServiceNonme
     logger.info(`inviteeConnectionCreateFromInvite >> connectionId=${connectionId}, invite: ${invite}`)
     const connection = await NonmediatedConnection.createInvitee(invite)
     logger.debug(`InviteeConnectionSM after created from invitation:\n${JSON.stringify(connection.serialize())}`)
-    connection.processInvite(invite)
+    await connection.processInvite(invite)
     await connection.sendRequest(endpointInfo)
     logger.debug(`InviteeConnectionSM after sending request:\n${JSON.stringify(connection.serialize())}`)
     await saveNonmediatedConnection(connectionId, connection)

agents/rust/aries-vcx-agent/Cargo.toml

@@ -8,6 +8,7 @@ edition.workspace = true
 [dependencies]
 serde = "1.0.145"
 aries-vcx = { path = "../../../aries_vcx" }
+async-trait = "0.1.64"
 derive_builder = "0.11.2"
 serde_json = "1.0.85"
 log = "0.4.17"

agents/rust/aries-vcx-agent/src/http_client.rs

@@ -0,0 +1,13 @@
+use aries_vcx::{agency_client::httpclient::post_message, errors::error::VcxResult, transport::Transport};
+
+use async_trait::async_trait;
+
+pub struct HttpClient;
+
+#[async_trait]
+impl Transport for HttpClient {
+    async fn send_message(&self, msg: Vec<u8>, service_endpoint: &str) -> VcxResult<()> {
+        post_message(msg, service_endpoint).await?;
+        Ok(())
+    }
+}

agents/rust/aries-vcx-agent/src/lib.rs

@@ -9,6 +9,7 @@ extern crate uuid;
 
 mod agent;
 mod error;
+mod http_client;
 mod services;
 mod storage;
 

agents/rust/aries-vcx-agent/src/services/connection.rs

@@ -1,24 +1,24 @@
 use std::sync::{Arc, Mutex};
 
 use crate::error::*;
+use crate::http_client::HttpClient;
 use crate::storage::object_cache::ObjectCache;
 use crate::storage::Storage;
-use aries_vcx::common::ledger::transactions::into_did_doc;
 use aries_vcx::core::profile::profile::Profile;
-use aries_vcx::handlers::connection::connection::{Connection, ConnectionState};
 use aries_vcx::messages::a2a::A2AMessage;
 use aries_vcx::messages::concepts::ack::Ack;
 use aries_vcx::messages::protocols::connection::invite::Invitation;
 use aries_vcx::messages::protocols::connection::request::Request;
 use aries_vcx::messages::protocols::connection::response::SignedResponse;
 use aries_vcx::protocols::connection::pairwise_info::PairwiseInfo;
+use aries_vcx::protocols::connection::{Connection, GenericConnection, State, ThinState};
 
 pub type ServiceEndpoint = String;
 
 pub struct ServiceConnections {
     profile: Arc<dyn Profile>,
     service_endpoint: ServiceEndpoint,
-    connections: Arc<ObjectCache<Connection>>,
+    connections: Arc<ObjectCache<GenericConnection>>,
 }
 
 impl ServiceConnections {
@@ -31,93 +31,126 @@ impl ServiceConnections {
     }
 
     pub async fn create_invitation(&self, pw_info: Option<PairwiseInfo>) -> AgentResult<Invitation> {
-        let inviter = Connection::create_inviter(&self.profile, pw_info)
-            .await?
-            .create_invite(self.service_endpoint.clone(), vec![])
-            .await?;
-        let invite = inviter
-            .get_invite_details()
-            .ok_or_else(|| AgentError::from_kind(AgentErrorKind::InviteDetails))?
-            .clone();
-        self.connections.insert(&inviter.get_thread_id(), inviter)?;
+        let pw_info = pw_info.unwrap_or(PairwiseInfo::create(&self.profile.inject_wallet()).await?);
+        let inviter =
+            Connection::new_inviter("".to_owned(), pw_info).create_invitation(vec![], self.service_endpoint.clone());
+        let invite = inviter.get_invitation().clone();
+        let thread_id = inviter.thread_id().to_owned();
+
+        self.connections.insert(&thread_id, inviter.into())?;
+
         Ok(invite)
     }
 
     pub async fn receive_invitation(&self, invite: Invitation) -> AgentResult<String> {
-        let did_doc = into_did_doc(&self.profile, &invite).await?;
-        let invitee = Connection::create_invitee(&self.profile, did_doc)
-            .await?
-            .process_invite(invite)?;
-        self.connections.insert(&invitee.get_thread_id(), invitee)
+        let pairwise_info = PairwiseInfo::create(&self.profile.inject_wallet()).await?;
+        let invitee = Connection::new_invitee("".to_owned(), pairwise_info)
+            .accept_invitation(&self.profile, invite)
+            .await?;
+
+        let thread_id = invitee.thread_id().to_owned();
+
+        self.connections.insert(&thread_id, invitee.into())
     }
 
     pub async fn send_request(&self, thread_id: &str) -> AgentResult<()> {
-        let invitee = self
-            .connections
-            .get(thread_id)?
-            .send_request(&self.profile, self.service_endpoint.clone(), vec![], None)
+        let invitee: Connection<_, _> = self.connections.get(thread_id)?.try_into()?;
+        let invitee = invitee
+            .send_request(
+                &self.profile.inject_wallet(),
+                self.service_endpoint.clone(),
+                vec![],
+                &HttpClient,
+            )
             .await?;
-        self.connections.insert(thread_id, invitee)?;
+
+        self.connections.insert(thread_id, invitee.into())?;
         Ok(())
     }
 
     pub async fn accept_request(&self, thread_id: &str, request: Request) -> AgentResult<()> {
-        let inviter = self
-            .connections
-            .get(thread_id)?
-            .process_request(&self.profile, request, self.service_endpoint.clone(), vec![], None)
+        let inviter = self.connections.get(thread_id)?;
+
+        let inviter = match inviter.state() {
+            ThinState::Inviter(State::Initial) => Connection::try_from(inviter)
+                .map_err(From::from)
+                .map(|c| c.into_invited(&request.id.0)),
+            ThinState::Inviter(State::Invited) => Connection::try_from(inviter).map_err(From::from),
+            s => Err(AgentError::from_msg(
+                AgentErrorKind::GenericAriesVcxError,
+                &format!(
+                    "Connection with handle {} cannot process a request; State: {:?}",
+                    thread_id, s
+                ),
+            )),
+        }?;
+
+        let inviter = inviter
+            .handle_request(
+                &self.profile.inject_wallet(),
+                request,
+                self.service_endpoint.clone(),
+                vec![],
+                &HttpClient,
+            )
             .await?;
-        self.connections.insert(thread_id, inviter)?;
+
+        self.connections.insert(thread_id, inviter.into())?;
+
         Ok(())
     }
 
     pub async fn send_response(&self, thread_id: &str) -> AgentResult<()> {
-        let inviter = self
-            .connections
-            .get(thread_id)?
-            .send_response(&self.profile, None)
+        let inviter: Connection<_, _> = self.connections.get(thread_id)?.try_into()?;
+        let inviter = inviter
+            .send_response(&self.profile.inject_wallet(), &HttpClient)
             .await?;
-        self.connections.insert(thread_id, inviter)?;
+
+        self.connections.insert(thread_id, inviter.into())?;
+
         Ok(())
     }
 
     pub async fn accept_response(&self, thread_id: &str, response: SignedResponse) -> AgentResult<()> {
-        let invitee = self
-            .connections
-            .get(thread_id)?
-            .process_response(&self.profile, response, None)
+        let invitee: Connection<_, _> = self.connections.get(thread_id)?.try_into()?;
+        let invitee = invitee
+            .handle_response(&self.profile.inject_wallet(), response, &HttpClient)
             .await?;
-        self.connections.insert(thread_id, invitee)?;
+
+        self.connections.insert(thread_id, invitee.into())?;
+
         Ok(())
     }
 
     pub async fn send_ack(&self, thread_id: &str) -> AgentResult<()> {
-        let invitee = self.connections.get(thread_id)?.send_ack(&self.profile, None).await?;
-        self.connections.insert(thread_id, invitee)?;
+        let invitee: Connection<_, _> = self.connections.get(thread_id)?.try_into()?;
+        let invitee = invitee.send_ack(&self.profile.inject_wallet(), &HttpClient).await?;
+
+        self.connections.insert(thread_id, invitee.into())?;
+
         Ok(())
     }
 
     pub async fn process_ack(&self, thread_id: &str, ack: Ack) -> AgentResult<()> {
-        let inviter = self
-            .connections
-            .get(thread_id)?
-            .process_ack(A2AMessage::Ack(ack))
-            .await?;
-        self.connections.insert(thread_id, inviter)?;
+        let inviter: Connection<_, _> = self.connections.get(thread_id)?.try_into()?;
+        let inviter = inviter.acknowledge_connection(&A2AMessage::Ack(ack))?;
+
+        self.connections.insert(thread_id, inviter.into())?;
+
         Ok(())
     }
 
-    pub fn get_state(&self, thread_id: &str) -> AgentResult<ConnectionState> {
-        Ok(self.connections.get(thread_id)?.get_state())
+    pub fn get_state(&self, thread_id: &str) -> AgentResult<ThinState> {
+        Ok(self.connections.get(thread_id)?.state())
     }
 
-    pub(in crate::services) fn get_by_id(&self, thread_id: &str) -> AgentResult<Connection> {
+    pub(in crate::services) fn get_by_id(&self, thread_id: &str) -> AgentResult<GenericConnection> {
         self.connections.get(thread_id)
     }
 
     pub fn get_by_their_vk(&self, their_vk: &str) -> AgentResult<Vec<String>> {
         let their_vk = their_vk.to_string();
-        let f = |(id, m): (&String, &Mutex<Connection>)| -> Option<String> {
+        let f = |(id, m): (&String, &Mutex<GenericConnection>)| -> Option<String> {
             let connection = m.lock().unwrap();
             match connection.remote_vk() {
                 Ok(remote_vk) if remote_vk == their_vk => Some(id.to_string()),

agents/rust/aries-vcx-agent/src/services/holder.rs

@@ -1,15 +1,18 @@
 use std::sync::Arc;
 
 use crate::error::*;
+use crate::http_client::HttpClient;
 use crate::services::connection::ServiceConnections;
 use crate::storage::object_cache::ObjectCache;
 use crate::storage::Storage;
 use aries_vcx::core::profile::profile::Profile;
 use aries_vcx::handlers::issuance::holder::Holder;
+use aries_vcx::messages::a2a::A2AMessage;
 use aries_vcx::messages::protocols::issuance::credential::Credential;
 use aries_vcx::messages::protocols::issuance::credential_offer::CredentialOffer;
 use aries_vcx::messages::protocols::issuance::credential_proposal::CredentialProposalData;
 use aries_vcx::protocols::issuance::holder::state_machine::HolderState;
+use aries_vcx::protocols::SendClosure;
 
 #[derive(Clone)]
 struct HolderWrapper {
@@ -57,13 +60,15 @@ impl ServiceCredentialsHolder {
         proposal_data: CredentialProposalData,
     ) -> AgentResult<String> {
         let connection = self.service_connections.get_by_id(connection_id)?;
+        let wallet = self.profile.inject_wallet();
+
+        let send_closure: SendClosure = Box::new(|msg: A2AMessage| {
+            Box::pin(async move { connection.send_message(&wallet, &msg, &HttpClient).await })
+        });
+
         let mut holder = Holder::create("")?;
-        holder
-            .send_proposal(
-                proposal_data,
-                connection.send_message_closure(&self.profile, None).await?,
-            )
-            .await?;
+        holder.send_proposal(proposal_data, send_closure).await?;
+
         self.creds_holder
             .insert(&holder.get_thread_id()?, HolderWrapper::new(holder, connection_id))
     }
@@ -87,13 +92,14 @@ impl ServiceCredentialsHolder {
             (None, None) => return Err(AgentError::from_kind(AgentErrorKind::InvalidArguments)),
         };
         let connection = self.service_connections.get_by_id(&connection_id)?;
-        holder
-            .send_request(
-                &self.profile,
-                connection.pairwise_info().pw_did.to_string(),
-                connection.send_message_closure(&self.profile, None).await?,
-            )
-            .await?;
+        let wallet = self.profile.inject_wallet();
+        let pw_did = connection.pairwise_info().pw_did.to_string();
+
+        let send_closure: SendClosure = Box::new(|msg: A2AMessage| {
+            Box::pin(async move { connection.send_message(&wallet, &msg, &HttpClient).await })
+        });
+
+        holder.send_request(&self.profile, pw_did, send_closure).await?;
         self.creds_holder
             .insert(&holder.get_thread_id()?, HolderWrapper::new(holder, &connection_id))
     }
@@ -102,12 +108,14 @@ impl ServiceCredentialsHolder {
         let mut holder = self.get_holder(thread_id)?;
         let connection_id = self.get_connection_id(thread_id)?;
         let connection = self.service_connections.get_by_id(&connection_id)?;
+        let wallet = self.profile.inject_wallet();
+
+        let send_closure: SendClosure = Box::new(|msg: A2AMessage| {
+            Box::pin(async move { connection.send_message(&wallet, &msg, &HttpClient).await })
+        });
+
         holder
-            .process_credential(
-                &self.profile,
-                credential,
-                connection.send_message_closure(&self.profile, None).await?,
-            )
+            .process_credential(&self.profile, credential, send_closure)
             .await?;
         self.creds_holder
             .insert(&holder.get_thread_id()?, HolderWrapper::new(holder, &connection_id))