Merge pull request hyperledger-indy#832 from jovfer/feature/sp_plugged

Feature Plugged StateProof parsing

Author: jovfer

doc/design/008-state-proof-pluggable-parsing/README.md

@@ -25,7 +25,6 @@ extern fn CustomFree(data: *mut c_char) -> ErrorCode;
 Libindy API will contain call to register handler for specific transaction type:
 ```rust
 extern fn indy_register_transaction_parser_for_sp(command_handle: i32,
-                                                  pool_handle: i32,
                                                   txn_type: *const c_char,
                                                   parser: CustomTransactionParser,
                                                   free: CustomFree,

libindy/src/api/ledger.rs

@@ -1616,4 +1616,70 @@ pub extern fn indy_parse_get_revoc_reg_delta_response(command_handle: i32,
     trace!("indy_parse_get_revoc_reg_delta_response: <<< res: {:?}", res);
 
     res
-}
+}
+
+/// Callback type for parsing Reply from Node to specific StateProof format
+///
+/// # params
+/// reply_from_node: string representation of node's reply ("as is")
+/// parsed_sp: out param to return serialized as string JSON with array of ParsedSP
+///
+/// # return
+/// result ErrorCode
+///
+/// Note: this method allocate memory for result string `CustomFree` should be called to deallocate it
+pub type CustomTransactionParser = extern fn(reply_from_node: *const c_char, parsed_sp: *mut *const c_char) -> ErrorCode;
+
+/// Callback type to deallocate result buffer `parsed_sp` from `CustomTransactionParser`
+pub type CustomFree = extern fn(data: *const c_char) -> ErrorCode;
+
+
+/// Register callbacks (see type description for `CustomTransactionParser` and `CustomFree`
+///
+/// # params
+/// command_handle: command handle to map callback to caller context.
+/// txn_type: type of transaction to apply `parse` callback.
+/// parse: required callback to parse reply for state proof.
+/// free: required callback to deallocate memory.
+/// cb: Callback that takes command result as parameter.
+///
+/// # returns
+/// Status of callbacks registration.
+///
+/// # errors
+/// Common*
+#[no_mangle]
+pub extern fn indy_register_transaction_parser_for_sp(command_handle: i32,
+                                                      txn_type: *const c_char,
+                                                      parser: Option<CustomTransactionParser>,
+                                                      free: Option<CustomFree>,
+                                                      cb: Option<extern fn(command_handle_: i32, err: ErrorCode)>) -> ErrorCode {
+    trace!("indy_register_transaction_parser_for_sp: >>> txn_type {:?}, parser {:?}, free {:?}",
+           txn_type, parser, free);
+
+    check_useful_c_str!(txn_type, ErrorCode::CommonInvalidParam2);
+    check_useful_c_callback!(parser, ErrorCode::CommonInvalidParam3);
+    check_useful_c_callback!(free, ErrorCode::CommonInvalidParam4);
+    check_useful_c_callback!(cb, ErrorCode::CommonInvalidParam5);
+
+    trace!("indy_register_transaction_parser_for_sp: entities: txn_type {}, parser {:?}, free {:?}",
+           txn_type, parser, free);
+
+    let res = CommandExecutor::instance()
+        .send(Command::Ledger(LedgerCommand::RegisterSPParser(
+            txn_type,
+            parser,
+            free,
+            Box::new(move |res| {
+                let res = result_to_err_code!(res);
+                trace!("indy_register_transaction_parser_for_sp: res: {:?}", res);
+                cb(command_handle, res)
+            }),
+        )));
+
+    let res = result_to_err_code!(res);
+
+    trace!("indy_register_transaction_parser_for_sp: <<< res: {:?}", res);
+
+    res
+}

libindy/src/commands/ledger.rs

@@ -3,6 +3,8 @@ extern crate indy_crypto;
 
 use self::serde_json::Value;
 
+use api::ledger::{CustomFree, CustomTransactionParser};
+
 use errors::common::CommonError;
 use errors::pool::PoolError;
 use errors::crypto::CryptoError;
@@ -170,7 +172,12 @@ pub enum LedgerCommand {
         Box<Fn(Result<String, IndyError>) + Send>),
     ParseGetRevocRegDeltaResponse(
         String, // get revocation registry delta response
-        Box<Fn(Result<(String, String, u64), IndyError>) + Send>)
+        Box<Fn(Result<(String, String, u64), IndyError>) + Send>),
+    RegisterSPParser(
+        String, // txn type
+        CustomTransactionParser,
+        CustomFree,
+        Box<Fn(Result<(), IndyError>) + Send>),
 }
 
 pub struct LedgerCommandExecutor {
@@ -212,6 +219,10 @@ impl LedgerCommandExecutor {
                     .expect("Expect callback to process ack command")
                     (result.map_err(IndyError::from));
             }
+            LedgerCommand::RegisterSPParser(txn_type, parser, free, cb) => {
+                info!(target: "ledger_command_executor", "RegisterSPParser command received");
+                cb(self.register_sp_parser(&txn_type, parser, free));
+            }
             LedgerCommand::SignRequest(wallet_handle, submitter_did, request_json, cb) => {
                 info!(target: "ledger_command_executor", "SignRequest command received");
                 cb(self.sign_request(wallet_handle, &submitter_did, &request_json));
@@ -335,6 +346,15 @@ impl LedgerCommandExecutor {
         };
     }
 
+    fn register_sp_parser(&self, txn_type: &str,
+                          parser: CustomTransactionParser, free: CustomFree) -> Result<(), IndyError> {
+        debug!("register_sp_parser >>> txn_type: {:?}, parser: {:?}, free: {:?}",
+               txn_type, parser, free);
+
+        PoolService::register_sp_parser(txn_type, parser, free)
+            .map_err(IndyError::from)
+    }
+
     fn sign_and_submit_request(&self,
                                pool_handle: i32,
                                wallet_handle: i32,

libindy/src/services/pool/mod.rs

@@ -30,6 +30,7 @@ use std::fmt::Debug;
 use std::io::{Read, BufRead, Write};
 use std::ops::{Add, Sub};
 
+use api::ledger::{CustomFree, CustomTransactionParser};
 use commands::{Command, CommandExecutor};
 use commands::ledger::LedgerCommand;
 use commands::pool::PoolCommand;
@@ -44,6 +45,7 @@ use utils::environment::EnvironmentUtils;
 use utils::sequence::SequenceUtils;
 use self::indy_crypto::bls::VerKey;
 use std::path::PathBuf;
+use std::sync::Mutex;
 
 use self::indy_crypto::utils::json::{JsonDecodable, JsonEncodable};
 
@@ -52,6 +54,10 @@ pub struct PoolService {
     open_pools: RefCell<HashMap<i32, Pool>>,
 }
 
+lazy_static! {
+    static ref REGISTERED_SP_PARSERS: Mutex<HashMap<String, (CustomTransactionParser, CustomFree)>> = Mutex::new(HashMap::new());
+}
+
 struct Pool {
     name: String,
     id: i32,
@@ -251,26 +257,26 @@ impl PoolWorker {
     }
 
     fn process_actions(&mut self, actions: Vec<ZMQLoopAction>) -> Result<(), PoolError> {
-        for action in &actions {
-            match action {
-                &ZMQLoopAction::Terminate(cmd_id) => {
+        for action in actions {
+            match action  {
+                ZMQLoopAction::Terminate(cmd_id) => {
                     let res = self.handler.flush_requests(Err(PoolError::Terminate));
                     if cmd_id >= 0 {
                         CommandExecutor::instance().send(Command::Pool(PoolCommand::CloseAck(cmd_id, res)))?;
                     }
                     return Err(PoolError::Terminate);
                 }
-                &ZMQLoopAction::Refresh(cmd_id) => {
+                ZMQLoopAction::Refresh(cmd_id) => {
                     self.refresh(cmd_id)?;
                 }
-                &ZMQLoopAction::MessageToProcess(ref msg) => {
+                ZMQLoopAction::MessageToProcess(ref msg) => {
                     if let Some(new_mt) = self.handler.process_msg(&msg.message, msg.node_idx)? {
                         self.handler.flush_requests(Ok(()))?;
                         self.handler = PoolWorkerHandler::TransactionHandler(Default::default());
                         self.connect_to_known_nodes(Some(&new_mt))?;
                     }
                 }
-                &ZMQLoopAction::RequestToSend(ref req) => {
+                ZMQLoopAction::RequestToSend(ref req) => {
                     self.handler.send_request(req.request.as_str(), req.id).or_else(|err| {
                         CommandExecutor::instance()
                             .send(Command::Ledger(LedgerCommand::SubmitAck(req.id, Err(err))))
@@ -279,7 +285,7 @@ impl PoolWorker {
                             })
                     })?;
                 }
-                &ZMQLoopAction::Timeout => {
+                ZMQLoopAction::Timeout => {
                     self.handler.process_timeout()?;
                 }
             }
@@ -783,6 +789,26 @@ impl PoolService {
         Ok(cmd_id)
     }
 
+    pub fn register_sp_parser(txn_type: &str,
+                              parser: CustomTransactionParser, free: CustomFree) -> Result<(), PoolError> {
+        if transaction_handler::REQUESTS_FOR_STATE_PROOFS.contains(&txn_type) {
+            return Err(PoolError::CommonError(CommonError::InvalidStructure(
+                format!("Try to override StateProof parser for default TXN_TYPE {}", txn_type))));
+        }
+        REGISTERED_SP_PARSERS.lock()
+            .map(|mut map| {
+                map.insert(txn_type.to_owned(), (parser, free));
+            })
+            .map_err(|_| PoolError::CommonError(CommonError::InvalidState(
+                "Can't register new SP parser: mutex lock error".to_owned())))
+    }
+
+    pub fn get_sp_parser(txn_type: &str) -> Option<(CustomTransactionParser, CustomFree)> {
+        REGISTERED_SP_PARSERS.lock().ok().and_then(|map| {
+            map.get(txn_type).map(Clone::clone)
+        })
+    }
+
     pub fn close(&self, handle: i32) -> Result<i32, PoolError> {
         let cmd_id: i32 = SequenceUtils::get_next_id();
         self.open_pools.try_borrow_mut().map_err(CommonError::from)?
@@ -988,7 +1014,10 @@ mod tests {
                 cmd_sock: zmq::Context::new().socket(zmq::SocketType::PAIR).unwrap(),
                 open_cmd_id: 0,
                 name: "".to_string(),
-                handler: PoolWorkerHandler::CatchupHandler(Default::default()),
+                handler: PoolWorkerHandler::CatchupHandler(CatchupHandler {
+                    timeout: time::now_utc().add(Duration::seconds(2)),
+                    ..Default::default()
+                }),
             }
         }
     }