1141 lines
43 KiB
Rust
1141 lines
43 KiB
Rust
use crate::common::binary_conversions::binary_to_string;
|
|
use crate::encode;
|
|
use crate::log::warn;
|
|
use crate::records::memory::enums::ClientType;
|
|
use crate::records::memory::response_channels::Command;
|
|
use crate::rpc::server::command_loop_state::next_incoming_command;
|
|
use crate::rpc::server::connection_memory_manager::{
|
|
remove_key_from_memory, remove_stream_from_memory,
|
|
};
|
|
use crate::rpc::*;
|
|
use crate::sled::Db;
|
|
use crate::wallets::structures::Wallet;
|
|
use crate::Arc;
|
|
use crate::AsyncWriteExt;
|
|
use crate::Instant;
|
|
use crate::Mutex;
|
|
use crate::TcpStream;
|
|
|
|
pub async fn start_loop(
|
|
stream_locked: Arc<Mutex<TcpStream>>,
|
|
db: Db,
|
|
connections_key: String,
|
|
wallet: Arc<Wallet>,
|
|
map: Arc<Mutex<Command>>,
|
|
) -> Result<(), String> {
|
|
'outer: loop {
|
|
let Some(incoming_command) =
|
|
next_incoming_command(stream_locked.clone(), &db, &connections_key, wallet.clone())
|
|
.await?
|
|
else {
|
|
break 'outer Ok(());
|
|
};
|
|
let command = incoming_command.command;
|
|
let ip = incoming_command.ip;
|
|
let client_type = incoming_command.client_type;
|
|
let command_started = Instant::now();
|
|
|
|
match command {
|
|
1 => {
|
|
// requst network info
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::network_info::request_network_info(&db).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
2 => {
|
|
// request the current block height
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::block_height::request_block_height(&db).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
4 => {
|
|
// request the current time
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::time::request_time().await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
5 => {
|
|
// request current block difficulty
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::difficulty::request_difficulty(&db).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
6 => {
|
|
// validate a torrent file
|
|
let (uid, result) = commands::validate_torrent::validate(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
&db,
|
|
wallet.clone(),
|
|
)
|
|
.await?;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
7 => {
|
|
// request a piece of a block
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let block_number = read_bytes_from_stream::read_u32_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let block_piece = read_bytes_from_stream::read_u8_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let info_hash = read_bytes_from_stream::read_u128_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::torrent::request_block_piece(
|
|
&db,
|
|
block_number,
|
|
block_piece,
|
|
info_hash,
|
|
)
|
|
.await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
8 => {
|
|
// submit a transaction
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let txtype = read_bytes_from_stream::read_u8_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let txsize = command_maps::get_bytes(txtype);
|
|
if txsize == 0 {
|
|
let result = responses::RpcResponse::Binary(
|
|
"error: No such transaction type".as_bytes().to_vec(),
|
|
);
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
continue;
|
|
}
|
|
let size = (txsize - 1) as usize;
|
|
let tx = read_bytes_from_stream::read_usize_from_stream(
|
|
&connections_key,
|
|
size,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result =
|
|
commands::tx_submit::save_and_submit(txtype, tx, &db, map.clone()).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
9 => {
|
|
// request a block by block height
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let block_number = read_bytes_from_stream::read_u32_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::block_by_height::request_block(block_number).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
10 => {
|
|
// request a block by block hash
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let hash = read_bytes_from_stream::read_usize_from_stream(
|
|
&connections_key,
|
|
32,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let hash = encode(hash);
|
|
|
|
let result = commands::block_by_hash::request_block(&db, &hash).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
11 => {
|
|
// request the latest block
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let result = commands::latest_block::request_latest_block(&db).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
12 => {
|
|
// request the torrent of a given block
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let block_number = read_bytes_from_stream::read_u32_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::torrent_by_block::request_block_torrent(&block_number).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
13 => {
|
|
// request the largest txfee
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::largest_tx_fee::request_largest_tx_fee().await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
14 => {
|
|
// get transaction from mempool by signature
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let signature = read_bytes_from_stream::read_signature_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result =
|
|
commands::memory_by_signature::request_transaction_by_signature(signature)
|
|
.await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
15 => {
|
|
// get total count of tranasctions in mempool
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::tx_count_from_mempool::request_tx_count_from_mempool().await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
16 => {
|
|
// get all transactions of a given address from memory
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let address = read_bytes_from_stream::read_short_address_string_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::transactions_by_address::request_transactions_by_address(
|
|
&db, &address,
|
|
)
|
|
.await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
17 => {
|
|
// get a transaction from given txid
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let hash = read_bytes_from_stream::read_usize_from_stream(
|
|
&connections_key,
|
|
32,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::transaction_by_txid::request_transaction_by_txid_with_block(
|
|
&db, hash,
|
|
)
|
|
.await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
18 => {
|
|
// get total confirmed transaction count
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::tx_count::request_tx_count(&db).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
19 => {
|
|
// get header of given height
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let block_number = read_bytes_from_stream::read_u32_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result =
|
|
commands::block_header_by_height::lookup_by_block_number(&db, block_number)
|
|
.await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
20 => {
|
|
// get header of given block hash
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let hash = read_bytes_from_stream::read_usize_from_stream(
|
|
&connections_key,
|
|
32,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let hash = encode(hash);
|
|
|
|
let result = commands::block_header_by_hash::lookup_by_hash(&db, &hash).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
21 => {
|
|
// get all block headers
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let result = commands::block_headers::get_all_headers(&db).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
22 => {
|
|
// get balance of address for given coin or token
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let coin = read_bytes_from_stream::read_usize_from_stream(
|
|
&connections_key,
|
|
15,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let coin = binary_to_string(coin);
|
|
let address = read_bytes_from_stream::read_short_address_string_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result =
|
|
commands::address_coin_lookup::lookup_wallet_coin(&db, address, coin).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
23 => {
|
|
// get complete balance sheet for a wallet.
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let address = read_bytes_from_stream::read_short_address_string_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result =
|
|
commands::address_complete_balance_sheet::get_token_balances(&db, address)
|
|
.await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
24 => {
|
|
// validate wallet address
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let address = read_bytes_from_stream::read_short_address_string_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::validate_address::validate(address, &db).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
25 => {
|
|
// validate signed message
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let message_size = read_bytes_from_stream::read_u16_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let message_bytes = read_bytes_from_stream::read_usize_from_stream(
|
|
&connections_key,
|
|
message_size as usize,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let message = binary_to_string(message_bytes);
|
|
let address = read_bytes_from_stream::read_short_address_string_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let signature = read_bytes_from_stream::read_signature_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result =
|
|
commands::validate_message::validate(message, signature, address, &db).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
26 => {
|
|
// unblock and ip address
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let ip = read_bytes_from_stream::read_ip_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let signature = read_bytes_from_stream::read_signature_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result =
|
|
commands::block_peer_ip::block_peer(&db, ip, signature, wallet.clone()).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
27 => {
|
|
// block an ip address
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let ip = read_bytes_from_stream::read_ip_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let signature = read_bytes_from_stream::read_signature_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result =
|
|
commands::unblock_peer_ip::unblock_peer(&db, ip, signature, wallet.clone())
|
|
.await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
28 => {
|
|
// add a new network node
|
|
let (uid, result) = commands::add_network_node::add_network_node(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
&db,
|
|
wallet.clone(),
|
|
map.clone(),
|
|
)
|
|
.await?;
|
|
let should_drop_rejected_miner = client_type == ClientType::Miner
|
|
&& matches!(&result, responses::RpcResponse::Binary(bytes) if String::from_utf8_lossy(bytes).starts_with("Error:"));
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
if should_drop_rejected_miner {
|
|
remove_key_from_memory(&connections_key).await;
|
|
break 'outer Ok(());
|
|
}
|
|
}
|
|
50 => {
|
|
// signed monitor-add event from a miner peer
|
|
let (uid, result) = commands::network_monitor_add::network_monitor_add(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
&db,
|
|
wallet.clone(),
|
|
map.clone(),
|
|
client_type,
|
|
)
|
|
.await?;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
51 => {
|
|
// signed monitor-remove event from a miner peer
|
|
let (uid, result) = commands::network_monitor_remove::network_monitor_remove(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
&db,
|
|
wallet.clone(),
|
|
map.clone(),
|
|
client_type,
|
|
)
|
|
.await?;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
30 => {
|
|
// request node list
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::request_valid_nodes::request_valid_nodes().await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
31 => {
|
|
// request token list
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::token_list::get_tokens(&db).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
32 => {
|
|
// request nft list
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::nft_list::get_nfts(&db).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
33 => {
|
|
// request contract and all payments
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let contract_hash = read_bytes_from_stream::read_usize_from_stream(
|
|
&connections_key,
|
|
32,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::contract::contract_details(contract_hash, &db).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
34 => {
|
|
// submit torrent file
|
|
let (uid, result) = commands::receive_torrent::receive_torrent(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
&db,
|
|
wallet.clone(),
|
|
map.clone(),
|
|
)
|
|
.await?;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
35 => {
|
|
// request token details
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let token_name = read_bytes_from_stream::read_usize_from_stream(
|
|
&connections_key,
|
|
15,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let token_name = binary_to_string(token_name);
|
|
|
|
let result = commands::token_lookup::lookup_token_details(&db, token_name).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
36 => {
|
|
// request nft details
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let nft_name = read_bytes_from_stream::read_usize_from_stream(
|
|
&connections_key,
|
|
15,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let nft_name = binary_to_string(nft_name);
|
|
let series = read_bytes_from_stream::read_u32_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::nft_lookup::lookup_nft_details(&db, nft_name, series).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
37 => {
|
|
// request contract details by address
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let address = read_bytes_from_stream::read_short_address_string_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::contract::contract_details_by_address(address, &db).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
38 => {
|
|
// register a deterministic short wallet address
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let short_address = read_bytes_from_stream::read_short_address_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let public_key = read_bytes_from_stream::read_public_key_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let signature = read_bytes_from_stream::read_signature_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::wallet_register::register(
|
|
short_address,
|
|
public_key,
|
|
signature,
|
|
&db,
|
|
map.clone(),
|
|
ip.clone(),
|
|
connections_key.clone(),
|
|
)
|
|
.await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
39 => {
|
|
// request the deterministic short wallet registry snapshot
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::wallet_registry_sync::request_registry(&db).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
40 => {
|
|
// request the vanity address registered to a canonical wallet
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let address = read_bytes_from_stream::read_short_address_string_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::wallet_vanity_lookup::lookup(address, &db).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
41 => {
|
|
// request canonical and staged torrent candidates from snapshot onward
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::torrent_candidates::request_torrent_candidates(&db).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
42 => {
|
|
// request the canonical block hash at a specific height
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let block_number = read_bytes_from_stream::read_u32_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result =
|
|
commands::block_hash_at_height::request_block_hash_at_height(block_number)
|
|
.await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
43 => {
|
|
// request whether a canonical short wallet address is registered
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let short_address = read_bytes_from_stream::read_short_address_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::wallet_registration_status::lookup(short_address, &db).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
44 => {
|
|
// request paged signature-stripped transaction records indexed to a wallet address
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let short_address = read_bytes_from_stream::read_short_address_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let skip = read_bytes_from_stream::read_u32_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let limit = read_bytes_from_stream::read_u32_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result =
|
|
commands::address_history::lookup(short_address, skip, limit, &db).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
45 => {
|
|
// request the canonical short address that owns a registered vanity address
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let vanity_bytes = read_bytes_from_stream::read_short_address_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let vanity_address =
|
|
match crate::wallets::structures::Wallet::bytes_to_vanity_address(&vanity_bytes)
|
|
{
|
|
Some(address) => address,
|
|
None => {
|
|
let result = crate::rpc::responses::RpcResponse::Binary(
|
|
b"error: Invalid vanity address bytes".to_vec(),
|
|
);
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
continue;
|
|
}
|
|
};
|
|
|
|
let result =
|
|
commands::wallet_vanity_owner_lookup::lookup(vanity_address, &db).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
46 => {
|
|
// request latest transaction records for a wallet address in one response
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let short_address = read_bytes_from_stream::read_short_address_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let limit = read_bytes_from_stream::read_u32_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result =
|
|
commands::address_history::latest_transactions(short_address, limit, &db).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
47 => {
|
|
// request marketing campaign summary and paged campaign records
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let advertiser = read_bytes_from_stream::read_short_address_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let campaign = read_bytes_from_stream::read_u64_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let skip = read_bytes_from_stream::read_u32_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let limit = read_bytes_from_stream::read_u32_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::marketing_campaign_history::lookup(
|
|
advertiser, campaign, skip, limit, &db,
|
|
)
|
|
.await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
48 => {
|
|
// request the full token catalog for wallet asset views
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::token_lookup::token_catalog(&db).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
49 => {
|
|
// request whether a loan contract's collateral has been claimed
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let contract_hash = read_bytes_from_stream::read_usize_from_stream(
|
|
&connections_key,
|
|
32,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::contract::collateral_status(contract_hash, &db).await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
52 => {
|
|
// request byte count and node fee for storage data lookup
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let storage_key = read_bytes_from_stream::read_usize_from_stream(
|
|
&connections_key,
|
|
32,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let data_key = read_bytes_from_stream::read_usize_from_stream(
|
|
&connections_key,
|
|
50,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let address = read_bytes_from_stream::read_short_address_string_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::storage_lookup::lookup_cost(
|
|
&db,
|
|
storage_key,
|
|
data_key,
|
|
address,
|
|
wallet.clone(),
|
|
)
|
|
.await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
53 => {
|
|
// paid storage data lookup
|
|
let (uid, _) = read_bytes_from_stream::read_uid_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let storage_key = read_bytes_from_stream::read_usize_from_stream(
|
|
&connections_key,
|
|
32,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let data_key = read_bytes_from_stream::read_usize_from_stream(
|
|
&connections_key,
|
|
50,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let address = read_bytes_from_stream::read_short_address_string_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let txtype = read_bytes_from_stream::read_u8_from_stream(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
let txsize = command_maps::get_bytes(txtype);
|
|
if txsize == 0 {
|
|
let result = responses::RpcResponse::Binary(
|
|
b"{\"error\":\"No such transaction type\"}".to_vec(),
|
|
);
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
continue;
|
|
}
|
|
// Read the full transaction body for the declared payment type.
|
|
// For paid storage lookups this should be a type 2 transfer
|
|
// body; the tx type byte was already consumed above.
|
|
let tx = read_bytes_from_stream::read_usize_from_stream(
|
|
&connections_key,
|
|
txsize - 1,
|
|
stream_locked.clone(),
|
|
)
|
|
.await?;
|
|
|
|
let result = commands::storage_lookup::lookup_store(
|
|
&db,
|
|
storage_key,
|
|
data_key,
|
|
address,
|
|
txtype,
|
|
tx,
|
|
wallet.clone(),
|
|
map.clone(),
|
|
)
|
|
.await;
|
|
result
|
|
.send(&stream_locked, Some(&connections_key), uid)
|
|
.await;
|
|
}
|
|
255 => {
|
|
commands::route_reply::route_reply(
|
|
&connections_key,
|
|
stream_locked.clone(),
|
|
&db,
|
|
wallet.clone(),
|
|
map.clone(),
|
|
&ip,
|
|
client_type,
|
|
)
|
|
.await?;
|
|
}
|
|
_ => {
|
|
// Unknown commands are ignored at the protocol level but
|
|
// still count as bad RPC behavior for scoring purposes.
|
|
commands::bad_rpc_call::record(&ip, client_type, &db, wallet.clone()).await;
|
|
}
|
|
}
|
|
|
|
let elapsed_ms = command_started.elapsed().as_millis();
|
|
if elapsed_ms >= 1_000 {
|
|
warn!(
|
|
"[rpc_trace] slow handler: cmd={command} peer={connections_key} elapsed_ms={elapsed_ms}"
|
|
);
|
|
}
|
|
|
|
// Wallet-backed client sessions are short-lived request/response
|
|
// connections, so close them after each non-reply command.
|
|
if client_type == ClientType::Client && command != command_maps::RPC_REPLY {
|
|
let mut stream = stream_locked.lock().await;
|
|
let _ = stream.shutdown().await;
|
|
drop(stream);
|
|
remove_stream_from_memory(&stream_locked).await;
|
|
break 'outer Ok(());
|
|
}
|
|
}
|
|
}
|