Sasedev/khadhroony-bobobot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

0.3.1

Browse Source
This commit is contained in:
SinuS von SifriduS
2026-04-21 07:46:18 +02:00
parent bbad47b0e8
commit 83a8a879aa
4 changed files with 292 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

289
kb_lib/src/ws_client.rs
View File

@@ -557,6 +557,171 @@ impl WsClient {
}
}
/// Subscribes to account change notifications.
pub async fn account_subscribe(
&self,
pubkey: std::string::String,
config: std::option::Option<serde_json::Value>,
) -> Result<u64, crate::KbError> {
let params = kb_build_single_key_optional_config_params(pubkey, config);
self.send_json_rpc_request("accountSubscribe".to_string(), params)
.await
}
/// Unsubscribes from an account change subscription.
pub async fn account_unsubscribe(&self, subscription_id: u64) -> Result<u64, crate::KbError> {
self.send_json_rpc_request(
"accountUnsubscribe".to_string(),
vec![serde_json::Value::from(subscription_id)],
)
.await
}
/// Subscribes to block notifications.
///
/// The Solana RPC documentation marks this subscription as unstable.
pub async fn block_subscribe(
&self,
filter: serde_json::Value,
config: std::option::Option<serde_json::Value>,
) -> Result<u64, crate::KbError> {
let params = kb_build_first_value_optional_config_params(filter, config);
self.send_json_rpc_request("blockSubscribe".to_string(), params)
.await
}
/// Unsubscribes from a block notification subscription.
pub async fn block_unsubscribe(&self, subscription_id: u64) -> Result<u64, crate::KbError> {
self.send_json_rpc_request(
"blockUnsubscribe".to_string(),
vec![serde_json::Value::from(subscription_id)],
)
.await
}
/// Subscribes to transaction log notifications.
pub async fn logs_subscribe(
&self,
filter: serde_json::Value,
config: std::option::Option<serde_json::Value>,
) -> Result<u64, crate::KbError> {
let params = kb_build_first_value_optional_config_params(filter, config);
self.send_json_rpc_request("logsSubscribe".to_string(), params)
.await
}
/// Unsubscribes from a logs subscription.
pub async fn logs_unsubscribe(&self, subscription_id: u64) -> Result<u64, crate::KbError> {
self.send_json_rpc_request(
"logsUnsubscribe".to_string(),
vec![serde_json::Value::from(subscription_id)],
)
.await
}
/// Subscribes to program-owned account notifications.
pub async fn program_subscribe(
&self,
program_id: std::string::String,
config: std::option::Option<serde_json::Value>,
) -> Result<u64, crate::KbError> {
let params = kb_build_single_key_optional_config_params(program_id, config);
self.send_json_rpc_request("programSubscribe".to_string(), params)
.await
}
/// Unsubscribes from a program subscription.
pub async fn program_unsubscribe(&self, subscription_id: u64) -> Result<u64, crate::KbError> {
self.send_json_rpc_request(
"programUnsubscribe".to_string(),
vec![serde_json::Value::from(subscription_id)],
)
.await
}
/// Subscribes to root notifications.
pub async fn root_subscribe(&self) -> Result<u64, crate::KbError> {
self.send_json_rpc_request("rootSubscribe".to_string(), std::vec::Vec::new())
.await
}
/// Unsubscribes from a root subscription.
pub async fn root_unsubscribe(&self, subscription_id: u64) -> Result<u64, crate::KbError> {
self.send_json_rpc_request(
"rootUnsubscribe".to_string(),
vec![serde_json::Value::from(subscription_id)],
)
.await
}
/// Subscribes to one transaction signature status.
pub async fn signature_subscribe(
&self,
signature: std::string::String,
config: std::option::Option<serde_json::Value>,
) -> Result<u64, crate::KbError> {
let params = kb_build_single_key_optional_config_params(signature, config);
self.send_json_rpc_request("signatureSubscribe".to_string(), params)
.await
}
/// Unsubscribes from a signature subscription.
pub async fn signature_unsubscribe(&self, subscription_id: u64) -> Result<u64, crate::KbError> {
self.send_json_rpc_request(
"signatureUnsubscribe".to_string(),
vec![serde_json::Value::from(subscription_id)],
)
.await
}
/// Subscribes to slot notifications.
pub async fn slot_subscribe(&self) -> Result<u64, crate::KbError> {
self.send_json_rpc_request("slotSubscribe".to_string(), std::vec::Vec::new())
.await
}
/// Unsubscribes from a slot subscription.
pub async fn slot_unsubscribe(&self, subscription_id: u64) -> Result<u64, crate::KbError> {
self.send_json_rpc_request(
"slotUnsubscribe".to_string(),
vec![serde_json::Value::from(subscription_id)],
)
.await
}
/// Subscribes to slot lifecycle update notifications.
pub async fn slots_updates_subscribe(&self) -> Result<u64, crate::KbError> {
self.send_json_rpc_request("slotsUpdatesSubscribe".to_string(), std::vec::Vec::new())
.await
}
/// Unsubscribes from a slot lifecycle update subscription.
pub async fn slots_updates_unsubscribe(
&self,
subscription_id: u64,
) -> Result<u64, crate::KbError> {
self.send_json_rpc_request(
"slotsUpdatesUnsubscribe".to_string(),
vec![serde_json::Value::from(subscription_id)],
)
.await
}
/// Subscribes to vote notifications.
pub async fn vote_subscribe(&self) -> Result<u64, crate::KbError> {
self.send_json_rpc_request("voteSubscribe".to_string(), std::vec::Vec::new())
.await
}
/// Unsubscribes from a vote subscription.
pub async fn vote_unsubscribe(&self, subscription_id: u64) -> Result<u64, crate::KbError> {
self.send_json_rpc_request(
"voteUnsubscribe".to_string(),
vec![serde_json::Value::from(subscription_id)],
)
.await
}
/// Initiates the close handshake.
pub async fn send_close(&self) -> Result<(), crate::KbError> {
self.send_message(WsOutgoingMessage::Close).await
@@ -1223,7 +1388,6 @@ fn kb_build_pending_json_rpc_request(
) -> std::option::Option<WsPendingJsonRpcRequest> {
let request_id_option = kb_json_value_to_u64(&request.id);
let request_id = request_id_option?;
if kb_is_subscribe_method(&request.method) {
let notification_method_option =
kb_infer_notification_method_from_subscribe(&request.method);
@@ -1261,6 +1425,28 @@ fn kb_build_pending_json_rpc_request(
})
}
fn kb_build_single_key_optional_config_params(
key: std::string::String,
config: std::option::Option<serde_json::Value>,
) -> std::vec::Vec<serde_json::Value> {
let mut params = vec![serde_json::Value::String(key)];
if let Some(config) = config {
params.push(config);
}
params
}
fn kb_build_first_value_optional_config_params(
first: serde_json::Value,
config: std::option::Option<serde_json::Value>,
) -> std::vec::Vec<serde_json::Value> {
let mut params = vec![first];
if let Some(config) = config {
params.push(config);
}
params
}
#[cfg(test)]
mod tests {
use futures_util::SinkExt;
@@ -1738,4 +1924,105 @@ mod tests {
client.disconnect().await.expect("disconnect must succeed");
server.shutdown().await;
}
#[tokio::test]
async fn helper_methods_send_expected_subscribe_method_names() {
let server = TestWsServer::spawn_json_rpc_server().await;
let endpoint = make_ws_endpoint(server.url.clone());
let client = crate::WsClient::new(endpoint).expect("client creation must succeed");
let mut receiver = client.subscribe_events();
client.connect().await.expect("connect must succeed");
let _ = recv_event(&mut receiver).await;
let _ = client
.account_subscribe(
"CM78CPUeXjn8o3yroDHxUtKsZZgoy4GPkPPXfouKNH12".to_string(),
Some(serde_json::json!({"encoding": "jsonParsed"})),
)
.await
.expect("account_subscribe must succeed");
let _ = client
.block_subscribe(
serde_json::json!("all"),
Some(serde_json::json!({"commitment": "confirmed"})),
)
.await
.expect("block_subscribe must succeed");
let _ = client
.logs_subscribe(
serde_json::json!("all"),
Some(serde_json::json!({"commitment": "finalized"})),
)
.await
.expect("logs_subscribe must succeed");
let _ = client
.program_subscribe(
"11111111111111111111111111111111".to_string(),
Some(serde_json::json!({"encoding": "base64"})),
)
.await
.expect("program_subscribe must succeed");
let _ = client.root_subscribe().await.expect("root_subscribe must succeed");
let _ = client
.signature_subscribe(
"2EBVM6cB8vAAD93Ktr6Vd8p67XPbQzCJX47MpReuiCXJAtcjaxpvWpcg9Ege1Nr5Tk3a2GFrByT7WPBjdsTycY9b".to_string(),
Some(serde_json::json!({"commitment": "confirmed"})),
)
.await
.expect("signature_subscribe must succeed");
let _ = client.slot_subscribe().await.expect("slot_subscribe must succeed");
let _ = client
.slots_updates_subscribe()
.await
.expect("slots_updates_subscribe must succeed");
let _ = client.vote_subscribe().await.expect("vote_subscribe must succeed");
tokio::time::sleep(std::time::Duration::from_millis(150)).await;
let observed_methods = server.observed_methods_snapshot().await;
assert!(observed_methods.iter().any(|method| method == "accountSubscribe"));
assert!(observed_methods.iter().any(|method| method == "blockSubscribe"));
assert!(observed_methods.iter().any(|method| method == "logsSubscribe"));
assert!(observed_methods.iter().any(|method| method == "programSubscribe"));
assert!(observed_methods.iter().any(|method| method == "rootSubscribe"));
assert!(observed_methods.iter().any(|method| method == "signatureSubscribe"));
assert!(observed_methods.iter().any(|method| method == "slotSubscribe"));
assert!(observed_methods.iter().any(|method| method == "slotsUpdatesSubscribe"));
assert!(observed_methods.iter().any(|method| method == "voteSubscribe"));
client.disconnect().await.expect("disconnect must succeed");
server.shutdown().await;
}
#[tokio::test]
async fn helper_methods_send_expected_unsubscribe_method_names() {
let server = TestWsServer::spawn_json_rpc_server().await;
let endpoint = make_ws_endpoint(server.url.clone());
let client = crate::WsClient::new(endpoint).expect("client creation must succeed");
let mut receiver = client.subscribe_events();
client.connect().await.expect("connect must succeed");
let _ = recv_event(&mut receiver).await;
let _ = client.account_unsubscribe(10).await.expect("account_unsubscribe must succeed");
let _ = client.block_unsubscribe(11).await.expect("block_unsubscribe must succeed");
let _ = client.logs_unsubscribe(12).await.expect("logs_unsubscribe must succeed");
let _ = client.program_unsubscribe(13).await.expect("program_unsubscribe must succeed");
let _ = client.root_unsubscribe(14).await.expect("root_unsubscribe must succeed");
let _ = client.signature_unsubscribe(15).await.expect("signature_unsubscribe must succeed");
let _ = client.slot_unsubscribe(16).await.expect("slot_unsubscribe must succeed");
let _ = client
.slots_updates_unsubscribe(17)
.await
.expect("slots_updates_unsubscribe must succeed");
let _ = client.vote_unsubscribe(18).await.expect("vote_unsubscribe must succeed");
tokio::time::sleep(std::time::Duration::from_millis(150)).await;
let observed_methods = server.observed_methods_snapshot().await;
assert!(observed_methods.iter().any(|method| method == "accountUnsubscribe"));
assert!(observed_methods.iter().any(|method| method == "blockUnsubscribe"));
assert!(observed_methods.iter().any(|method| method == "logsUnsubscribe"));
assert!(observed_methods.iter().any(|method| method == "programUnsubscribe"));
assert!(observed_methods.iter().any(|method| method == "rootUnsubscribe"));
assert!(observed_methods.iter().any(|method| method == "signatureUnsubscribe"));
assert!(observed_methods.iter().any(|method| method == "slotUnsubscribe"));
assert!(observed_methods.iter().any(|method| method == "slotsUpdatesUnsubscribe"));
assert!(observed_methods.iter().any(|method| method == "voteUnsubscribe"));
client.disconnect().await.expect("disconnect must succeed");
server.shutdown().await;
}
}