1042 lines
39 KiB
Rust
1042 lines
39 KiB
Rust
// file: kb_app/src/demo_ws.rs
|
|
|
|
//! Demo WebSocket window commands and runtime state.
|
|
//!
|
|
//! This module isolates the manual WebSocket subscription test bench from the
|
|
//! main application window.
|
|
|
|
use tauri::Emitter;
|
|
use tauri::Manager;
|
|
|
|
/// Endpoint summary sent to the demo frontend.
|
|
#[derive(Clone, Debug, serde::Serialize)]
|
|
#[serde(rename_all = "camelCase")]
|
|
pub(crate) struct KbDemoWsEndpointSummary {
|
|
name: std::string::String,
|
|
resolved_url: std::string::String,
|
|
provider: std::string::String,
|
|
enabled: bool,
|
|
roles: std::vec::Vec<std::string::String>,
|
|
}
|
|
|
|
/// Current demo window runtime status.
|
|
#[derive(Clone, Debug, serde::Serialize)]
|
|
#[serde(rename_all = "camelCase")]
|
|
pub(crate) struct KbDemoWsStatusPayload {
|
|
connection_state: std::string::String,
|
|
endpoint_name: std::option::Option<std::string::String>,
|
|
endpoint_url: std::option::Option<std::string::String>,
|
|
current_subscription_id: std::option::Option<u64>,
|
|
current_subscribe_method: std::option::Option<std::string::String>,
|
|
current_unsubscribe_method: std::option::Option<std::string::String>,
|
|
current_notification_method: std::option::Option<std::string::String>,
|
|
event_count_total: u64,
|
|
notification_count_total: u64,
|
|
ui_log_count: u64,
|
|
suppressed_log_count: u64,
|
|
last_event_kind: std::option::Option<std::string::String>,
|
|
}
|
|
|
|
/// Subscribe request sent by the demo frontend.
|
|
#[derive(Clone, Debug, serde::Deserialize)]
|
|
#[serde(rename_all = "camelCase")]
|
|
pub(crate) struct KbDemoWsSubscribeRequest {
|
|
method: std::string::String,
|
|
mode: std::string::String,
|
|
target: std::option::Option<std::string::String>,
|
|
filter_json: std::option::Option<std::string::String>,
|
|
config_json: std::option::Option<std::string::String>,
|
|
}
|
|
|
|
/// Runtime state for the demo websocket window.
|
|
#[derive(Debug)]
|
|
pub(crate) struct KbDemoWsRuntimeState {
|
|
client: std::option::Option<kb_lib::WsClient>,
|
|
relay_task: std::option::Option<tauri::async_runtime::JoinHandle<()>>,
|
|
keepalive_task: std::option::Option<tauri::async_runtime::JoinHandle<()>>,
|
|
endpoint_name: std::option::Option<std::string::String>,
|
|
endpoint_url: std::option::Option<std::string::String>,
|
|
connection_state: kb_lib::KbConnectionState,
|
|
current_subscription: std::option::Option<kb_lib::WsSubscriptionInfo>,
|
|
event_count_total: u64,
|
|
notification_count_total: u64,
|
|
ui_log_count: u64,
|
|
suppressed_log_count: u64,
|
|
last_event_kind: std::option::Option<std::string::String>,
|
|
last_status_emit_at: std::option::Option<std::time::Instant>,
|
|
}
|
|
|
|
impl KbDemoWsRuntimeState {
|
|
/// Creates a new empty runtime state.
|
|
pub(crate) fn new() -> Self {
|
|
Self {
|
|
client: None,
|
|
relay_task: None,
|
|
keepalive_task: None,
|
|
endpoint_name: None,
|
|
endpoint_url: None,
|
|
connection_state: kb_lib::KbConnectionState::Disconnected,
|
|
current_subscription: None,
|
|
event_count_total: 0,
|
|
notification_count_total: 0,
|
|
ui_log_count: 0,
|
|
suppressed_log_count: 0,
|
|
last_event_kind: None,
|
|
last_status_emit_at: None,
|
|
}
|
|
}
|
|
|
|
fn to_status_payload(&self) -> KbDemoWsStatusPayload {
|
|
let current_subscription_id = self
|
|
.current_subscription
|
|
.as_ref()
|
|
.map(|subscription| subscription.subscription_id);
|
|
let current_subscribe_method = self
|
|
.current_subscription
|
|
.as_ref()
|
|
.map(|subscription| subscription.subscribe_method.clone());
|
|
let current_unsubscribe_method = self
|
|
.current_subscription
|
|
.as_ref()
|
|
.map(|subscription| subscription.unsubscribe_method.clone());
|
|
let current_notification_method = self
|
|
.current_subscription
|
|
.as_ref()
|
|
.map(|subscription| subscription.notification_method.clone());
|
|
KbDemoWsStatusPayload {
|
|
connection_state: kb_connection_state_to_string(self.connection_state),
|
|
endpoint_name: self.endpoint_name.clone(),
|
|
endpoint_url: self.endpoint_url.clone(),
|
|
current_subscription_id,
|
|
current_subscribe_method,
|
|
current_unsubscribe_method,
|
|
current_notification_method,
|
|
event_count_total: self.event_count_total,
|
|
notification_count_total: self.notification_count_total,
|
|
ui_log_count: self.ui_log_count,
|
|
suppressed_log_count: self.suppressed_log_count,
|
|
last_event_kind: self.last_event_kind.clone(),
|
|
}
|
|
}
|
|
|
|
fn clear(&mut self) {
|
|
self.client = None;
|
|
self.relay_task = None;
|
|
self.keepalive_task = None;
|
|
self.endpoint_name = None;
|
|
self.endpoint_url = None;
|
|
self.connection_state = kb_lib::KbConnectionState::Disconnected;
|
|
self.current_subscription = None;
|
|
self.event_count_total = 0;
|
|
self.notification_count_total = 0;
|
|
self.ui_log_count = 0;
|
|
self.suppressed_log_count = 0;
|
|
self.last_event_kind = None;
|
|
self.last_status_emit_at = None;
|
|
}
|
|
}
|
|
|
|
/// Shows and focuses the preconfigured `demo_ws` window.
|
|
#[tauri::command]
|
|
pub(crate) fn open_demo_ws_window(app_handle: tauri::AppHandle) -> Result<(), std::string::String> {
|
|
let existing_window_option = app_handle.get_webview_window("demo_ws");
|
|
let demo_window = match existing_window_option {
|
|
Some(demo_window) => demo_window,
|
|
None => {
|
|
let builder = tauri::WebviewWindowBuilder::new(
|
|
&app_handle,
|
|
"demo_ws",
|
|
tauri::WebviewUrl::App("demo_ws.html".into()),
|
|
)
|
|
.title("Demo Ws Subscribe")
|
|
.inner_size(1400.0, 768.0)
|
|
.min_inner_size(800.0, 600.0)
|
|
.center()
|
|
.visible(true)
|
|
.transparent(false)
|
|
.decorations(true);
|
|
let build_result = builder.build();
|
|
match build_result {
|
|
Ok(window) => window,
|
|
Err(error) => {
|
|
return Err(format!("cannot create demo_ws window: {error:?}"));
|
|
}
|
|
}
|
|
}
|
|
};
|
|
let show_result = demo_window.show();
|
|
if let Err(error) = show_result {
|
|
return Err(format!("cannot show demo_ws window: {error:?}"));
|
|
}
|
|
let focus_result = demo_window.set_focus();
|
|
if let Err(error) = focus_result {
|
|
return Err(format!("cannot focus demo_ws window: {error:?}"));
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Returns the list of configured websocket endpoints.
|
|
#[tauri::command]
|
|
pub(crate) async fn demo_ws_list_endpoints(
|
|
state: tauri::State<'_, crate::KbAppState>,
|
|
) -> Result<std::vec::Vec<KbDemoWsEndpointSummary>, std::string::String> {
|
|
let mut endpoints = std::vec::Vec::new();
|
|
for endpoint in &state.config.solana.ws_endpoints {
|
|
if !endpoint.enabled {
|
|
continue;
|
|
}
|
|
let resolved_url_result = endpoint.resolved_url();
|
|
let resolved_url = match resolved_url_result {
|
|
Ok(resolved_url) => resolved_url,
|
|
Err(error) => {
|
|
tracing::warn!(
|
|
endpoint_name = %endpoint.name,
|
|
"cannot resolve ws endpoint url from environment: {}",
|
|
error
|
|
);
|
|
format!("UNRESOLVED_ENV [{}] {}", endpoint.name, endpoint.url)
|
|
}
|
|
};
|
|
endpoints.push(KbDemoWsEndpointSummary {
|
|
name: endpoint.name.clone(),
|
|
resolved_url,
|
|
provider: endpoint.provider.clone(),
|
|
enabled: endpoint.enabled,
|
|
roles: endpoint.roles.clone(),
|
|
});
|
|
}
|
|
Ok(endpoints)
|
|
}
|
|
|
|
/// Returns the current demo websocket runtime status.
|
|
#[tauri::command]
|
|
pub(crate) async fn demo_ws_get_status(
|
|
state: tauri::State<'_, crate::KbAppState>,
|
|
) -> Result<KbDemoWsStatusPayload, std::string::String> {
|
|
let runtime_guard = state.demo_ws_runtime.lock().await;
|
|
Ok(runtime_guard.to_status_payload())
|
|
}
|
|
|
|
/// Connects the demo websocket runtime to the selected endpoint.
|
|
#[tauri::command]
|
|
pub(crate) async fn demo_ws_connect(
|
|
app_handle: tauri::AppHandle,
|
|
state: tauri::State<'_, crate::KbAppState>,
|
|
endpoint_name: std::string::String,
|
|
) -> Result<KbDemoWsStatusPayload, std::string::String> {
|
|
let endpoint_option = state.config.find_ws_endpoint(&endpoint_name);
|
|
let endpoint = match endpoint_option {
|
|
Some(endpoint) => endpoint.clone(),
|
|
None => {
|
|
return Err(format!("unknown websocket endpoint '{}'", endpoint_name));
|
|
}
|
|
};
|
|
let runtime_arc = state.demo_ws_runtime.clone();
|
|
{
|
|
let runtime_guard = runtime_arc.lock().await;
|
|
if runtime_guard.client.is_some() {
|
|
return Err("demo websocket client is already connected or connecting".to_string());
|
|
}
|
|
}
|
|
let client_result = kb_lib::WsClient::new(endpoint.clone());
|
|
let client = match client_result {
|
|
Ok(client) => client,
|
|
Err(error) => {
|
|
return Err(format!("cannot create websocket client: {error}"));
|
|
}
|
|
};
|
|
{
|
|
let mut runtime_guard = runtime_arc.lock().await;
|
|
runtime_guard.endpoint_name = Some(endpoint.name.clone());
|
|
runtime_guard.endpoint_url = Some(client.endpoint_url().to_string());
|
|
runtime_guard.connection_state = kb_lib::KbConnectionState::Connecting;
|
|
runtime_guard.current_subscription = None;
|
|
}
|
|
kb_emit_demo_ws_status(&app_handle, &runtime_arc).await;
|
|
kb_emit_demo_ws_log(
|
|
&app_handle,
|
|
&format!(
|
|
"[demo] connecting endpoint '{}' ({})",
|
|
endpoint.name,
|
|
client.endpoint_url()
|
|
),
|
|
);
|
|
let mut event_receiver = client.subscribe_events();
|
|
let relay_runtime = runtime_arc.clone();
|
|
let relay_app_handle = app_handle.clone();
|
|
let relay_task = tauri::async_runtime::spawn(async move {
|
|
loop {
|
|
let recv_result = event_receiver.recv().await;
|
|
match recv_result {
|
|
Ok(event) => {
|
|
let (emit_ui_log, emit_ui_status) =
|
|
kb_register_demo_ws_event_and_decide_emission(&relay_runtime, &event).await;
|
|
if emit_ui_log {
|
|
kb_emit_demo_ws_log(&relay_app_handle, &kb_format_demo_ws_event(&event));
|
|
}
|
|
if emit_ui_status {
|
|
kb_emit_demo_ws_status(&relay_app_handle, &relay_runtime).await;
|
|
}
|
|
}
|
|
Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
|
|
kb_emit_demo_ws_log(
|
|
&relay_app_handle,
|
|
&format!(
|
|
"[demo] event receiver lagged and skipped {} event(s)",
|
|
skipped
|
|
),
|
|
);
|
|
}
|
|
Err(tokio::sync::broadcast::error::RecvError::Closed) => {
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
});
|
|
let keepalive_client = client.clone();
|
|
let keepalive_app_handle = app_handle.clone();
|
|
let keepalive_task = tauri::async_runtime::spawn(async move {
|
|
kb_demo_ws_keepalive_loop(&keepalive_app_handle, &keepalive_client).await;
|
|
});
|
|
let connect_result = client.connect().await;
|
|
if let Err(error) = connect_result {
|
|
relay_task.abort();
|
|
keepalive_task.abort();
|
|
{
|
|
let mut runtime_guard = runtime_arc.lock().await;
|
|
runtime_guard.clear();
|
|
}
|
|
kb_emit_demo_ws_status(&app_handle, &runtime_arc).await;
|
|
return Err(format!("cannot connect websocket client: {error}"));
|
|
}
|
|
{
|
|
let mut runtime_guard = runtime_arc.lock().await;
|
|
runtime_guard.client = Some(client);
|
|
runtime_guard.relay_task = Some(relay_task);
|
|
runtime_guard.keepalive_task = Some(keepalive_task);
|
|
runtime_guard.endpoint_name = Some(endpoint.name.clone());
|
|
runtime_guard.endpoint_url = Some(endpoint.resolved_url().unwrap_or(endpoint.url));
|
|
runtime_guard.connection_state = kb_lib::KbConnectionState::Connected;
|
|
}
|
|
kb_emit_demo_ws_status(&app_handle, &runtime_arc).await;
|
|
let runtime_guard = runtime_arc.lock().await;
|
|
Ok(runtime_guard.to_status_payload())
|
|
}
|
|
|
|
/// Disconnects the demo websocket runtime.
|
|
#[tauri::command]
|
|
pub(crate) async fn demo_ws_disconnect(
|
|
app_handle: tauri::AppHandle,
|
|
state: tauri::State<'_, crate::KbAppState>,
|
|
) -> Result<KbDemoWsStatusPayload, std::string::String> {
|
|
let runtime_arc = state.demo_ws_runtime.clone();
|
|
{
|
|
let mut runtime_guard = runtime_arc.lock().await;
|
|
runtime_guard.connection_state = kb_lib::KbConnectionState::Disconnecting;
|
|
}
|
|
kb_emit_demo_ws_status(&app_handle, &runtime_arc).await;
|
|
let (client_option, relay_task_option, keepalive_task_option) = {
|
|
let mut runtime_guard = runtime_arc.lock().await;
|
|
(
|
|
runtime_guard.client.take(),
|
|
runtime_guard.relay_task.take(),
|
|
runtime_guard.keepalive_task.take(),
|
|
)
|
|
};
|
|
if let Some(keepalive_task) = keepalive_task_option {
|
|
keepalive_task.abort();
|
|
}
|
|
if let Some(client) = &client_option {
|
|
let disconnect_result = client.disconnect().await;
|
|
if let Err(error) = disconnect_result {
|
|
kb_emit_demo_ws_log(&app_handle, &format!("[demo] disconnect error: {}", error));
|
|
}
|
|
}
|
|
if let Some(relay_task) = relay_task_option {
|
|
relay_task.abort();
|
|
}
|
|
{
|
|
let mut runtime_guard = runtime_arc.lock().await;
|
|
runtime_guard.clear();
|
|
}
|
|
kb_emit_demo_ws_status(&app_handle, &runtime_arc).await;
|
|
let runtime_guard = runtime_arc.lock().await;
|
|
Ok(runtime_guard.to_status_payload())
|
|
}
|
|
|
|
/// Sends one demo subscription request.
|
|
#[tauri::command]
|
|
pub(crate) async fn demo_ws_subscribe(
|
|
state: tauri::State<'_, crate::KbAppState>,
|
|
request: KbDemoWsSubscribeRequest,
|
|
) -> Result<u64, std::string::String> {
|
|
let client_option = {
|
|
let runtime_guard = state.demo_ws_runtime.lock().await;
|
|
if runtime_guard.current_subscription.is_some() {
|
|
return Err("a subscription is already active, unsubscribe it first".to_string());
|
|
}
|
|
runtime_guard.client.clone()
|
|
};
|
|
let client = match client_option {
|
|
Some(client) => client,
|
|
None => {
|
|
return Err("demo websocket client is not connected".to_string());
|
|
}
|
|
};
|
|
kb_execute_demo_ws_subscribe(&client, &request).await
|
|
}
|
|
|
|
/// Sends one unsubscribe request for the current active subscription.
|
|
#[tauri::command]
|
|
pub(crate) async fn demo_ws_unsubscribe_current(
|
|
state: tauri::State<'_, crate::KbAppState>,
|
|
) -> Result<u64, std::string::String> {
|
|
let (client_option, subscription_option) = {
|
|
let runtime_guard = state.demo_ws_runtime.lock().await;
|
|
(
|
|
runtime_guard.client.clone(),
|
|
runtime_guard.current_subscription.clone(),
|
|
)
|
|
};
|
|
let client = match client_option {
|
|
Some(client) => client,
|
|
None => {
|
|
return Err("demo websocket client is not connected".to_string());
|
|
}
|
|
};
|
|
let subscription = match subscription_option {
|
|
Some(subscription) => subscription,
|
|
None => {
|
|
return Err("no active subscription is currently registered".to_string());
|
|
}
|
|
};
|
|
let params = vec![serde_json::Value::from(subscription.subscription_id)];
|
|
let send_result = client
|
|
.send_json_rpc_request(subscription.unsubscribe_method.clone(), params)
|
|
.await;
|
|
match send_result {
|
|
Ok(request_id) => Ok(request_id),
|
|
Err(error) => Err(format!("cannot send unsubscribe request: {error}")),
|
|
}
|
|
}
|
|
|
|
async fn kb_execute_demo_ws_subscribe(
|
|
client: &kb_lib::WsClient,
|
|
request: &KbDemoWsSubscribeRequest,
|
|
) -> Result<u64, std::string::String> {
|
|
let method = request.method.trim();
|
|
let mode = request.mode.trim();
|
|
if method == "account" {
|
|
let target_result = kb_required_target(request, "account pubkey");
|
|
let target = match target_result {
|
|
Ok(target) => target,
|
|
Err(error) => return Err(error),
|
|
};
|
|
if mode == "typed" {
|
|
let config_result = kb_parse_optional_json_typed::<
|
|
solana_rpc_client_api::config::RpcAccountInfoConfig,
|
|
>(&request.config_json, "account typed config");
|
|
let config = match config_result {
|
|
Ok(config) => config,
|
|
Err(error) => return Err(error),
|
|
};
|
|
let result = client.account_subscribe_typed(target, config).await;
|
|
return result.map_err(|error| format!("account typed subscribe failed: {error}"));
|
|
}
|
|
let config_result =
|
|
kb_parse_optional_json_value(&request.config_json, "account raw config");
|
|
let config = match config_result {
|
|
Ok(config) => config,
|
|
Err(error) => return Err(error),
|
|
};
|
|
let result = client.account_subscribe_raw(target, config).await;
|
|
return result.map_err(|error| format!("account raw subscribe failed: {error}"));
|
|
}
|
|
if method == "block" {
|
|
if mode == "typed" {
|
|
let filter_result = kb_parse_required_json_typed::<
|
|
solana_rpc_client_api::config::RpcBlockSubscribeFilter,
|
|
>(&request.filter_json, "block typed filter");
|
|
let filter = match filter_result {
|
|
Ok(filter) => filter,
|
|
Err(error) => return Err(error),
|
|
};
|
|
let config_result = kb_parse_optional_json_typed::<
|
|
solana_rpc_client_api::config::RpcBlockSubscribeConfig,
|
|
>(&request.config_json, "block typed config");
|
|
let config = match config_result {
|
|
Ok(config) => config,
|
|
Err(error) => return Err(error),
|
|
};
|
|
let result = client.block_subscribe_typed(filter, config).await;
|
|
return result.map_err(|error| format!("block typed subscribe failed: {error}"));
|
|
}
|
|
let filter_result =
|
|
kb_parse_required_json_value(&request.filter_json, "block raw filter");
|
|
let filter = match filter_result {
|
|
Ok(filter) => filter,
|
|
Err(error) => return Err(error),
|
|
};
|
|
let config_result =
|
|
kb_parse_optional_json_value(&request.config_json, "block raw config");
|
|
let config = match config_result {
|
|
Ok(config) => config,
|
|
Err(error) => return Err(error),
|
|
};
|
|
let result = client.block_subscribe_raw(filter, config).await;
|
|
return result.map_err(|error| format!("block raw subscribe failed: {error}"));
|
|
}
|
|
if method == "logs" {
|
|
if mode == "typed" {
|
|
let filter_result = kb_parse_required_json_typed::<
|
|
solana_rpc_client_api::config::RpcTransactionLogsFilter,
|
|
>(&request.filter_json, "logs typed filter");
|
|
let filter = match filter_result {
|
|
Ok(filter) => filter,
|
|
Err(error) => return Err(error),
|
|
};
|
|
let config_result = kb_parse_optional_json_typed::<
|
|
solana_rpc_client_api::config::RpcTransactionLogsConfig,
|
|
>(&request.config_json, "logs typed config");
|
|
let config = match config_result {
|
|
Ok(config) => config,
|
|
Err(error) => return Err(error),
|
|
};
|
|
let result = client.logs_subscribe_typed(filter, config).await;
|
|
return result.map_err(|error| format!("logs typed subscribe failed: {error}"));
|
|
}
|
|
let filter_result =
|
|
kb_parse_required_json_value(&request.filter_json, "logs raw filter");
|
|
let filter = match filter_result {
|
|
Ok(filter) => filter,
|
|
Err(error) => return Err(error),
|
|
};
|
|
let config_result =
|
|
kb_parse_optional_json_value(&request.config_json, "logs raw config");
|
|
let config = match config_result {
|
|
Ok(config) => config,
|
|
Err(error) => return Err(error),
|
|
};
|
|
let result = client.logs_subscribe_raw(filter, config).await;
|
|
return result.map_err(|error| format!("logs raw subscribe failed: {error}"));
|
|
}
|
|
if method == "program" {
|
|
let target_result = kb_required_target(request, "program id");
|
|
let target = match target_result {
|
|
Ok(target) => target,
|
|
Err(error) => return Err(error),
|
|
};
|
|
if mode == "typed" {
|
|
let config_result = kb_parse_optional_json_typed::<
|
|
solana_rpc_client_api::config::RpcProgramAccountsConfig,
|
|
>(&request.config_json, "program typed config");
|
|
let config = match config_result {
|
|
Ok(config) => config,
|
|
Err(error) => return Err(error),
|
|
};
|
|
let result = client.program_subscribe_typed(target, config).await;
|
|
return result.map_err(|error| format!("program typed subscribe failed: {error}"));
|
|
}
|
|
let config_result =
|
|
kb_parse_optional_json_value(&request.config_json, "program raw config");
|
|
let config = match config_result {
|
|
Ok(config) => config,
|
|
Err(error) => return Err(error),
|
|
};
|
|
let result = client.program_subscribe_raw(target, config).await;
|
|
return result.map_err(|error| format!("program raw subscribe failed: {error}"));
|
|
}
|
|
if method == "root" {
|
|
let result = client.root_subscribe().await;
|
|
return result.map_err(|error| format!("root subscribe failed: {error}"));
|
|
}
|
|
if method == "signature" {
|
|
let target_result = kb_required_target(request, "signature");
|
|
let target = match target_result {
|
|
Ok(target) => target,
|
|
Err(error) => return Err(error),
|
|
};
|
|
if mode == "typed" {
|
|
let config_result = kb_parse_optional_json_typed::<
|
|
solana_rpc_client_api::config::RpcSignatureSubscribeConfig,
|
|
>(&request.config_json, "signature typed config");
|
|
let config = match config_result {
|
|
Ok(config) => config,
|
|
Err(error) => return Err(error),
|
|
};
|
|
let result = client.signature_subscribe_typed(target, config).await;
|
|
return result.map_err(|error| format!("signature typed subscribe failed: {error}"));
|
|
}
|
|
let config_result =
|
|
kb_parse_optional_json_value(&request.config_json, "signature raw config");
|
|
let config = match config_result {
|
|
Ok(config) => config,
|
|
Err(error) => return Err(error),
|
|
};
|
|
let result = client.signature_subscribe_raw(target, config).await;
|
|
return result.map_err(|error| format!("signature raw subscribe failed: {error}"));
|
|
}
|
|
if method == "slot" {
|
|
let result = client.slot_subscribe().await;
|
|
return result.map_err(|error| format!("slot subscribe failed: {error}"));
|
|
}
|
|
if method == "slotsUpdates" {
|
|
let result = client.slots_updates_subscribe().await;
|
|
return result.map_err(|error| format!("slotsUpdates subscribe failed: {error}"));
|
|
}
|
|
if method == "vote" {
|
|
let result = client.vote_subscribe().await;
|
|
return result.map_err(|error| format!("vote subscribe failed: {error}"));
|
|
}
|
|
Err(format!("unsupported demo subscribe method '{}'", method))
|
|
}
|
|
|
|
fn kb_required_target(
|
|
request: &KbDemoWsSubscribeRequest,
|
|
label: &str,
|
|
) -> Result<std::string::String, std::string::String> {
|
|
let target_option = request.target.as_ref();
|
|
let target = match target_option {
|
|
Some(target) => target.trim(),
|
|
None => {
|
|
return Err(format!("{} is required", label));
|
|
}
|
|
};
|
|
if target.is_empty() {
|
|
return Err(format!("{} is required", label));
|
|
}
|
|
Ok(target.to_string())
|
|
}
|
|
|
|
fn kb_parse_optional_json_value(
|
|
input: &std::option::Option<std::string::String>,
|
|
label: &str,
|
|
) -> Result<std::option::Option<serde_json::Value>, std::string::String> {
|
|
match input {
|
|
Some(input) => {
|
|
if input.trim().is_empty() {
|
|
return Ok(None);
|
|
}
|
|
let parse_result = serde_json::from_str::<serde_json::Value>(input);
|
|
match parse_result {
|
|
Ok(value) => Ok(Some(value)),
|
|
Err(error) => Err(format!("cannot parse {}: {}", label, error)),
|
|
}
|
|
}
|
|
None => Ok(None),
|
|
}
|
|
}
|
|
|
|
fn kb_parse_required_json_value(
|
|
input: &std::option::Option<std::string::String>,
|
|
label: &str,
|
|
) -> Result<serde_json::Value, std::string::String> {
|
|
let input_option = input.as_ref();
|
|
let input = match input_option {
|
|
Some(input) => input.trim(),
|
|
None => {
|
|
return Err(format!("{} is required", label));
|
|
}
|
|
};
|
|
if input.is_empty() {
|
|
return Err(format!("{} is required", label));
|
|
}
|
|
let parse_result = serde_json::from_str::<serde_json::Value>(input);
|
|
match parse_result {
|
|
Ok(value) => Ok(value),
|
|
Err(error) => Err(format!("cannot parse {}: {}", label, error)),
|
|
}
|
|
}
|
|
|
|
fn kb_parse_optional_json_typed<T>(
|
|
input: &std::option::Option<std::string::String>,
|
|
label: &str,
|
|
) -> Result<std::option::Option<T>, std::string::String>
|
|
where
|
|
T: serde::de::DeserializeOwned,
|
|
{
|
|
match input {
|
|
Some(input) => {
|
|
if input.trim().is_empty() {
|
|
return Ok(None);
|
|
}
|
|
let parse_result = serde_json::from_str::<T>(input);
|
|
match parse_result {
|
|
Ok(value) => Ok(Some(value)),
|
|
Err(error) => Err(format!("cannot parse {}: {}", label, error)),
|
|
}
|
|
}
|
|
None => Ok(None),
|
|
}
|
|
}
|
|
|
|
fn kb_parse_required_json_typed<T>(
|
|
input: &std::option::Option<std::string::String>,
|
|
label: &str,
|
|
) -> Result<T, std::string::String>
|
|
where
|
|
T: serde::de::DeserializeOwned,
|
|
{
|
|
let input_option = input.as_ref();
|
|
let input = match input_option {
|
|
Some(input) => input.trim(),
|
|
None => {
|
|
return Err(format!("{} is required", label));
|
|
}
|
|
};
|
|
if input.is_empty() {
|
|
return Err(format!("{} is required", label));
|
|
}
|
|
let parse_result = serde_json::from_str::<T>(input);
|
|
match parse_result {
|
|
Ok(value) => Ok(value),
|
|
Err(error) => Err(format!("cannot parse {}: {}", label, error)),
|
|
}
|
|
}
|
|
|
|
async fn kb_register_demo_ws_event_and_decide_emission(
|
|
runtime_arc: &std::sync::Arc<tokio::sync::Mutex<KbDemoWsRuntimeState>>,
|
|
event: &kb_lib::WsEvent,
|
|
) -> (bool, bool) {
|
|
let mut runtime_guard = runtime_arc.lock().await;
|
|
runtime_guard.event_count_total = runtime_guard.event_count_total.saturating_add(1);
|
|
runtime_guard.last_event_kind = Some(kb_demo_ws_event_kind_name(event).to_string());
|
|
let mut emit_ui_log = true;
|
|
let force_status_emit = matches!(
|
|
event,
|
|
kb_lib::WsEvent::Connected { .. }
|
|
| kb_lib::WsEvent::Disconnected { .. }
|
|
| kb_lib::WsEvent::SubscriptionRegistered { .. }
|
|
| kb_lib::WsEvent::SubscriptionUnregistered { .. }
|
|
| kb_lib::WsEvent::Error { .. }
|
|
);
|
|
match event {
|
|
kb_lib::WsEvent::Connected {
|
|
endpoint_name,
|
|
endpoint_url,
|
|
} => {
|
|
runtime_guard.connection_state = kb_lib::KbConnectionState::Connected;
|
|
runtime_guard.endpoint_name = Some(endpoint_name.clone());
|
|
runtime_guard.endpoint_url = Some(endpoint_url.clone());
|
|
}
|
|
kb_lib::WsEvent::SubscriptionRegistered { subscription, .. } => {
|
|
runtime_guard.current_subscription = Some(subscription.clone());
|
|
runtime_guard.notification_count_total = 0;
|
|
}
|
|
kb_lib::WsEvent::SubscriptionNotification { subscription, .. } => {
|
|
runtime_guard.notification_count_total =
|
|
runtime_guard.notification_count_total.saturating_add(1);
|
|
let subscribe_method = subscription.subscribe_method.as_str();
|
|
let notif_count = runtime_guard.notification_count_total;
|
|
if subscribe_method == "logsSubscribe" || subscribe_method == "programSubscribe" {
|
|
emit_ui_log = notif_count % 100 == 1;
|
|
} else if subscribe_method == "slotsUpdatesSubscribe" {
|
|
emit_ui_log = notif_count % 20 == 1;
|
|
}
|
|
}
|
|
kb_lib::WsEvent::TextMessage { .. } | kb_lib::WsEvent::JsonRpcMessage { .. } => {
|
|
let subscribe_method_option = runtime_guard
|
|
.current_subscription
|
|
.as_ref()
|
|
.map(|subscription| subscription.subscribe_method.as_str());
|
|
if let Some(subscribe_method) = subscribe_method_option {
|
|
if subscribe_method == "logsSubscribe"
|
|
|| subscribe_method == "programSubscribe"
|
|
|| subscribe_method == "slotsUpdatesSubscribe"
|
|
{
|
|
emit_ui_log = false;
|
|
}
|
|
}
|
|
}
|
|
kb_lib::WsEvent::Pong { .. } => {
|
|
emit_ui_log = false;
|
|
}
|
|
kb_lib::WsEvent::SubscriptionUnregistered {
|
|
subscription_id, ..
|
|
} => {
|
|
let current_subscription_id = runtime_guard
|
|
.current_subscription
|
|
.as_ref()
|
|
.map(|subscription| subscription.subscription_id);
|
|
if current_subscription_id == Some(*subscription_id) {
|
|
runtime_guard.current_subscription = None;
|
|
runtime_guard.notification_count_total = 0;
|
|
}
|
|
}
|
|
kb_lib::WsEvent::Disconnected { .. } => {
|
|
runtime_guard.client = None;
|
|
runtime_guard.relay_task = None;
|
|
runtime_guard.keepalive_task = None;
|
|
runtime_guard.connection_state = kb_lib::KbConnectionState::Disconnected;
|
|
runtime_guard.current_subscription = None;
|
|
runtime_guard.notification_count_total = 0;
|
|
}
|
|
_ => {}
|
|
}
|
|
if emit_ui_log {
|
|
runtime_guard.ui_log_count = runtime_guard.ui_log_count.saturating_add(1);
|
|
} else {
|
|
runtime_guard.suppressed_log_count = runtime_guard.suppressed_log_count.saturating_add(1);
|
|
}
|
|
let now = std::time::Instant::now();
|
|
let emit_ui_status = if force_status_emit {
|
|
true
|
|
} else {
|
|
match runtime_guard.last_status_emit_at {
|
|
Some(last_status_emit_at) => {
|
|
now.duration_since(last_status_emit_at) >= std::time::Duration::from_millis(250)
|
|
}
|
|
None => true,
|
|
}
|
|
};
|
|
if emit_ui_status {
|
|
runtime_guard.last_status_emit_at = Some(now);
|
|
}
|
|
(emit_ui_log, emit_ui_status)
|
|
}
|
|
|
|
async fn kb_emit_demo_ws_status(
|
|
app_handle: &tauri::AppHandle,
|
|
runtime_arc: &std::sync::Arc<tokio::sync::Mutex<KbDemoWsRuntimeState>>,
|
|
) {
|
|
let status_payload = {
|
|
let runtime_guard = runtime_arc.lock().await;
|
|
runtime_guard.to_status_payload()
|
|
};
|
|
let demo_window_option = app_handle.get_webview_window("demo_ws");
|
|
let demo_window = match demo_window_option {
|
|
Some(demo_window) => demo_window,
|
|
None => {
|
|
return;
|
|
}
|
|
};
|
|
let emit_result = demo_window.emit("demo-ws-status", status_payload);
|
|
if let Err(error) = emit_result {
|
|
tracing::error!("error emitting demo-ws-status: {error:?}");
|
|
}
|
|
}
|
|
|
|
fn kb_emit_demo_ws_log(app_handle: &tauri::AppHandle, line: &str) {
|
|
tracing::debug!("{}", line);
|
|
let demo_window_option = app_handle.get_webview_window("demo_ws");
|
|
let demo_window = match demo_window_option {
|
|
Some(demo_window) => demo_window,
|
|
None => {
|
|
return;
|
|
}
|
|
};
|
|
let emit_result = demo_window.emit("demo-ws-log", line.to_string());
|
|
if let Err(error) = emit_result {
|
|
tracing::error!("error emitting demo-ws-log: {error:?}");
|
|
}
|
|
}
|
|
|
|
fn kb_connection_state_to_string(state: kb_lib::KbConnectionState) -> std::string::String {
|
|
match state {
|
|
kb_lib::KbConnectionState::Disconnected => "Disconnected".to_string(),
|
|
kb_lib::KbConnectionState::Connecting => "Connecting".to_string(),
|
|
kb_lib::KbConnectionState::Connected => "Connected".to_string(),
|
|
kb_lib::KbConnectionState::Disconnecting => "Disconnecting".to_string(),
|
|
}
|
|
}
|
|
|
|
fn kb_format_demo_ws_event(event: &kb_lib::WsEvent) -> std::string::String {
|
|
match event {
|
|
kb_lib::WsEvent::Connected {
|
|
endpoint_name,
|
|
endpoint_url,
|
|
} => {
|
|
format!("[demo:{endpoint_name}] connected to {endpoint_url}")
|
|
}
|
|
kb_lib::WsEvent::TextMessage {
|
|
endpoint_name,
|
|
text,
|
|
} => {
|
|
format!(
|
|
"[demo:{endpoint_name}] text: {}",
|
|
kb_shorten_log_text(text, 1200)
|
|
)
|
|
}
|
|
kb_lib::WsEvent::JsonRpcMessage {
|
|
endpoint_name,
|
|
message,
|
|
} => {
|
|
let rendered = format!("{message:?}");
|
|
format!(
|
|
"[demo:{endpoint_name}] json-rpc: {}",
|
|
kb_shorten_log_text(&rendered, 1800)
|
|
)
|
|
}
|
|
kb_lib::WsEvent::JsonRpcParseError {
|
|
endpoint_name,
|
|
text,
|
|
error,
|
|
} => {
|
|
format!(
|
|
"[demo:{endpoint_name}] json-rpc parse error: {} | raw={}",
|
|
error,
|
|
kb_shorten_log_text(text, 1200)
|
|
)
|
|
}
|
|
kb_lib::WsEvent::SubscriptionRegistered {
|
|
endpoint_name,
|
|
subscription,
|
|
} => {
|
|
format!(
|
|
"[demo:{endpoint_name}] subscription registered request_id={} subscription_id={} subscribe={} unsubscribe={} notification={}",
|
|
subscription.request_id,
|
|
subscription.subscription_id,
|
|
subscription.subscribe_method,
|
|
subscription.unsubscribe_method,
|
|
subscription.notification_method
|
|
)
|
|
}
|
|
kb_lib::WsEvent::SubscriptionNotification {
|
|
endpoint_name,
|
|
subscription,
|
|
notification,
|
|
method_matches_registry,
|
|
} => {
|
|
let result_text = notification.params.result.to_string();
|
|
let typed_suffix = match kb_lib::parse_kb_solana_ws_typed_notification(notification) {
|
|
Ok(typed_notification) => {
|
|
let rendered = format!("{typed_notification:?}");
|
|
format!(" | typed={}", kb_shorten_log_text(&rendered, 1200))
|
|
}
|
|
Err(_) => std::string::String::new(),
|
|
};
|
|
format!(
|
|
"[demo:{endpoint_name}] tracked notification subscription_id={} method={} expected={} matches={} result={}{}",
|
|
subscription.subscription_id,
|
|
notification.method,
|
|
subscription.notification_method,
|
|
method_matches_registry,
|
|
kb_shorten_log_text(&result_text, 1600),
|
|
typed_suffix
|
|
)
|
|
}
|
|
kb_lib::WsEvent::JsonRpcNotificationWithoutSubscription {
|
|
endpoint_name,
|
|
notification,
|
|
} => {
|
|
let result_text = notification.params.result.to_string();
|
|
let typed_suffix = match kb_lib::parse_kb_solana_ws_typed_notification(notification) {
|
|
Ok(typed_notification) => {
|
|
let rendered = format!("{typed_notification:?}");
|
|
format!(" | typed={}", kb_shorten_log_text(&rendered, 1200))
|
|
}
|
|
Err(_) => std::string::String::new(),
|
|
};
|
|
format!(
|
|
"[demo:{endpoint_name}] untracked notification method={} subscription={} result={}{}",
|
|
notification.method,
|
|
notification.params.subscription,
|
|
kb_shorten_log_text(&result_text, 1600),
|
|
typed_suffix
|
|
)
|
|
}
|
|
kb_lib::WsEvent::SubscriptionUnregistered {
|
|
endpoint_name,
|
|
subscription_id,
|
|
unsubscribe_method,
|
|
was_active,
|
|
} => {
|
|
format!(
|
|
"[demo:{endpoint_name}] subscription unregistered subscription_id={} unsubscribe_method={} was_active={}",
|
|
subscription_id, unsubscribe_method, was_active
|
|
)
|
|
}
|
|
kb_lib::WsEvent::BinaryMessage {
|
|
endpoint_name,
|
|
data,
|
|
} => {
|
|
format!(
|
|
"[demo:{endpoint_name}] binary message ({} bytes)",
|
|
data.len()
|
|
)
|
|
}
|
|
kb_lib::WsEvent::Ping {
|
|
endpoint_name,
|
|
data,
|
|
} => {
|
|
format!("[demo:{endpoint_name}] ping ({} bytes)", data.len())
|
|
}
|
|
kb_lib::WsEvent::Pong {
|
|
endpoint_name,
|
|
data,
|
|
} => {
|
|
format!("[demo:{endpoint_name}] pong ({} bytes)", data.len())
|
|
}
|
|
kb_lib::WsEvent::CloseReceived {
|
|
endpoint_name,
|
|
code,
|
|
reason,
|
|
} => {
|
|
format!(
|
|
"[demo:{endpoint_name}] close received code={:?} reason={:?}",
|
|
code, reason
|
|
)
|
|
}
|
|
kb_lib::WsEvent::Disconnected { endpoint_name } => {
|
|
format!("[demo:{endpoint_name}] disconnected")
|
|
}
|
|
kb_lib::WsEvent::Error {
|
|
endpoint_name,
|
|
error,
|
|
} => {
|
|
format!("[demo:{endpoint_name}] error: {error}")
|
|
}
|
|
}
|
|
}
|
|
|
|
fn kb_shorten_log_text(input: &str, max_chars: usize) -> std::string::String {
|
|
let char_count = input.chars().count();
|
|
if char_count <= max_chars {
|
|
return input.to_string();
|
|
}
|
|
|
|
let shortened: std::string::String = input.chars().take(max_chars).collect();
|
|
format!("{shortened} …[truncated {} chars]", char_count - max_chars)
|
|
}
|
|
|
|
async fn kb_demo_ws_keepalive_loop(app_handle: &tauri::AppHandle, client: &kb_lib::WsClient) {
|
|
loop {
|
|
tokio::time::sleep(std::time::Duration::from_secs(30)).await;
|
|
let state = client.connection_state().await;
|
|
if state != kb_lib::KbConnectionState::Connected {
|
|
break;
|
|
}
|
|
let send_result = client.send_ping(b"demo-keepalive".to_vec()).await;
|
|
if let Err(error) = send_result {
|
|
kb_emit_demo_ws_log(
|
|
app_handle,
|
|
&format!(
|
|
"[demo:{}] keepalive ping failed: {}",
|
|
client.endpoint_name(),
|
|
error
|
|
),
|
|
);
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
fn kb_demo_ws_event_kind_name(event: &kb_lib::WsEvent) -> &'static str {
|
|
match event {
|
|
kb_lib::WsEvent::Connected { .. } => "connected",
|
|
kb_lib::WsEvent::TextMessage { .. } => "text_message",
|
|
kb_lib::WsEvent::JsonRpcMessage { .. } => "json_rpc_message",
|
|
kb_lib::WsEvent::JsonRpcParseError { .. } => "json_rpc_parse_error",
|
|
kb_lib::WsEvent::SubscriptionRegistered { .. } => "subscription_registered",
|
|
kb_lib::WsEvent::SubscriptionNotification { .. } => "subscription_notification",
|
|
kb_lib::WsEvent::JsonRpcNotificationWithoutSubscription { .. } => "untracked_notification",
|
|
kb_lib::WsEvent::SubscriptionUnregistered { .. } => "subscription_unregistered",
|
|
kb_lib::WsEvent::BinaryMessage { .. } => "binary_message",
|
|
kb_lib::WsEvent::Ping { .. } => "ping",
|
|
kb_lib::WsEvent::Pong { .. } => "pong",
|
|
kb_lib::WsEvent::CloseReceived { .. } => "close_received",
|
|
kb_lib::WsEvent::Disconnected { .. } => "disconnected",
|
|
kb_lib::WsEvent::Error { .. } => "error",
|
|
}
|
|
}
|