Remove msg-sending IO for issuer and holder

agents/node/vcxagent-core/demo/alice.js

@@ -41,7 +41,7 @@ async function getInvitationString (fetchInviteUrl) {
 }
 
 async function runAlice (options) {
-  logger.info('Starting.')
+  logger.info('Starting alice.')
 
   initRustLogger(process.env.RUST_LOG || 'vcx=error')
   const agentName = `alice-${uuid.v4()}`

agents/node/vcxagent-core/demo/faber.js

@@ -17,7 +17,7 @@ const { testTailsUrl, initRustLogger } = require('../src')
 const tailsDir = '/tmp/tails'
 
 async function runFaber (options) {
-  logger.info(`Starting. Revocation enabled=${options.revocation}`)
+  logger.info(`Starting Faber. Revocation enabled=${options.revocation}`)
   initRustLogger(process.env.RUST_LOG || 'vcx=error')
 
   let faberServer
@@ -84,7 +84,9 @@ async function runFaber (options) {
     const schemaAttrs = getAliceSchemaAttrs()
     await vcxAgent.serviceCredIssuer.sendOfferAndCredential(issuerCredId, revRegId, connectionId, credDefId, schemaAttrs)
     if (options.revocation === true) {
+      logger.info('Faber is revoking issued credential')
       await vcxAgent.serviceCredIssuer.revokeCredentialLocal(issuerCredId)
+      logger.info('Faber is publishing revocation')
       await revReg.publishRevocations()
     }
 

agents/node/vcxagent-core/demo/logger.js

@@ -5,7 +5,7 @@ const prettyFormatter = format.combine(
   format.printf(
     msg => {
       const extras = (global.expect) ? `${global.expect.getState().currentTestName}` : ''
-      return `[${msg.timestamp}] [${msg.filename}] [${msg.level}] ${extras}: ${msg.message}`
+      return `[${msg.timestamp}] [${msg.filename}] [${msg.label}] [${msg.level}] ${extras}: ${msg.message}`
     }
   )
 )
@@ -14,6 +14,7 @@ module.exports = loggerLabel => {
   return createLogger({
     level: 'debug',
     format: format.combine(
+      format.label({ label: loggerLabel }),
       format.timestamp({
         format: 'YYYY-MM-DD HH:mm:ss.SSS'
       }),

agents/node/vcxagent-core/src/services/service-cred-issuer.js

@@ -19,10 +19,7 @@ module.exports.createServiceCredIssuer = function createServiceCredIssuer ({ log
     })
     const state1 = await issuerCred.getState()
     assert.equal(state1, IssuerStateType.OfferSet)
-    const credOfferMsg = await issuerCred.getCredentialOfferMsg()
-    await issuerCred.markCredentialOfferMsgSent()
-    const state2 = await issuerCred.getState()
-    assert.equal(state2, IssuerStateType.OfferSent)
+    const credOfferMsg = issuerCred.getCredentialOfferMsg()
     await saveIssuerCredential(issuerCredId, issuerCred)
 
     return credOfferMsg
@@ -44,8 +41,6 @@ module.exports.createServiceCredIssuer = function createServiceCredIssuer ({ log
     const state1 = await issuerCred.getState()
     assert.equal(state1, IssuerStateType.OfferSet)
     await issuerCred.sendOfferV2(connection)
-    const state2 = await issuerCred.getState()
-    assert.equal(state2, IssuerStateType.OfferSent)
     await saveIssuerCredential(issuerCredId, issuerCred)
   }
 
@@ -108,9 +103,11 @@ module.exports.createServiceCredIssuer = function createServiceCredIssuer ({ log
 
   async function _progressIssuerCredentialToState (issuerCredential, connection, credentialStateTarget, attemptsThreshold, timeoutMs) {
     async function progressToAcceptedState () {
-      if (await issuerCredential.updateStateV2(connection) !== credentialStateTarget) {
+      const currentState = await issuerCredential.updateStateV2(connection)
+      if (currentState !== credentialStateTarget) {
         return { result: undefined, isFinished: false }
       } else {
+        logger.debug(`Cred issuer: _progressIssuerCredentialToState, currentState: ${currentState}, credentialStateTarget: ${credentialStateTarget}`)
         return { result: null, isFinished: true }
       }
     }

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

@@ -57,17 +57,16 @@ impl ServiceCredentialsHolder {
     pub async fn send_credential_proposal(
         &self,
         connection_id: &str,
-        proposal_data: ProposeCredential,
+        propose_credential: ProposeCredential,
     ) -> 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: AriesMessage| {
-            Box::pin(async move { connection.send_message(&wallet, &msg, &HttpClient).await })
-        });
-
         let mut holder = Holder::create("")?;
-        holder.send_proposal(proposal_data, send_closure).await?;
+        holder.set_proposal(propose_credential.clone())?;
+        connection
+            .send_message(&wallet, &propose_credential.into(), &HttpClient)
+            .await?;
 
         self.creds_holder
             .insert(&holder.get_thread_id()?, HolderWrapper::new(holder, connection_id))
@@ -98,37 +97,44 @@ impl ServiceCredentialsHolder {
         let send_closure: SendClosure = Box::new(|msg: AriesMessage| {
             Box::pin(async move { connection.send_message(&wallet, &msg, &HttpClient).await })
         });
-
-        holder
-            .send_request(
+        let msg_response = holder
+            .prepare_credential_request(
                 &self.profile.inject_anoncreds_ledger_read(),
                 &self.profile.inject_anoncreds(),
                 pw_did,
-                send_closure,
             )
             .await?;
+        send_closure(msg_response).await?;
         self.creds_holder
             .insert(&holder.get_thread_id()?, HolderWrapper::new(holder, &connection_id))
     }
 
-    pub async fn process_credential(&self, thread_id: &str, credential: IssueCredential) -> AgentResult<String> {
+    pub async fn process_credential(
+        &self,
+        thread_id: &str,
+        msg_issue_credential: IssueCredential,
+    ) -> AgentResult<String> {
         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: AriesMessage| {
-            Box::pin(async move { connection.send_message(&wallet, &msg, &HttpClient).await })
-        });
-
         holder
             .process_credential(
                 &self.profile.inject_anoncreds_ledger_read(),
                 &self.profile.inject_anoncreds(),
-                credential,
-                send_closure,
+                msg_issue_credential.clone(),
             )
             .await?;
+        match holder.get_final_message()? {
+            None => {}
+            Some(msg_response) => {
+                let send_closure: SendClosure = Box::new(|msg: AriesMessage| {
+                    Box::pin(async move { connection.send_message(&wallet, &msg, &HttpClient).await })
+                });
+                send_closure(msg_response).await?;
+            }
+        }
         self.creds_holder
             .insert(&holder.get_thread_id()?, HolderWrapper::new(holder, &connection_id))
     }

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

@@ -84,7 +84,8 @@ impl ServiceCredentialsIssuer {
             Box::pin(async move { connection.send_message(&wallet, &msg, &HttpClient).await })
         });
 
-        issuer.send_credential_offer(send_closure).await?;
+        let credential_offer = issuer.get_credential_offer_msg()?;
+        send_closure(credential_offer).await?;
         self.creds_issuer
             .insert(&issuer.get_thread_id()?, IssuerWrapper::new(issuer, &connection_id))
     }
@@ -124,9 +125,17 @@ impl ServiceCredentialsIssuer {
             Box::pin(async move { connection.send_message(&wallet, &msg, &HttpClient).await })
         });
 
-        issuer
-            .send_credential(&self.profile.inject_anoncreds(), send_closure)
-            .await?;
+        issuer.build_credential(&self.profile.inject_anoncreds()).await?;
+        match issuer.get_state() {
+            IssuerState::Failed => {
+                let problem_report = issuer.get_problem_report()?;
+                send_closure(problem_report.into()).await?;
+            }
+            _ => {
+                let msg_issue_credential = issuer.get_msg_issue_credential()?;
+                send_closure(msg_issue_credential.into()).await?;
+            }
+        }
         self.creds_issuer
             .insert(&issuer.get_thread_id()?, IssuerWrapper::new(issuer, &connection_id))?;
         Ok(())

aries_vcx/src/handlers/issuance/holder.rs

@@ -1,21 +1,38 @@
+use chrono::Utc;
 use std::sync::Arc;
+use uuid::Uuid;
 
 use aries_vcx_core::anoncreds::base_anoncreds::BaseAnonCreds;
 use aries_vcx_core::ledger::base_ledger::AnoncredsLedgerRead;
 use aries_vcx_core::wallet::base_wallet::BaseWallet;
+use messages::decorators::thread::Thread;
+use messages::decorators::timing::Timing;
+use messages::msg_fields::protocols::cred_issuance::ack::{AckCredential, AckCredentialContent};
 use messages::msg_fields::protocols::cred_issuance::issue_credential::IssueCredential;
 use messages::msg_fields::protocols::cred_issuance::offer_credential::OfferCredential;
 use messages::msg_fields::protocols::cred_issuance::propose_credential::ProposeCredential;
+use messages::msg_fields::protocols::cred_issuance::request_credential::RequestCredential;
 use messages::msg_fields::protocols::cred_issuance::CredentialIssuance;
+use messages::msg_fields::protocols::notification::ack::{AckDecorators, AckStatus};
+use messages::msg_fields::protocols::report_problem::ProblemReport;
 use messages::msg_fields::protocols::revocation::revoke::Revoke;
 use messages::AriesMessage;
 
 use crate::common::credentials::get_cred_rev_id;
 use crate::errors::error::prelude::*;
 use crate::handlers::connection::mediated_connection::MediatedConnection;
 use crate::handlers::revocation_notification::receiver::RevocationNotificationReceiver;
-use crate::protocols::issuance::holder::state_machine::{HolderSM, HolderState};
-use crate::protocols::SendClosure;
+use crate::protocols::issuance::holder::state_machine::{HolderFullState, HolderSM, HolderState};
+
+fn build_credential_ack(thread_id: &str) -> AckCredential {
+    let content = AckCredentialContent::new(AckStatus::Ok);
+    let mut decorators = AckDecorators::new(Thread::new(thread_id.to_owned()));
+    let mut timing = Timing::default();
+    timing.out_time = Some(Utc::now());
+    decorators.timing = Some(timing);
+
+    AckCredential::with_decorators(Uuid::new_v4().to_string(), content, decorators)
+}
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct Holder {
@@ -29,6 +46,16 @@ impl Holder {
         Ok(Holder { holder_sm })
     }
 
+    pub fn create_with_proposal(source_id: &str, propose_credential: ProposeCredential) -> VcxResult<Holder> {
+        trace!(
+            "Holder::create_with_proposal >>> source_id: {:?}, propose_credential: {:?}",
+            source_id,
+            propose_credential
+        );
+        let holder_sm = HolderSM::with_proposal(propose_credential, source_id.to_string());
+        Ok(Holder { holder_sm })
+    }
+
     pub fn create_from_offer(source_id: &str, credential_offer: OfferCredential) -> VcxResult<Holder> {
         trace!(
             "Holder::create_from_offer >>> source_id: {:?}, credential_offer: {:?}",
@@ -39,54 +66,63 @@ impl Holder {
         Ok(Holder { holder_sm })
     }
 
-    pub async fn send_proposal(
-        &mut self,
-        credential_proposal: ProposeCredential,
-        send_message: SendClosure,
-    ) -> VcxResult<()> {
-        self.holder_sm = self
-            .holder_sm
-            .clone()
-            .send_proposal(credential_proposal, send_message)
-            .await?;
+    pub fn set_proposal(&mut self, credential_proposal: ProposeCredential) -> VcxResult<()> {
+        self.holder_sm = self.holder_sm.clone().set_proposal(credential_proposal)?;
         Ok(())
     }
 
-    pub async fn send_request(
+    pub async fn prepare_credential_request(
         &mut self,
         ledger: &Arc<dyn AnoncredsLedgerRead>,
         anoncreds: &Arc<dyn BaseAnonCreds>,
         my_pw_did: String,
-        send_message: SendClosure,
-    ) -> VcxResult<()> {
+    ) -> VcxResult<AriesMessage> {
         self.holder_sm = self
             .holder_sm
             .clone()
-            .send_request(ledger, anoncreds, my_pw_did, send_message)
+            .prepare_credential_request(ledger, anoncreds, my_pw_did)
             .await?;
-        Ok(())
+        match self.get_state() {
+            HolderState::Failed => {
+                Ok(self.get_problem_report()?.into())
+            }
+            HolderState::RequestSet => {
+                Ok(self.get_msg_credential_request()?.into())
+            }
+            _ => {
+                Err(AriesVcxError::from_msg(AriesVcxErrorKind::InvalidState, "Holder::prepare_credential_request >> reached unexpected state after calling prepare_credential_request"))
+            }
+        }
     }
 
-    pub async fn decline_offer<'a>(&'a mut self, comment: Option<&'a str>, send_message: SendClosure) -> VcxResult<()> {
-        self.holder_sm = self
-            .holder_sm
-            .clone()
-            .decline_offer(comment.map(String::from), send_message)
-            .await?;
-        Ok(())
+    pub fn get_msg_credential_request(&self) -> VcxResult<RequestCredential> {
+        match self.holder_sm.state {
+            HolderFullState::RequestSet(ref state) => {
+                let mut msg: RequestCredential = state.msg_credential_request.clone().into();
+                let mut timing = Timing::default();
+                timing.out_time = Some(Utc::now());
+                msg.decorators.timing = Some(timing);
+                Ok(msg)
+            }
+            _ => Err(AriesVcxError::from_msg(AriesVcxErrorKind::NotReady, "Invalid action")),
+        }
+    }
+
+    pub fn decline_offer<'a>(&'a mut self, comment: Option<&'a str>) -> VcxResult<ProblemReport> {
+        self.holder_sm = self.holder_sm.clone().decline_offer(comment.map(String::from))?;
+        self.get_problem_report()
     }
 
     pub async fn process_credential(
         &mut self,
         ledger: &Arc<dyn AnoncredsLedgerRead>,
         anoncreds: &Arc<dyn BaseAnonCreds>,
         credential: IssueCredential,
-        send_message: SendClosure,
     ) -> VcxResult<()> {
         self.holder_sm = self
             .holder_sm
             .clone()
-            .receive_credential(ledger, anoncreds, credential, send_message)
+            .receive_credential(ledger, anoncreds, credential)
             .await?;
         Ok(())
     }
@@ -186,25 +222,25 @@ impl Holder {
         }
     }
 
+    pub fn get_problem_report(&self) -> VcxResult<ProblemReport> {
+        self.holder_sm.get_problem_report()
+    }
+
+    // todo 0109: send ack/problem-report in upper layer
     pub async fn process_aries_msg(
         &mut self,
         ledger: &Arc<dyn AnoncredsLedgerRead>,
         anoncreds: &Arc<dyn BaseAnonCreds>,
         message: AriesMessage,
-        send_message: Option<SendClosure>,
     ) -> VcxResult<()> {
         let holder_sm = match message {
             AriesMessage::CredentialIssuance(CredentialIssuance::OfferCredential(offer)) => {
                 self.holder_sm.clone().receive_offer(offer)?
             }
             AriesMessage::CredentialIssuance(CredentialIssuance::IssueCredential(credential)) => {
-                let send_message = send_message.ok_or(AriesVcxError::from_msg(
-                    AriesVcxErrorKind::InvalidState,
-                    "Attempted to call undefined send_message callback",
-                ))?;
                 self.holder_sm
                     .clone()
-                    .receive_credential(ledger, anoncreds, credential, send_message)
+                    .receive_credential(ledger, anoncreds, credential)
                     .await?
             }
             AriesMessage::ReportProblem(report) => self.holder_sm.clone().receive_problem_report(report)?,
@@ -213,4 +249,19 @@ impl Holder {
         self.holder_sm = holder_sm;
         Ok(())
     }
+
+    pub fn get_final_message(&self) -> VcxResult<Option<AriesMessage>> {
+        match &self.holder_sm.state {
+            HolderFullState::Finished(state) => {
+                if let Some(ack_requested) = state.ack_requested {
+                    if ack_requested {
+                        let ack_msg = build_credential_ack(&self.get_thread_id()?);
+                        return Ok(Some(ack_msg.into()));
+                    }
+                }
+            }
+            _ => {}
+        };
+        return Ok(None);
+    }
 }