use crate::bch::cashaddr::version_byte_flags as bch_flags;
use crate::chaindef::BlockHash;
use crate::chaindef::BlockHeader;
use crate::chaindef::ScriptHash;
use crate::chaindef::Transaction;
use crate::chaindef::TxOut;
use crate::doslimit::ConnectionLimits;
use crate::edition::Edition;
use crate::encode::compute_outpoint_hash_from_tx;
use crate::errors::rpc_invalid_params;
use crate::mempool::MEMPOOL_HEIGHT;
use crate::nexa::cashaddr::version_byte_flags as nexa_flags;
use crate::query::queryfilter::QueryFilter;
use crate::query::scripthash_subscriptions::ScripthashSubscriptions;
use crate::query::tx::get_addresses;
use crate::query::Query;
use crate::rpc::daemon::server::ConnectionId;
use crate::rpc::parseutil::{
bool_from_value_or, hash_from_value, str_from_value, usize_from_value, usize_from_value_or,
};
use crate::rpc::rpcstats::RpcStats;
use crate::rpc::scripthash::{
get_balance, get_first_spend, get_first_use, get_history, get_mempool, listunspent,
};
use crate::scripthash::addr_to_scripthash;
use crate::scripthash::decode_address;
use crate::signal::NetworkNotifier;
use anyhow::{Context, Result};
use bitcoincash::consensus::encode::{deserialize, serialize};
use bitcoincash::hash_types::Txid;
use bitcoincash::hashes::hex::ToHex;
use bitcoincash::Network;
use serde_json::Value;
use std::collections::HashMap;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use tokio::sync::Mutex;
use super::scripthash::get_volume;
struct Subscription {
statushash: Option<[u8; 32]>,
alias: Option<String>,
}
pub struct BlockchainRpc {
query: Arc<Query>,
stats: Arc<RpcStats>,
subscriptions: Mutex<HashMap<ScriptHash, Subscription>>,
last_header_entry: Mutex<Option<BlockHeader>>,
doslimits: ConnectionLimits,
network_notifier: Arc<NetworkNotifier>,
#[allow(dead_code)] network: Network,
alias_bytes_used: AtomicUsize,
conn_id: ConnectionId,
subscription_manager: Arc<ScripthashSubscriptions>,
edition: Edition,
}
#[allow(clippy::too_many_arguments)]
impl BlockchainRpc {
pub fn new(
query: Arc<Query>,
stats: Arc<RpcStats>,
doslimits: ConnectionLimits,
network: Network,
network_notifier: Arc<NetworkNotifier>,
conn_id: ConnectionId,
subscription_manager: Arc<ScripthashSubscriptions>,
edition: Edition,
) -> BlockchainRpc {
BlockchainRpc {
query,
stats,
subscriptions: Mutex::new(HashMap::new()),
last_header_entry: Mutex::new(None), doslimits,
network_notifier,
network,
alias_bytes_used: AtomicUsize::new(0),
conn_id,
subscription_manager,
edition,
}
}
pub fn set_edition(&mut self, edition: Edition) {
self.edition = edition;
}
pub fn edition(&self) -> Edition {
self.edition
}
pub(crate) fn address_decode(&self, params: &[Value]) -> Result<Value> {
let address = str_from_value(params.first(), "address")?;
let (payload, payload_type) = decode_address(&address)?;
assert_eq!(bch_flags::TYPE_P2PKH, nexa_flags::TYPE_P2PKH);
assert_eq!(bch_flags::TYPE_P2SH, nexa_flags::TYPE_P2SH);
let token_aware = [
bch_flags::TYPE_P2PKH_TOKEN,
bch_flags::TYPE_P2SH_TOKEN,
nexa_flags::TYPE_SCRIPT_TEMPLATE,
nexa_flags::TYPE_GROUP,
];
Ok(json!({
"payload": payload.to_hex(),
"type": match payload_type {
bch_flags::TYPE_P2PKH | bch_flags::TYPE_P2PKH_TOKEN => "p2pkh",
bch_flags::TYPE_P2SH | bch_flags::TYPE_P2SH_TOKEN => "p2sh",
nexa_flags::TYPE_SCRIPT_TEMPLATE => "scripttemplate",
nexa_flags::TYPE_GROUP => "group",
_=> "Unknown"
},
"is_token_aware": token_aware.contains(&payload_type)
}))
}
pub async fn address_get_balance(&self, params: &[Value]) -> Result<Value> {
let addr = str_from_value(params.first(), "address")?;
let filter = QueryFilter::from_param(params.get(1))?;
let scripthash = addr_to_scripthash(&addr)?;
get_balance(&self.query, scripthash, &filter).await
}
pub async fn address_get_first_use(&self, params: &[Value]) -> Result<Value> {
let addr = str_from_value(params.first(), "address")?;
let filter = QueryFilter::from_param(params.get(1))?;
let scripthash = addr_to_scripthash(&addr)?;
get_first_use(&self.query, scripthash, &filter, self.edition()).await
}
pub async fn address_get_first_spend(&self, params: &[Value]) -> Result<Value> {
let addr = str_from_value(params.first(), "address")?;
let filter = QueryFilter::from_param(params.get(1))?;
let scripthash = addr_to_scripthash(&addr)?;
get_first_spend(&self.query, scripthash, &filter).await
}
pub async fn address_get_history(&self, params: &[Value]) -> Result<Value> {
let addr = str_from_value(params.first(), "address")?;
let filter = QueryFilter::from_param(params.get(1))?;
let scripthash = addr_to_scripthash(&addr)?;
get_history(
self.query.confirmed_index(),
self.query.unconfirmed_index(),
scripthash,
filter,
)
.await
}
pub async fn address_get_mempool(&self, params: &[Value]) -> Result<Value> {
let addr = str_from_value(params.first(), "address")?;
let filter = QueryFilter::from_param(params.get(1))?;
let scripthash = addr_to_scripthash(&addr)?;
get_mempool(&self.query, scripthash, filter).await
}
pub fn address_get_scripthash(&self, params: &[Value]) -> Result<Value> {
let scripthash = addr_to_scripthash(&str_from_value(params.first(), "address")?)?;
Ok(json!(scripthash))
}
pub async fn address_get_volume(&self, params: &[Value]) -> Result<Value> {
let addr = str_from_value(params.first(), "address")?;
let filter = QueryFilter::from_param(params.get(1))?;
let scripthash = addr_to_scripthash(&addr)?;
get_volume(&self.query, scripthash, &filter).await
}
pub async fn address_listunspent(&self, params: &[Value]) -> Result<Value> {
let addr = str_from_value(params.first(), "address")?;
let filter = QueryFilter::from_param(params.get(1))?;
let scripthash = addr_to_scripthash(&addr)?;
listunspent(&self.query, scripthash, filter).await
}
pub async fn address_subscribe(&self, params: &[Value]) -> Result<Value> {
let addr = str_from_value(params.first(), "address")?;
let scripthash = addr_to_scripthash(&addr)?;
self.remove_subscription(&scripthash).await;
self.doslimits
.check_subscriptions(self.get_num_subscriptions().await as u32 + 1)?;
self.doslimits
.check_alias_usage(self.alias_bytes_used.load(Ordering::Relaxed) + addr.len())?;
let statushash = self
.subscription_manager
.subscribe(self.conn_id, scripthash, Some(addr.clone()))
.await?;
let result = statushash.map_or(Value::Null, |h| json!(hex::encode(h)));
self.alias_bytes_used
.fetch_add(addr.len(), Ordering::Relaxed);
self.subscriptions.lock().await.insert(
scripthash,
Subscription {
statushash,
alias: Some(addr),
},
);
self.stats.subscriptions.inc();
Ok(result)
}
pub async fn address_unsubscribe(&self, params: &[Value]) -> Result<Value> {
let addr = str_from_value(params.first(), "address")?;
let scripthash = addr_to_scripthash(&addr)?;
Ok(json!(self.remove_subscription(&scripthash).await))
}
pub(crate) async fn block_get(&self, params: &[Value]) -> Result<Value> {
if let Ok(blockhash) = hash_from_value::<BlockHash>(params.first()) {
let block = self.query.get_block(&blockhash).await?;
return Ok(json!(hex::encode(serialize(&block))));
}
if let Ok(height) = usize_from_value(params.first(), "height") {
let header = self
.query
.header()
.at_height(height)
.await
.ok_or_else(|| rpc_invalid_params("Invalid block height".to_string()))?;
let block = self.query.get_block(&header.block_hash()).await?;
return Ok(json!(hex::encode(serialize(&block))));
}
Err(rpc_invalid_params(
"Argument must be hex encoded blockhash or numeric height".to_string(),
))
}
pub async fn block_header(&self, params: &[Value]) -> Result<Value> {
let height = usize_from_value(params.first(), "height")?;
let cp_height = usize_from_value_or(params.get(1), "cp_height", 0)?;
let raw_header_hex: String = self
.query
.get_headers(&[height])
.await
.into_iter()
.map(|entry| hex::encode(serialize(&entry)))
.collect();
if cp_height == 0 {
return Ok(json!(raw_header_hex));
}
let (branch, root) = self
.query
.get_header_merkle_proof(height, cp_height)
.await?;
let branch_vec: Vec<String> = branch.into_iter().map(|b| b.to_hex()).collect();
Ok(json!({
"header": raw_header_hex,
"root": root.to_hex(),
"branch": branch_vec
}))
}
pub async fn block_header_verbose(&self, params: &[Value]) -> Result<Value> {
let height = match usize_from_value(params.first(), "height") {
Ok(h) => h,
Err(_) => {
match hash_from_value::<BlockHash>(params.first()) {
Ok(hash) => match self.query.chain().get_block_height(&hash).await {
Some(h) => h as usize,
None => {
return Err(rpc_invalid_params(format!(
"block '{}' does not exist",
hash.to_hex()
)))
}
},
Err(_) => {
return Err(rpc_invalid_params(
"expected block height or block hash".into(),
));
}
}
}
};
let header = match self.query.chain().get_block_header(height).await {
Some(h) => h,
None => {
return Err(rpc_invalid_params(format!(
"block at height {} does not exist",
height
)))
}
};
let mtp = self
.query
.chain()
.get_mtp(height)
.await
.context("unable to calculate mtp")?;
#[cfg(bch)]
{
Ok(json!({
"version": header.version,
"previousblockhash": header.prev_blockhash.to_hex(),
"merkleroot": header.merkle_root.to_hex(),
"time": header.time,
"bits": header.bits,
"nonce": header.nonce,
"hash": header.block_hash(),
"height": height,
"hex": serialize(&header).to_hex(),
"mediantime": mtp,
}))
}
#[cfg(nexa)]
{
Ok(json!({
"previousblockhash": header.prev_blockhash.to_hex(),
"bits": header.bits,
"ancestorhash": header.ancestor.to_hex(),
"merkleroot": header.merkle_root.to_hex(),
"txfilter": header.tx_filter,
"time": header.time,
"height": header.height.0,
"chainwork": header.chain_work,
"size": header.size,
"txcount": header.tx_count.0,
"feepoolamt": header.fee_pool_amt.0,
"utxocommitment": header.utxo_commitment.to_hex(),
"minerdata": header.miner_data.to_hex(),
"nonce": header.nonce.to_hex(),
"mediantime": mtp,
"hex": serialize(&header).to_hex(),
"hash": header.block_hash()
}))
}
}
pub async fn block_headers(&self, params: &[Value]) -> Result<Value> {
let start_height = usize_from_value(params.first(), "start_height")?;
let count = usize_from_value(params.get(1), "count")?;
let cp_height = usize_from_value_or(params.get(2), "cp_height", 0)?;
let heights: Vec<usize> = (start_height..(start_height + count)).collect();
let headers: Vec<String> = self
.query
.get_headers(&heights)
.await
.into_iter()
.map(|entry| hex::encode(serialize(&entry)))
.collect();
if count == 0 || cp_height == 0 {
return Ok(json!({
"count": headers.len(),
"hex": headers.join(""),
"max": 2016,
}));
}
let (branch, root) = self
.query
.get_header_merkle_proof(start_height + (count - 1), cp_height)
.await?;
let branch_vec: Vec<String> = branch.into_iter().map(|b| b.to_hex()).collect();
Ok(json!({
"count": headers.len(),
"hex": headers.join(""),
"max": 2016,
"root": root.to_hex(),
"branch" : branch_vec
}))
}
pub async fn estimatefee(&self, params: &[Value]) -> Result<Value> {
let blocks_count = usize_from_value(params.first(), "blocks_count")?;
let fee_rate = self.query.estimate_fee(blocks_count).await; Ok(json!(fee_rate.max(self.query.get_relayfee().await?)))
}
pub async fn headers_tip(&self) -> Result<Value> {
let (height, header) = self.query.get_best_header().await?;
let hex = hex::encode(serialize(&header));
Ok(json!({"hex":hex, "height":height}))
}
pub async fn headers_subscribe(&self) -> Result<Value> {
let (height, header) = self.query.get_best_header().await?;
let hex = hex::encode(serialize(&header));
let result = json!({"hex":hex, "height":height});
let mut last_entry = self.last_header_entry.lock().await;
*last_entry = Some(header);
Ok(result)
}
pub async fn relayfee(&self) -> Result<Value> {
Ok(json!(self.query.get_relayfee().await?)) }
pub async fn scripthash_get_balance(&self, params: &[Value]) -> Result<Value> {
let scripthash = hash_from_value::<ScriptHash>(params.first())?;
let filter = QueryFilter::from_param(params.get(1))?;
get_balance(&self.query, scripthash, &filter).await
}
pub async fn scripthash_get_first_use(&self, params: &[Value]) -> Result<Value> {
let scripthash = hash_from_value::<ScriptHash>(params.first())?;
let filter = QueryFilter::from_param(params.get(1))?;
get_first_use(&self.query, scripthash, &filter, self.edition()).await
}
pub async fn scripthash_get_first_spend(&self, params: &[Value]) -> Result<Value> {
let scripthash = hash_from_value::<ScriptHash>(params.first())?;
let filter = QueryFilter::from_param(params.get(1))?;
get_first_spend(&self.query, scripthash, &filter).await
}
pub async fn scripthash_get_history(&self, params: &[Value]) -> Result<Value> {
let scripthash = hash_from_value::<ScriptHash>(params.first())?;
let filter = QueryFilter::from_param(params.get(1))?;
get_history(
self.query.confirmed_index(),
self.query.unconfirmed_index(),
scripthash,
filter,
)
.await
}
pub async fn scripthash_get_mempool(&self, params: &[Value]) -> Result<Value> {
let scripthash = hash_from_value::<ScriptHash>(params.first())?;
let filter = QueryFilter::from_param(params.get(1))?;
get_mempool(&self.query, scripthash, filter).await
}
pub async fn scripthash_get_volume(&self, params: &[Value]) -> Result<Value> {
let scripthash = hash_from_value::<ScriptHash>(params.first())?;
let filter = QueryFilter::from_param(params.get(1))?;
get_volume(&self.query, scripthash, &filter).await
}
pub async fn scripthash_listunspent(&self, params: &[Value]) -> Result<Value> {
let scripthash = hash_from_value::<ScriptHash>(params.first())?;
let filter = QueryFilter::from_param(params.get(1))?;
listunspent(&self.query, scripthash, filter).await
}
pub async fn scripthash_subscribe(&self, params: &[Value]) -> Result<Value> {
let scripthash = hash_from_value::<ScriptHash>(params.first())?;
self.remove_subscription(&scripthash).await;
self.doslimits
.check_subscriptions(self.get_num_subscriptions().await as u32 + 1)?;
let statushash = self
.subscription_manager
.subscribe(self.conn_id, scripthash, None)
.await?;
let result = statushash.map_or(Value::Null, |h| json!(hex::encode(h)));
self.subscriptions.lock().await.insert(
scripthash,
Subscription {
statushash,
alias: None,
},
);
self.stats.subscriptions.inc();
Ok(result)
}
pub async fn scripthash_unsubscribe(&self, params: &[Value]) -> Result<Value> {
let scripthash = hash_from_value::<ScriptHash>(params.first())?;
Ok(json!(self.remove_subscription(&scripthash).await))
}
fn parse_broadcast_opt(&self, param: Option<&Value>) -> Result<bool> {
let opt = match param {
Some(p) => p,
None => return Ok(false),
};
if let Some(p2p) = opt.get("p2p_only") {
let p2p = p2p.as_bool().context(rpc_invalid_params(
"option 'p2p_only' not a boolean".to_string(),
))?;
if p2p {
if self.network != Network::Regtest {
bail!("p2p broadcast only supported on {}", Network::Regtest)
}
return Ok(true);
}
}
Ok(false)
}
pub async fn transaction_broadcast(&self, params: &[Value]) -> Result<Value> {
let tx = params
.first()
.context(rpc_invalid_params("missing tx".to_string()))?;
let tx = tx
.as_str()
.context(rpc_invalid_params("non-string tx".to_string()))?;
let tx = hex::decode(tx).context(rpc_invalid_params("non-hex tx".to_string()))?;
let tx: Transaction =
deserialize(&tx).context(rpc_invalid_params("failed to parse tx".to_string()))?;
let p2p_broadcast = self.parse_broadcast_opt(params.get(1))?;
if p2p_broadcast {
#[cfg(nexa)]
let txidentifier = tx.txidem();
#[cfg(bch)]
let txidentifier = tx.txid();
debug!("p2p broadcast of tx {}", txidentifier);
self.query
.broadcast_p2p(tx)
.await
.context("p2p broadcast failed")?;
return Ok(json!(txidentifier.to_hex()));
}
let txid = self
.query
.broadcast(&tx)
.await
.context(rpc_invalid_params("rejected by network".to_string()))?;
if let Err(e) = self.network_notifier.full_tx_channel.sender().try_send(tx) {
debug!("Failed to notify self of broadcasted tx: {}", e);
}
Ok(json!(txid.to_hex()))
}
pub async fn transaction_get(&self, params: &[Value]) -> Result<Value> {
let tx_hash = hash_from_value::<Txid>(params.first())?;
let verbose = match params.get(1) {
Some(value) => value.as_bool().context("non-bool verbose value")?,
None => false,
};
if !verbose {
let tx = self.query.tx().get(&tx_hash, None, None, true).await?;
Ok(json!(hex::encode(serialize(&tx))))
} else {
self.query.tx().get_verbose(&tx_hash, true).await
}
}
pub async fn transaction_get_confirmed_blockhash(&self, params: &[Value]) -> Result<Value> {
let tx_hash: Txid = hash_from_value(params.first()).context("bad tx_hash")?;
self.query.get_confirmed_blockhash(&tx_hash).await
}
pub async fn transaction_get_merkle(&self, params: &[Value]) -> Result<Value> {
let tx_hash = hash_from_value::<Txid>(params.first())?;
let height = if params.get(1).is_some() {
usize_from_value(params.get(1), "height")
} else {
let header = self.query.header().get_by_txid(&tx_hash, None).await;
match header {
Ok(Some(header)) => {
let height = self
.query
.header()
.get_height(&header)
.await
.context(anyhow!("Block height for {} missing", header.block_hash()))?;
Ok(height as usize)
}
_ => Err(rpc_invalid_params(
"Transaction is not confirmed in a block".to_string(),
)),
}
}?;
let (merkle, pos) = self
.query
.get_merkle_proof(&tx_hash, height)
.await
.context("cannot create merkle proof")?;
let merkle: Vec<String> = merkle.into_iter().map(|txid| txid.to_hex()).collect();
Ok(json!({
"block_height": height,
"merkle": merkle,
"pos": pos}))
}
pub async fn transaction_id_from_pos(&self, params: &[Value]) -> Result<Value> {
let height = usize_from_value(params.first(), "height")?;
let tx_pos = usize_from_value(params.get(1), "tx_pos")?;
let want_merkle = bool_from_value_or(params.get(2), "merkle", false)?;
let (txid, merkle) = self
.query
.get_id_from_pos(height, tx_pos, want_merkle)
.await?;
if !want_merkle {
return Ok(json!(txid.to_hex()));
}
let merkle_vec: Vec<String> = merkle.into_iter().map(|entry| entry.to_hex()).collect();
Ok(json!({
"tx_hash" : txid.to_hex(),
"merkle" : merkle_vec}))
}
#[cfg(nexa)]
pub async fn utxo_get(&self, params: &[Value]) -> Result<Value> {
use crate::{
nexa::transaction::{parse_script_template, TxOutType},
rpc::parseutil::param_to_outpointhash,
};
let outpoint = param_to_outpointhash(params.first())?;
let (tx, output_index, _) = match self.query.get_tx_funding_prevout(&outpoint).await? {
Some(found) => found,
None => return Err(rpc_invalid_params("Outpoint not found".to_string())),
};
let (mut value, utxo) = self.utxo_get_inner(&tx, output_index).await?;
{
let v = value.as_object_mut().unwrap();
v.insert("tx_idem".to_string(), json!(tx.txidem().to_hex()));
v.insert("tx_hash".to_string(), json!(tx.txid().to_hex()));
v.insert("tx_pos".to_string(), json!(output_index));
let template = if utxo.txout_type == TxOutType::TEMPLATE as u8 {
parse_script_template(&utxo.script_pubkey)
} else {
None
};
v.insert(
"group".to_string(),
json!(template.as_ref().map(|t| t
.token
.as_ref()
.map(|token| token.as_cashaddr(self.network)))),
);
v.insert(
"token_id_hex".to_string(),
json!(template
.as_ref()
.map(|t| t.token.as_ref().map(|token| token.to_hex()))),
);
v.insert(
"group_quantity".to_string(),
json!(template.as_ref().map(|t| t.quantity)),
);
v.insert(
"group_authority".to_string(),
json!(template.as_ref().map(|t| t.group_authority().unwrap_or(0))),
);
v.insert(
"template_scripthash".to_string(),
json!(template.as_ref().map(|t| t.scripthash_human())),
);
v.insert(
"template_argumenthash".to_string(),
json!(template.map(|t| t.argumenthash.to_hex())),
);
}
Ok(value)
}
#[cfg(bch)]
pub async fn utxo_get(&self, params: &[Value]) -> Result<Value> {
let txid = hash_from_value::<Txid>(params.first())?;
let out_n = usize_from_value(params.get(1), "out_n")?;
if out_n > u32::MAX as usize {
return Err(rpc_invalid_params(format!(
"Too large value for out_n parameter ({} > {})",
out_n,
u32::MAX
)));
}
let utxo_creation_tx = self.query.tx().get(&txid, None, None, false).await?;
let (mut value, utxo) = self.utxo_get_inner(&utxo_creation_tx, out_n as u32).await?;
{
let v = value.as_object_mut().unwrap();
v.insert(
"token_id".to_string(),
json!(utxo.token.as_ref().map(|t| t.id.to_hex())),
);
v.insert(
"commitment".to_string(),
json!(utxo.token.as_ref().map(|t| t.commitment.to_hex())),
);
v.insert(
"token_amount".to_string(),
json!(utxo.token.as_ref().map(|t| t.amount)),
);
v.insert(
"token_bitfield".to_string(),
json!(utxo.token.as_ref().map(|t| t.bitfield)),
);
}
Ok(value)
}
async fn utxo_get_inner(
&self,
utxo_creation_tx: &Transaction,
out_n: u32,
) -> Result<(Value, TxOut)> {
let utxo = match utxo_creation_tx.output.get(out_n as usize) {
Some(utxo) => utxo,
None => {
bail!(rpc_invalid_params(format!(
"out_n {} does not exist on tx {}, the transaction has {} outputs",
out_n,
utxo_creation_tx.txid(),
utxo_creation_tx.output.len()
)));
}
};
let utxo_spending_tx = self
.query
.get_tx_spending_prevout(&compute_outpoint_hash_from_tx(utxo_creation_tx, out_n))
.await?;
let status = if utxo_spending_tx.is_some() {
"spent"
} else {
"unspent"
};
let spent_json = match utxo_spending_tx {
Some((tx, input_index, height)) => {
json!({
"tx_hash": Some(tx.txid().to_string()),
"tx_pos": Some(input_index),
"height": Some(if height == MEMPOOL_HEIGHT { 0 } else { height}),
})
}
None => {
json!({
"tx_hash": None::<String>,
"tx_pos": None::<u32>,
"height": None::<i64>,
})
}
};
let utxo_confirmation_height = self
.query
.tx()
.get_confirmation_height(&utxo_creation_tx.txid())
.await;
let utxo_scripthash = ScriptHash::from_script(&utxo.script_pubkey);
Ok((
json!({
"status": status,
"amount": utxo.value,
"scriptpubkey": utxo.script_pubkey,
"scripthash": utxo_scripthash,
"height": utxo_confirmation_height,
"spent": spent_json,
"addresses": get_addresses(utxo, self.network),
}),
utxo.clone(),
))
}
pub async fn on_chaintip_change(&self, chaintip: BlockHeader) -> Result<Option<Value>> {
let timer = self
.stats
.latency
.with_label_values(&["chaintip_update"])
.start_timer();
let mut last_entry = self.last_header_entry.lock().await;
if last_entry.is_none() {
return Ok(None);
}
if last_entry.as_ref() == Some(&chaintip) {
return Ok(None);
}
let height = self
.query
.header()
.get_height(&chaintip)
.await
.context("header missing")?;
*last_entry = Some(chaintip);
let hex_header = hex::encode(serialize(&last_entry.as_ref().unwrap()));
let header = json!({"hex": hex_header, "height": height});
timer.observe_duration();
Ok(Some(json!({
"jsonrpc": "2.0",
"method": "blockchain.headers.subscribe",
"params": [header]})))
}
pub async fn on_scripthash_change(
&self,
updates: Vec<super::daemon::ScriptHashUpdate>,
) -> Vec<Value> {
if updates.is_empty() {
return Vec::default();
}
let timer = self
.stats
.latency
.with_label_values(&["statushash_update"])
.start_timer();
let mut notifications: Vec<Value> = Vec::default();
let mut subscriptions = self.subscriptions.lock().await;
for update in updates {
let subscription_name = update
.alias
.clone()
.unwrap_or_else(|| update.scripthash.to_hex());
let method = if update.alias.is_some() {
"blockchain.address.subscribe"
} else {
"blockchain.scripthash.subscribe"
};
let new_statushash_hex = update
.new_statushash
.map_or(Value::Null, |h| json!(hex::encode(h)));
notifications.push(json!({
"jsonrpc": "2.0",
"method": method,
"params": [subscription_name, new_statushash_hex]
}));
if let Some(s) = subscriptions.get_mut(&update.scripthash) {
s.statushash = update.new_statushash;
}
}
timer.observe_duration();
notifications
}
pub async fn get_num_subscriptions(&self) -> i64 {
self.subscriptions.lock().await.len() as i64
}
async fn remove_subscription(&self, scripthash: &ScriptHash) -> bool {
let removed = self.subscriptions.lock().await.remove(scripthash);
match removed {
Some(subscription) => {
if let Some(alias) = subscription.alias {
self.alias_bytes_used
.fetch_sub(alias.len(), Ordering::Relaxed);
}
self.subscription_manager
.unsubscribe(self.conn_id, *scripthash)
.await;
self.stats.subscriptions.dec();
true
}
None => false,
}
}
}