use crate::api;
use crate::chain::types::NoopAdapter;
use crate::chain::Chain;
use crate::config::WalletConfig;
use crate::core::core::verifier_cache::LruVerifierCache;
use crate::core::core::Transaction;
use crate::core::global::{set_mining_mode, ChainTypes};
use crate::core::{pow, ser};
use crate::keychain::Keychain;
use crate::libwallet::api_impl::foreign;
use crate::libwallet::{NodeClient, NodeVersionInfo, Slate, TxWrapper, WalletInst};
use crate::util;
use crate::util::secp::pedersen;
use crate::util::secp::pedersen::Commitment;
use crate::util::{Mutex, RwLock};
use crate::{libwallet, WalletCommAdapter};
use failure::ResultExt;
use serde_json;
use std::collections::HashMap;
use std::marker::PhantomData;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{channel, Receiver, Sender};
use std::sync::Arc;
use std::thread;
use std::time::Duration;
#[derive(Clone, Debug)]
pub struct WalletProxyMessage {
pub sender_id: String,
pub dest: String,
pub method: String,
pub body: String,
}
pub struct WalletProxy<C, K>
where
C: NodeClient,
K: Keychain,
{
pub chain_dir: String,
pub chain: Arc<Chain>,
pub wallets: HashMap<
String,
(
Sender<WalletProxyMessage>,
Arc<Mutex<dyn WalletInst<LocalWalletClient, K>>>,
),
>,
pub tx: Sender<WalletProxyMessage>,
pub rx: Receiver<WalletProxyMessage>,
pub running: Arc<AtomicBool>,
phantom_c: PhantomData<C>,
phantom_k: PhantomData<K>,
}
impl<C, K> WalletProxy<C, K>
where
C: NodeClient,
K: Keychain,
{
pub fn new(chain_dir: &str) -> Self {
set_mining_mode(ChainTypes::AutomatedTesting);
let genesis_block = pow::mine_genesis_block().unwrap();
let verifier_cache = Arc::new(RwLock::new(LruVerifierCache::new()));
let dir_name = format!("{}/.grin", chain_dir);
let c = Chain::init(
dir_name.to_string(),
Arc::new(NoopAdapter {}),
genesis_block,
pow::verify_size,
verifier_cache,
false,
)
.unwrap();
let (tx, rx) = channel();
let retval = WalletProxy {
chain_dir: chain_dir.to_owned(),
chain: Arc::new(c),
tx: tx,
rx: rx,
wallets: HashMap::new(),
running: Arc::new(AtomicBool::new(false)),
phantom_c: PhantomData,
phantom_k: PhantomData,
};
retval
}
pub fn add_wallet(
&mut self,
addr: &str,
tx: Sender<WalletProxyMessage>,
wallet: Arc<Mutex<dyn WalletInst<LocalWalletClient, K>>>,
) {
self.wallets.insert(addr.to_owned(), (tx, wallet));
}
pub fn stop(&mut self) {
self.running.store(false, Ordering::Relaxed);
}
pub fn run(&mut self) -> Result<(), libwallet::Error> {
self.running.store(true, Ordering::Relaxed);
loop {
thread::sleep(Duration::from_millis(10));
let m = self.rx.recv().unwrap();
trace!("Wallet Client Proxy Received: {:?}", m);
let resp = match m.method.as_ref() {
"get_chain_height" => self.get_chain_height(m)?,
"get_outputs_from_node" => self.get_outputs_from_node(m)?,
"get_outputs_by_pmmr_index" => self.get_outputs_by_pmmr_index(m)?,
"send_tx_slate" => self.send_tx_slate(m)?,
"post_tx" => self.post_tx(m)?,
_ => panic!("Unknown Wallet Proxy Message"),
};
self.respond(resp);
if !self.running.load(Ordering::Relaxed) {
return Ok(());
}
}
}
fn respond(&mut self, m: WalletProxyMessage) {
if let Some(s) = self.wallets.get_mut(&m.dest) {
if let Err(e) = s.0.send(m.clone()) {
panic!("Error sending response from proxy: {:?}, {}", m, e);
}
} else {
panic!("Unknown wallet recipient for response message: {:?}", m);
}
}
fn post_tx(&mut self, m: WalletProxyMessage) -> Result<WalletProxyMessage, libwallet::Error> {
let dest_wallet = self.wallets.get_mut(&m.sender_id).unwrap().1.clone();
let wrapper: TxWrapper = serde_json::from_str(&m.body).context(
libwallet::ErrorKind::ClientCallback("Error parsing TxWrapper".to_owned()),
)?;
let tx_bin = util::from_hex(wrapper.tx_hex).context(
libwallet::ErrorKind::ClientCallback("Error parsing TxWrapper: tx_bin".to_owned()),
)?;
let tx: Transaction = ser::deserialize(&mut &tx_bin[..]).context(
libwallet::ErrorKind::ClientCallback("Error parsing TxWrapper: tx".to_owned()),
)?;
super::award_block_to_wallet(&self.chain, vec![&tx], dest_wallet)?;
Ok(WalletProxyMessage {
sender_id: "node".to_owned(),
dest: m.sender_id,
method: m.method,
body: "".to_owned(),
})
}
fn send_tx_slate(
&mut self,
m: WalletProxyMessage,
) -> Result<WalletProxyMessage, libwallet::Error> {
let dest_wallet = self.wallets.get_mut(&m.dest);
let wallet = match dest_wallet {
None => panic!("Unknown wallet destination for send_tx_slate: {:?}", m),
Some(w) => w,
};
let mut slate = serde_json::from_str(&m.body).context(
libwallet::ErrorKind::ClientCallback("Error parsing TxWrapper".to_owned()),
)?;
;
{
let mut w = wallet.1.lock();
w.open_with_credentials()?;
slate = foreign::receive_tx(&mut *w, &slate, None, None, false)?;
w.close()?;
}
Ok(WalletProxyMessage {
sender_id: m.dest,
dest: m.sender_id,
method: m.method,
body: serde_json::to_string(&slate).unwrap(),
})
}
fn get_chain_height(
&mut self,
m: WalletProxyMessage,
) -> Result<WalletProxyMessage, libwallet::Error> {
Ok(WalletProxyMessage {
sender_id: "node".to_owned(),
dest: m.sender_id,
method: m.method,
body: format!("{}", self.chain.head().unwrap().height).to_owned(),
})
}
fn get_outputs_from_node(
&mut self,
m: WalletProxyMessage,
) -> Result<WalletProxyMessage, libwallet::Error> {
let split = m.body.split(",");
let mut outputs: Vec<api::Output> = vec![];
for o in split {
let o_str = String::from(o);
if o_str.len() == 0 {
continue;
}
let c = util::from_hex(o_str).unwrap();
let commit = Commitment::from_vec(c);
let out = super::get_output_local(&self.chain.clone(), &commit);
if let Some(o) = out {
outputs.push(o);
}
}
Ok(WalletProxyMessage {
sender_id: "node".to_owned(),
dest: m.sender_id,
method: m.method,
body: serde_json::to_string(&outputs).unwrap(),
})
}
fn get_outputs_by_pmmr_index(
&mut self,
m: WalletProxyMessage,
) -> Result<WalletProxyMessage, libwallet::Error> {
let split = m.body.split(",").collect::<Vec<&str>>();
let start_index = split[0].parse::<u64>().unwrap();
let max = split[1].parse::<u64>().unwrap();
let ol = super::get_outputs_by_pmmr_index_local(self.chain.clone(), start_index, max);
Ok(WalletProxyMessage {
sender_id: "node".to_owned(),
dest: m.sender_id,
method: m.method,
body: serde_json::to_string(&ol).unwrap(),
})
}
}
#[derive(Clone)]
pub struct LocalWalletClient {
pub id: String,
pub proxy_tx: Arc<Mutex<Sender<WalletProxyMessage>>>,
pub rx: Arc<Mutex<Receiver<WalletProxyMessage>>>,
pub tx: Arc<Mutex<Sender<WalletProxyMessage>>>,
}
impl LocalWalletClient {
pub fn new(id: &str, proxy_rx: Sender<WalletProxyMessage>) -> Self {
let (tx, rx) = channel();
LocalWalletClient {
id: id.to_owned(),
proxy_tx: Arc::new(Mutex::new(proxy_rx)),
rx: Arc::new(Mutex::new(rx)),
tx: Arc::new(Mutex::new(tx)),
}
}
pub fn get_send_instance(&self) -> Sender<WalletProxyMessage> {
self.tx.lock().clone()
}
pub fn send_tx_slate_direct(
&self,
dest: &str,
slate: &Slate,
) -> Result<Slate, libwallet::Error> {
let m = WalletProxyMessage {
sender_id: self.id.clone(),
dest: dest.to_owned(),
method: "send_tx_slate".to_owned(),
body: serde_json::to_string(slate).unwrap(),
};
{
let p = self.proxy_tx.lock();
p.send(m).context(libwallet::ErrorKind::ClientCallback(
"Send TX Slate".to_owned(),
))?;
}
let r = self.rx.lock();
let m = r.recv().unwrap();
trace!("Received send_tx_slate response: {:?}", m.clone());
Ok(
serde_json::from_str(&m.body).context(libwallet::ErrorKind::ClientCallback(
"Parsing send_tx_slate response".to_owned(),
))?,
)
}
}
impl WalletCommAdapter for LocalWalletClient {
fn supports_sync(&self) -> bool {
true
}
fn send_tx_sync(&self, dest: &str, slate: &Slate) -> Result<Slate, libwallet::Error> {
let m = WalletProxyMessage {
sender_id: self.id.clone(),
dest: dest.to_owned(),
method: "send_tx_slate".to_owned(),
body: serde_json::to_string(slate).unwrap(),
};
{
let p = self.proxy_tx.lock();
p.send(m).context(libwallet::ErrorKind::ClientCallback(
"Send TX Slate".to_owned(),
))?;
}
let r = self.rx.lock();
let m = r.recv().unwrap();
trace!("Received send_tx_slate response: {:?}", m.clone());
Ok(
serde_json::from_str(&m.body).context(libwallet::ErrorKind::ClientCallback(
"Parsing send_tx_slate response".to_owned(),
))?,
)
}
fn send_tx_async(&self, _dest: &str, _slate: &Slate) -> Result<(), libwallet::Error> {
unimplemented!();
}
fn receive_tx_async(&self, _params: &str) -> Result<Slate, libwallet::Error> {
unimplemented!();
}
fn listen(
&self,
_params: HashMap<String, String>,
_config: WalletConfig,
_passphrase: &str,
_account: &str,
_node_api_secret: Option<String>,
) -> Result<(), libwallet::Error> {
unimplemented!();
}
}
impl NodeClient for LocalWalletClient {
fn node_url(&self) -> &str {
"node"
}
fn node_api_secret(&self) -> Option<String> {
None
}
fn set_node_url(&mut self, _node_url: &str) {}
fn set_node_api_secret(&mut self, _node_api_secret: Option<String>) {}
fn get_version_info(&mut self) -> Option<NodeVersionInfo> {
None
}
fn post_tx(&self, tx: &TxWrapper, _fluff: bool) -> Result<(), libwallet::Error> {
let m = WalletProxyMessage {
sender_id: self.id.clone(),
dest: self.node_url().to_owned(),
method: "post_tx".to_owned(),
body: serde_json::to_string(tx).unwrap(),
};
{
let p = self.proxy_tx.lock();
p.send(m).context(libwallet::ErrorKind::ClientCallback(
"post_tx send".to_owned(),
))?;
}
let r = self.rx.lock();
let m = r.recv().unwrap();
trace!("Received post_tx response: {:?}", m.clone());
Ok(())
}
fn get_chain_height(&self) -> Result<u64, libwallet::Error> {
let m = WalletProxyMessage {
sender_id: self.id.clone(),
dest: self.node_url().to_owned(),
method: "get_chain_height".to_owned(),
body: "".to_owned(),
};
{
let p = self.proxy_tx.lock();
p.send(m).context(libwallet::ErrorKind::ClientCallback(
"Get chain height send".to_owned(),
))?;
}
let r = self.rx.lock();
let m = r.recv().unwrap();
trace!("Received get_chain_height response: {:?}", m.clone());
Ok(m.body
.parse::<u64>()
.context(libwallet::ErrorKind::ClientCallback(
"Parsing get_height response".to_owned(),
))?)
}
fn get_outputs_from_node(
&self,
wallet_outputs: Vec<pedersen::Commitment>,
) -> Result<HashMap<pedersen::Commitment, (String, u64, u64)>, libwallet::Error> {
let query_params: Vec<String> = wallet_outputs
.iter()
.map(|commit| format!("{}", util::to_hex(commit.as_ref().to_vec())))
.collect();
let query_str = query_params.join(",");
let m = WalletProxyMessage {
sender_id: self.id.clone(),
dest: self.node_url().to_owned(),
method: "get_outputs_from_node".to_owned(),
body: query_str,
};
{
let p = self.proxy_tx.lock();
p.send(m).context(libwallet::ErrorKind::ClientCallback(
"Get outputs from node send".to_owned(),
))?;
}
let r = self.rx.lock();
let m = r.recv().unwrap();
let outputs: Vec<api::Output> = serde_json::from_str(&m.body).unwrap();
let mut api_outputs: HashMap<pedersen::Commitment, (String, u64, u64)> = HashMap::new();
for out in outputs {
api_outputs.insert(
out.commit.commit(),
(util::to_hex(out.commit.to_vec()), out.height, out.mmr_index),
);
}
Ok(api_outputs)
}
fn get_outputs_by_pmmr_index(
&self,
start_height: u64,
max_outputs: u64,
) -> Result<
(
u64,
u64,
Vec<(pedersen::Commitment, pedersen::RangeProof, bool, u64, u64)>,
),
libwallet::Error,
> {
let query_str = format!("{},{}", start_height, max_outputs);
let m = WalletProxyMessage {
sender_id: self.id.clone(),
dest: self.node_url().to_owned(),
method: "get_outputs_by_pmmr_index".to_owned(),
body: query_str,
};
{
let p = self.proxy_tx.lock();
p.send(m).context(libwallet::ErrorKind::ClientCallback(
"Get outputs from node by PMMR index send".to_owned(),
))?;
}
let r = self.rx.lock();
let m = r.recv().unwrap();
let o: api::OutputListing = serde_json::from_str(&m.body).unwrap();
let mut api_outputs: Vec<(pedersen::Commitment, pedersen::RangeProof, bool, u64, u64)> =
Vec::new();
for out in o.outputs {
let is_coinbase = match out.output_type {
api::OutputType::Coinbase => true,
api::OutputType::Transaction => false,
};
api_outputs.push((
out.commit,
out.range_proof().unwrap(),
is_coinbase,
out.block_height.unwrap(),
out.mmr_index,
));
}
Ok((o.highest_index, o.last_retrieved_index, api_outputs))
}
}