/*
* Prediction Trading Streams
*
* Binance Web3 Prediction pushes 14 kinds of real-time events over the Binance SApi WebSocket channel, covering market and limit trading, reward claims, fund transfers and market resolution, plus order book (bid/ask) quotes per market. Trading events go only to the user they belong to; order book quotes go to everyone subscribed. A message may occasionally be missed, order is not guaranteed, and nothing is stored while you are offline — after reconnecting, fetch the latest state from the REST API and drop duplicates using `pushId` (business events) or `marketId` + `updateTimestampMs` (order book).
*
* The version of the OpenAPI document: 1.0.0
*
*
* NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
* https://openapi-generator.tech
* Do not edit the class manually.
*/
#![allow(unused_imports)]
use async_trait::async_trait;
use derive_builder::Builder;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::{collections::HashMap, sync::Arc};
use crate::common::{
models::ParamBuildError,
utils::replace_websocket_streams_placeholders,
websocket::{WebsocketBase, WebsocketStream, WebsocketStreams, create_stream_handler},
};
use crate::models::StreamId;
use crate::w3w_prediction::websocket_streams::models;
#[async_trait]
pub trait Api: Send + Sync {
async fn batch_claim_partially_succeeded(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::BatchClaimPartiallySucceededResponse>>>;
async fn fund_transfer_failed(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::FundTransferFailedResponse>>>;
async fn fund_transfer_succeeded(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::FundTransferSucceededResponse>>>;
async fn limit_order_fully_filled(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::LimitOrderFullyFilledResponse>>>;
async fn limit_order_partially_filled(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::LimitOrderPartiallyFilledResponse>>>;
async fn limit_order_submission_failed(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::LimitOrderSubmissionFailedResponse>>>;
async fn limit_order_submitted_successfully(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::LimitOrderSubmittedSuccessfullyResponse>>>;
async fn market_buy_order_failed(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::MarketBuyOrderFailedResponse>>>;
async fn market_buy_order_fully_filled(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::MarketBuyOrderFullyFilledResponse>>>;
async fn market_resolution_notification(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::MarketResolutionNotificationResponse>>>;
async fn market_sell_order_failed_expired(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::MarketSellOrderFailedExpiredResponse>>>;
async fn market_sell_order_fully_filled(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::MarketSellOrderFullyFilledResponse>>>;
async fn prediction_order_book_by_market_id(
&self,
params: PredictionOrderBookByMarketIdParams,
) -> anyhow::Result<Arc<WebsocketStream<models::PredictionOrderBookByMarketIdResponse>>>;
async fn reward_claim_failed(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::RewardClaimFailedResponse>>>;
async fn reward_claim_succeeded(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::RewardClaimSucceededResponse>>>;
}
pub struct ApiClient {
websocket_streams_base: Arc<WebsocketStreams>,
}
impl ApiClient {
pub fn new(websocket_streams_base: Arc<WebsocketStreams>) -> Self {
Self {
websocket_streams_base,
}
}
}
/// Request parameters for the [`prediction_order_book_by_market_id`] operation.
///
/// This struct holds all of the inputs you can pass when calling
/// [`prediction_order_book_by_market_id`](#method.prediction_order_book_by_market_id).
#[derive(Clone, Debug, Builder, Deserialize)]
#[builder(pattern = "owned", build_fn(error = "ParamBuildError"))]
pub struct PredictionOrderBookByMarketIdParams {
/// Market ID to subscribe to — the `marketId` inside `markets[]` returned by `market/list`
/// or `market/detail`. It is **not** `marketTopicId` and **not** `externalId`. A wrong id is
/// never rejected: subscribing with `marketTopicId`, `externalId`, or with a closed or
/// resolved market returns the same successful `COMMAND` reply as a correct id and then
/// pushes nothing at all, so a stream that stays silent is far more likely a wrong id than
/// a broken connection.
///
/// This field is **required.
#[builder(setter(into))]
#[serde(rename = "marketId")]
pub market_id: String,
}
impl PredictionOrderBookByMarketIdParams {
/// Create a builder for [`prediction_order_book_by_market_id`].
///
/// Required parameters:
///
/// * `market_id` — Market ID to subscribe to — the `marketId` inside `markets[]` returned by `market/list` or `market/detail`. It is **not** `marketTopicId` and **not** `externalId`. A wrong id is never rejected: subscribing with `marketTopicId`, `externalId`, or with a closed or resolved market returns the same successful `COMMAND` reply as a correct id and then pushes nothing at all, so a stream that stays silent is far more likely a wrong id than a broken connection.
///
#[must_use]
pub fn builder(market_id: String) -> PredictionOrderBookByMarketIdParamsBuilder {
PredictionOrderBookByMarketIdParamsBuilder::default().market_id(market_id)
}
}
#[async_trait]
impl Api for ApiClient {
async fn batch_claim_partially_succeeded(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::BatchClaimPartiallySucceededResponse>>> {
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);
let stream = replace_websocket_streams_placeholders(
"/web3_prediction_pm_claim_partial_success",
&vars,
);
Ok(
create_stream_handler::<models::BatchClaimPartiallySucceededResponse>(
WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
stream,
id_opt.map(|s| {
if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
if let Ok(n) = s.parse::<u32>() {
return StreamId::Number(n);
}
}
StreamId::Str(s)
}),
None,
)
.await,
)
}
async fn fund_transfer_failed(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::FundTransferFailedResponse>>> {
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);
let stream =
replace_websocket_streams_placeholders("/web3_prediction_pm_transfer_fail", &vars);
Ok(create_stream_handler::<models::FundTransferFailedResponse>(
WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
stream,
id_opt.map(|s| {
if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
if let Ok(n) = s.parse::<u32>() {
return StreamId::Number(n);
}
}
StreamId::Str(s)
}),
None,
)
.await)
}
async fn fund_transfer_succeeded(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::FundTransferSucceededResponse>>> {
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);
let stream =
replace_websocket_streams_placeholders("/web3_prediction_pm_transfer_success", &vars);
Ok(
create_stream_handler::<models::FundTransferSucceededResponse>(
WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
stream,
id_opt.map(|s| {
if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
if let Ok(n) = s.parse::<u32>() {
return StreamId::Number(n);
}
}
StreamId::Str(s)
}),
None,
)
.await,
)
}
async fn limit_order_fully_filled(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::LimitOrderFullyFilledResponse>>> {
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);
let stream =
replace_websocket_streams_placeholders("/web3_prediction_pm_limit_order_filled", &vars);
Ok(
create_stream_handler::<models::LimitOrderFullyFilledResponse>(
WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
stream,
id_opt.map(|s| {
if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
if let Ok(n) = s.parse::<u32>() {
return StreamId::Number(n);
}
}
StreamId::Str(s)
}),
None,
)
.await,
)
}
async fn limit_order_partially_filled(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::LimitOrderPartiallyFilledResponse>>> {
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);
let stream = replace_websocket_streams_placeholders(
"/web3_prediction_pm_limit_order_partial_fill",
&vars,
);
Ok(
create_stream_handler::<models::LimitOrderPartiallyFilledResponse>(
WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
stream,
id_opt.map(|s| {
if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
if let Ok(n) = s.parse::<u32>() {
return StreamId::Number(n);
}
}
StreamId::Str(s)
}),
None,
)
.await,
)
}
async fn limit_order_submission_failed(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::LimitOrderSubmissionFailedResponse>>> {
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);
let stream =
replace_websocket_streams_placeholders("/web3_prediction_pm_limit_submit_fail", &vars);
Ok(
create_stream_handler::<models::LimitOrderSubmissionFailedResponse>(
WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
stream,
id_opt.map(|s| {
if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
if let Ok(n) = s.parse::<u32>() {
return StreamId::Number(n);
}
}
StreamId::Str(s)
}),
None,
)
.await,
)
}
async fn limit_order_submitted_successfully(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::LimitOrderSubmittedSuccessfullyResponse>>> {
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);
let stream = replace_websocket_streams_placeholders(
"/web3_prediction_pm_limit_submit_success",
&vars,
);
Ok(
create_stream_handler::<models::LimitOrderSubmittedSuccessfullyResponse>(
WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
stream,
id_opt.map(|s| {
if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
if let Ok(n) = s.parse::<u32>() {
return StreamId::Number(n);
}
}
StreamId::Str(s)
}),
None,
)
.await,
)
}
async fn market_buy_order_failed(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::MarketBuyOrderFailedResponse>>> {
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);
let stream =
replace_websocket_streams_placeholders("/web3_prediction_pm_market_buy_fail", &vars);
Ok(
create_stream_handler::<models::MarketBuyOrderFailedResponse>(
WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
stream,
id_opt.map(|s| {
if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
if let Ok(n) = s.parse::<u32>() {
return StreamId::Number(n);
}
}
StreamId::Str(s)
}),
None,
)
.await,
)
}
async fn market_buy_order_fully_filled(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::MarketBuyOrderFullyFilledResponse>>> {
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);
let stream =
replace_websocket_streams_placeholders("/web3_prediction_pm_market_buy_success", &vars);
Ok(
create_stream_handler::<models::MarketBuyOrderFullyFilledResponse>(
WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
stream,
id_opt.map(|s| {
if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
if let Ok(n) = s.parse::<u32>() {
return StreamId::Number(n);
}
}
StreamId::Str(s)
}),
None,
)
.await,
)
}
async fn market_resolution_notification(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::MarketResolutionNotificationResponse>>> {
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);
let stream =
replace_websocket_streams_placeholders("/web3_prediction_pm_market_close", &vars);
Ok(
create_stream_handler::<models::MarketResolutionNotificationResponse>(
WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
stream,
id_opt.map(|s| {
if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
if let Ok(n) = s.parse::<u32>() {
return StreamId::Number(n);
}
}
StreamId::Str(s)
}),
None,
)
.await,
)
}
async fn market_sell_order_failed_expired(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::MarketSellOrderFailedExpiredResponse>>> {
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);
let stream =
replace_websocket_streams_placeholders("/web3_prediction_pm_market_sell_fail", &vars);
Ok(
create_stream_handler::<models::MarketSellOrderFailedExpiredResponse>(
WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
stream,
id_opt.map(|s| {
if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
if let Ok(n) = s.parse::<u32>() {
return StreamId::Number(n);
}
}
StreamId::Str(s)
}),
None,
)
.await,
)
}
async fn market_sell_order_fully_filled(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::MarketSellOrderFullyFilledResponse>>> {
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);
let stream = replace_websocket_streams_placeholders(
"/web3_prediction_pm_market_sell_success",
&vars,
);
Ok(
create_stream_handler::<models::MarketSellOrderFullyFilledResponse>(
WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
stream,
id_opt.map(|s| {
if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
if let Ok(n) = s.parse::<u32>() {
return StreamId::Number(n);
}
}
StreamId::Str(s)
}),
None,
)
.await,
)
}
async fn prediction_order_book_by_market_id(
&self,
params: PredictionOrderBookByMarketIdParams,
) -> anyhow::Result<Arc<WebsocketStream<models::PredictionOrderBookByMarketIdResponse>>> {
let PredictionOrderBookByMarketIdParams { market_id } = params;
let pairs: &[(&str, Option<String>)] = &[("marketId", Some(market_id.clone()))];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);
let stream =
replace_websocket_streams_placeholders("/web3_prediction_orderbook_<marketId>", &vars);
Ok(
create_stream_handler::<models::PredictionOrderBookByMarketIdResponse>(
WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
stream,
id_opt.map(|s| {
if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
if let Ok(n) = s.parse::<u32>() {
return StreamId::Number(n);
}
}
StreamId::Str(s)
}),
None,
)
.await,
)
}
async fn reward_claim_failed(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::RewardClaimFailedResponse>>> {
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);
let stream =
replace_websocket_streams_placeholders("/web3_prediction_pm_claim_fail", &vars);
Ok(create_stream_handler::<models::RewardClaimFailedResponse>(
WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
stream,
id_opt.map(|s| {
if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
if let Ok(n) = s.parse::<u32>() {
return StreamId::Number(n);
}
}
StreamId::Str(s)
}),
None,
)
.await)
}
async fn reward_claim_succeeded(
&self,
) -> anyhow::Result<Arc<WebsocketStream<models::RewardClaimSucceededResponse>>> {
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let id_opt: Option<String> = vars.get("id").map(std::string::ToString::to_string);
let stream =
replace_websocket_streams_placeholders("/web3_prediction_pm_claim_success", &vars);
Ok(
create_stream_handler::<models::RewardClaimSucceededResponse>(
WebsocketBase::WebsocketStreams(Arc::clone(&self.websocket_streams_base)),
stream,
id_opt.map(|s| {
if !s.is_empty() && s.bytes().all(|b| b.is_ascii_digit()) {
if let Ok(n) = s.parse::<u32>() {
return StreamId::Number(n);
}
}
StreamId::Str(s)
}),
None,
)
.await,
)
}
}
#[cfg(all(test, feature = "w3w_prediction"))]
mod tests {
use super::*;
use crate::TOKIO_SHARED_RT;
use crate::{
common::websocket::{WebsocketConnection, WebsocketHandler},
config::ConfigurationWebsocketStreams,
};
use serde_json::json;
use std::sync::atomic::{AtomicBool, Ordering};
use tokio::task::yield_now;
async fn make_streams_base() -> (Arc<WebsocketStreams>, Arc<WebsocketConnection>) {
let conn = WebsocketConnection::new("test");
let config = ConfigurationWebsocketStreams::builder()
.api_key("test-api-key")
.api_secret("test-api-secret")
.build()
.expect("Failed to build configuration");
let streams_base = WebsocketStreams::new(config, vec![conn.clone()], vec![]);
streams_base.signed_streams.store(true, Ordering::Relaxed);
conn.set_handler(streams_base.clone() as Arc<dyn WebsocketHandler>)
.await;
(streams_base, conn)
}
#[test]
fn batch_claim_partially_succeeded_should_execute_successfully() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, _) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders(
"/web3_prediction_pm_claim_partial_success",
&vars,
);
let ws_stream = api
.batch_claim_partially_succeeded()
.await
.expect("batch_claim_partially_succeeded should return a WebsocketStream");
assert!(
streams_base.is_subscribed(&stream).await,
"expected stream '{stream}' to be subscribed"
);
assert_eq!(ws_stream.id, None, "signed streams carry no request id");
});
}
#[test]
fn batch_claim_partially_succeeded_should_handle_incoming_message() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders(
"/web3_prediction_pm_claim_partial_success",
&vars,
);
let ws_stream = api.batch_claim_partially_succeeded().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_with_message = called.clone();
ws_stream.on_message(
move |_payload: models::BatchClaimPartiallySucceededResponse| {
called_with_message.store(true, Ordering::SeqCst);
},
);
let payload: Value = serde_json::from_str(
r#"{"pushId":"pm_CLAIM_BATCH_20260528003_pm_claim_partial_success_e1f2a3b4"}"#,
)
.unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(
called.load(Ordering::SeqCst),
"expected our callback to have been invoked"
);
});
}
#[test]
fn batch_claim_partially_succeeded_should_not_fire_after_unsubscribe() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders(
"/web3_prediction_pm_claim_partial_success",
&vars,
);
let ws_stream = api.batch_claim_partially_succeeded().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_clone = called.clone();
ws_stream.on_message(
move |_payload: models::BatchClaimPartiallySucceededResponse| {
called_clone.store(true, Ordering::SeqCst);
},
);
assert!(
streams_base.is_subscribed(&stream).await,
"should be subscribed before unsubscribe"
);
ws_stream.unsubscribe().await;
let payload: Value = serde_json::from_str(
r#"{"pushId":"pm_CLAIM_BATCH_20260528003_pm_claim_partial_success_e1f2a3b4"}"#,
)
.unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(
!called.load(Ordering::SeqCst),
"callback should not be invoked after unsubscribe"
);
});
}
#[test]
fn fund_transfer_failed_should_execute_successfully() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, _) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream =
replace_websocket_streams_placeholders("/web3_prediction_pm_transfer_fail", &vars);
let ws_stream = api
.fund_transfer_failed()
.await
.expect("fund_transfer_failed should return a WebsocketStream");
assert!(
streams_base.is_subscribed(&stream).await,
"expected stream '{stream}' to be subscribed"
);
assert_eq!(ws_stream.id, None, "signed streams carry no request id");
});
}
#[test]
fn fund_transfer_failed_should_handle_incoming_message() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_pm_transfer_fail", &vars);
let ws_stream = api.fund_transfer_failed().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_with_message = called.clone();
ws_stream.on_message(move |_payload: models::FundTransferFailedResponse| {
called_with_message.store(true, Ordering::SeqCst);
});
let payload: Value = serde_json::from_str(r#"{"pushId":"pm_TRANSFER_20260528002_pm_transfer_fail_a3b4c5d6","amount":"100.00"}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
});
}
#[test]
fn fund_transfer_failed_should_not_fire_after_unsubscribe() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_pm_transfer_fail", &vars);
let ws_stream = api.fund_transfer_failed().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_clone = called.clone();
ws_stream.on_message(move |_payload: models::FundTransferFailedResponse| {
called_clone.store(true, Ordering::SeqCst);
});
assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");
ws_stream.unsubscribe().await;
let payload: Value = serde_json::from_str(r#"{"pushId":"pm_TRANSFER_20260528002_pm_transfer_fail_a3b4c5d6","amount":"100.00"}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
});
}
#[test]
fn fund_transfer_succeeded_should_execute_successfully() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, _) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders(
"/web3_prediction_pm_transfer_success",
&vars,
);
let ws_stream = api
.fund_transfer_succeeded()
.await
.expect("fund_transfer_succeeded should return a WebsocketStream");
assert!(
streams_base.is_subscribed(&stream).await,
"expected stream '{stream}' to be subscribed"
);
assert_eq!(ws_stream.id, None, "signed streams carry no request id");
});
}
#[test]
fn fund_transfer_succeeded_should_handle_incoming_message() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_pm_transfer_success", &vars);
let ws_stream = api.fund_transfer_succeeded().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_with_message = called.clone();
ws_stream.on_message(move |_payload: models::FundTransferSucceededResponse| {
called_with_message.store(true, Ordering::SeqCst);
});
let payload: Value = serde_json::from_str(r#"{"pushId":"pm_TRANSFER_20260528001_pm_transfer_success_f2a3b4c5","amount":"100.00"}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
});
}
#[test]
fn fund_transfer_succeeded_should_not_fire_after_unsubscribe() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_pm_transfer_success", &vars);
let ws_stream = api.fund_transfer_succeeded().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_clone = called.clone();
ws_stream.on_message(move |_payload: models::FundTransferSucceededResponse| {
called_clone.store(true, Ordering::SeqCst);
});
assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");
ws_stream.unsubscribe().await;
let payload: Value = serde_json::from_str(r#"{"pushId":"pm_TRANSFER_20260528001_pm_transfer_success_f2a3b4c5","amount":"100.00"}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
});
}
#[test]
fn limit_order_fully_filled_should_execute_successfully() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, _) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders(
"/web3_prediction_pm_limit_order_filled",
&vars,
);
let ws_stream = api
.limit_order_fully_filled()
.await
.expect("limit_order_fully_filled should return a WebsocketStream");
assert!(
streams_base.is_subscribed(&stream).await,
"expected stream '{stream}' to be subscribed"
);
assert_eq!(ws_stream.id, None, "signed streams carry no request id");
});
}
#[test]
fn limit_order_fully_filled_should_handle_incoming_message() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_pm_limit_order_filled", &vars);
let ws_stream = api.limit_order_fully_filled().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_with_message = called.clone();
ws_stream.on_message(move |_payload: models::LimitOrderFullyFilledResponse| {
called_with_message.store(true, Ordering::SeqCst);
});
let payload: Value = serde_json::from_str(r#"{"pushId":"pm_PM2026052812347_pm_limit_order_filled_a7b8c9d0","amount":"100.00","topic":"BTC will hi.."}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
});
}
#[test]
fn limit_order_fully_filled_should_not_fire_after_unsubscribe() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_pm_limit_order_filled", &vars);
let ws_stream = api.limit_order_fully_filled().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_clone = called.clone();
ws_stream.on_message(move |_payload: models::LimitOrderFullyFilledResponse| {
called_clone.store(true, Ordering::SeqCst);
});
assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");
ws_stream.unsubscribe().await;
let payload: Value = serde_json::from_str(r#"{"pushId":"pm_PM2026052812347_pm_limit_order_filled_a7b8c9d0","amount":"100.00","topic":"BTC will hi.."}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
});
}
#[test]
fn limit_order_partially_filled_should_execute_successfully() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, _) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders(
"/web3_prediction_pm_limit_order_partial_fill",
&vars,
);
let ws_stream = api
.limit_order_partially_filled()
.await
.expect("limit_order_partially_filled should return a WebsocketStream");
assert!(
streams_base.is_subscribed(&stream).await,
"expected stream '{stream}' to be subscribed"
);
assert_eq!(ws_stream.id, None, "signed streams carry no request id");
});
}
#[test]
fn limit_order_partially_filled_should_handle_incoming_message() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_pm_limit_order_partial_fill", &vars);
let ws_stream = api.limit_order_partially_filled().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_with_message = called.clone();
ws_stream.on_message(move |_payload: models::LimitOrderPartiallyFilledResponse| {
called_with_message.store(true, Ordering::SeqCst);
});
let payload: Value = serde_json::from_str(r#"{"pushId":"pm_PM2026052812347_pm_limit_order_partial_fill_b8c9d0e1","amount":"30.00","topic":"BTC will hi.."}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
});
}
#[test]
fn limit_order_partially_filled_should_not_fire_after_unsubscribe() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_pm_limit_order_partial_fill", &vars);
let ws_stream = api.limit_order_partially_filled().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_clone = called.clone();
ws_stream.on_message(move |_payload: models::LimitOrderPartiallyFilledResponse| {
called_clone.store(true, Ordering::SeqCst);
});
assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");
ws_stream.unsubscribe().await;
let payload: Value = serde_json::from_str(r#"{"pushId":"pm_PM2026052812347_pm_limit_order_partial_fill_b8c9d0e1","amount":"30.00","topic":"BTC will hi.."}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
});
}
#[test]
fn limit_order_submission_failed_should_execute_successfully() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, _) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders(
"/web3_prediction_pm_limit_submit_fail",
&vars,
);
let ws_stream = api
.limit_order_submission_failed()
.await
.expect("limit_order_submission_failed should return a WebsocketStream");
assert!(
streams_base.is_subscribed(&stream).await,
"expected stream '{stream}' to be subscribed"
);
assert_eq!(ws_stream.id, None, "signed streams carry no request id");
});
}
#[test]
fn limit_order_submission_failed_should_handle_incoming_message() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_pm_limit_submit_fail", &vars);
let ws_stream = api.limit_order_submission_failed().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_with_message = called.clone();
ws_stream.on_message(move |_payload: models::LimitOrderSubmissionFailedResponse| {
called_with_message.store(true, Ordering::SeqCst);
});
let payload: Value = serde_json::from_str(r#"{"pushId":"pm_PM2026052812347_pm_limit_submit_fail_f6a7b8c9","topic":"BTC will hi.."}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
});
}
#[test]
fn limit_order_submission_failed_should_not_fire_after_unsubscribe() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_pm_limit_submit_fail", &vars);
let ws_stream = api.limit_order_submission_failed().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_clone = called.clone();
ws_stream.on_message(move |_payload: models::LimitOrderSubmissionFailedResponse| {
called_clone.store(true, Ordering::SeqCst);
});
assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");
ws_stream.unsubscribe().await;
let payload: Value = serde_json::from_str(r#"{"pushId":"pm_PM2026052812347_pm_limit_submit_fail_f6a7b8c9","topic":"BTC will hi.."}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
});
}
#[test]
fn limit_order_submitted_successfully_should_execute_successfully() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, _) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders(
"/web3_prediction_pm_limit_submit_success",
&vars,
);
let ws_stream = api
.limit_order_submitted_successfully()
.await
.expect("limit_order_submitted_successfully should return a WebsocketStream");
assert!(
streams_base.is_subscribed(&stream).await,
"expected stream '{stream}' to be subscribed"
);
assert_eq!(ws_stream.id, None, "signed streams carry no request id");
});
}
#[test]
fn limit_order_submitted_successfully_should_handle_incoming_message() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_pm_limit_submit_success", &vars);
let ws_stream = api.limit_order_submitted_successfully().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_with_message = called.clone();
ws_stream.on_message(move |_payload: models::LimitOrderSubmittedSuccessfullyResponse| {
called_with_message.store(true, Ordering::SeqCst);
});
let payload: Value = serde_json::from_str(r#"{"pushId":"pm_PM2026052812347_pm_limit_submit_success_e5f6a7b8","amount":"100.00","topic":"BTC will hi.."}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
});
}
#[test]
fn limit_order_submitted_successfully_should_not_fire_after_unsubscribe() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_pm_limit_submit_success", &vars);
let ws_stream = api.limit_order_submitted_successfully().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_clone = called.clone();
ws_stream.on_message(move |_payload: models::LimitOrderSubmittedSuccessfullyResponse| {
called_clone.store(true, Ordering::SeqCst);
});
assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");
ws_stream.unsubscribe().await;
let payload: Value = serde_json::from_str(r#"{"pushId":"pm_PM2026052812347_pm_limit_submit_success_e5f6a7b8","amount":"100.00","topic":"BTC will hi.."}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
});
}
#[test]
fn market_buy_order_failed_should_execute_successfully() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, _) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders(
"/web3_prediction_pm_market_buy_fail",
&vars,
);
let ws_stream = api
.market_buy_order_failed()
.await
.expect("market_buy_order_failed should return a WebsocketStream");
assert!(
streams_base.is_subscribed(&stream).await,
"expected stream '{stream}' to be subscribed"
);
assert_eq!(ws_stream.id, None, "signed streams carry no request id");
});
}
#[test]
fn market_buy_order_failed_should_handle_incoming_message() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_pm_market_buy_fail", &vars);
let ws_stream = api.market_buy_order_failed().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_with_message = called.clone();
ws_stream.on_message(move |_payload: models::MarketBuyOrderFailedResponse| {
called_with_message.store(true, Ordering::SeqCst);
});
let payload: Value = serde_json::from_str(r#"{"pushId":"pm_PM2026052812345_pm_market_buy_fail_b2c3d4e5","amount":"12.34","topic":"BTC will hi.."}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
});
}
#[test]
fn market_buy_order_failed_should_not_fire_after_unsubscribe() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_pm_market_buy_fail", &vars);
let ws_stream = api.market_buy_order_failed().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_clone = called.clone();
ws_stream.on_message(move |_payload: models::MarketBuyOrderFailedResponse| {
called_clone.store(true, Ordering::SeqCst);
});
assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");
ws_stream.unsubscribe().await;
let payload: Value = serde_json::from_str(r#"{"pushId":"pm_PM2026052812345_pm_market_buy_fail_b2c3d4e5","amount":"12.34","topic":"BTC will hi.."}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
});
}
#[test]
fn market_buy_order_fully_filled_should_execute_successfully() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, _) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders(
"/web3_prediction_pm_market_buy_success",
&vars,
);
let ws_stream = api
.market_buy_order_fully_filled()
.await
.expect("market_buy_order_fully_filled should return a WebsocketStream");
assert!(
streams_base.is_subscribed(&stream).await,
"expected stream '{stream}' to be subscribed"
);
assert_eq!(ws_stream.id, None, "signed streams carry no request id");
});
}
#[test]
fn market_buy_order_fully_filled_should_handle_incoming_message() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_pm_market_buy_success", &vars);
let ws_stream = api.market_buy_order_fully_filled().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_with_message = called.clone();
ws_stream.on_message(move |_payload: models::MarketBuyOrderFullyFilledResponse| {
called_with_message.store(true, Ordering::SeqCst);
});
let payload: Value = serde_json::from_str(r#"{"pushId":"pm_PM2026052812345_pm_market_buy_success_a1b2c3d4","amount":"12.34","topic":"BTC will hi.."}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
});
}
#[test]
fn market_buy_order_fully_filled_should_not_fire_after_unsubscribe() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_pm_market_buy_success", &vars);
let ws_stream = api.market_buy_order_fully_filled().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_clone = called.clone();
ws_stream.on_message(move |_payload: models::MarketBuyOrderFullyFilledResponse| {
called_clone.store(true, Ordering::SeqCst);
});
assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");
ws_stream.unsubscribe().await;
let payload: Value = serde_json::from_str(r#"{"pushId":"pm_PM2026052812345_pm_market_buy_success_a1b2c3d4","amount":"12.34","topic":"BTC will hi.."}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
});
}
#[test]
fn market_resolution_notification_should_execute_successfully() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, _) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream =
replace_websocket_streams_placeholders("/web3_prediction_pm_market_close", &vars);
let ws_stream = api
.market_resolution_notification()
.await
.expect("market_resolution_notification should return a WebsocketStream");
assert!(
streams_base.is_subscribed(&stream).await,
"expected stream '{stream}' to be subscribed"
);
assert_eq!(ws_stream.id, None, "signed streams carry no request id");
});
}
#[test]
fn market_resolution_notification_should_handle_incoming_message() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream =
replace_websocket_streams_placeholders("/web3_prediction_pm_market_close", &vars);
let ws_stream = api.market_resolution_notification().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_with_message = called.clone();
ws_stream.on_message(
move |_payload: models::MarketResolutionNotificationResponse| {
called_with_message.store(true, Ordering::SeqCst);
},
);
let payload: Value = serde_json::from_str(
r#"{"pushId":"pm_8859231_pm_market_close_b4c5d6e7","topic":"BTC will hi.."}"#,
)
.unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(
called.load(Ordering::SeqCst),
"expected our callback to have been invoked"
);
});
}
#[test]
fn market_resolution_notification_should_not_fire_after_unsubscribe() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream =
replace_websocket_streams_placeholders("/web3_prediction_pm_market_close", &vars);
let ws_stream = api.market_resolution_notification().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_clone = called.clone();
ws_stream.on_message(
move |_payload: models::MarketResolutionNotificationResponse| {
called_clone.store(true, Ordering::SeqCst);
},
);
assert!(
streams_base.is_subscribed(&stream).await,
"should be subscribed before unsubscribe"
);
ws_stream.unsubscribe().await;
let payload: Value = serde_json::from_str(
r#"{"pushId":"pm_8859231_pm_market_close_b4c5d6e7","topic":"BTC will hi.."}"#,
)
.unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(
!called.load(Ordering::SeqCst),
"callback should not be invoked after unsubscribe"
);
});
}
#[test]
fn market_sell_order_failed_expired_should_execute_successfully() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, _) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders(
"/web3_prediction_pm_market_sell_fail",
&vars,
);
let ws_stream = api
.market_sell_order_failed_expired()
.await
.expect("market_sell_order_failed_expired should return a WebsocketStream");
assert!(
streams_base.is_subscribed(&stream).await,
"expected stream '{stream}' to be subscribed"
);
assert_eq!(ws_stream.id, None, "signed streams carry no request id");
});
}
#[test]
fn market_sell_order_failed_expired_should_handle_incoming_message() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_pm_market_sell_fail", &vars);
let ws_stream = api.market_sell_order_failed_expired().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_with_message = called.clone();
ws_stream.on_message(move |_payload: models::MarketSellOrderFailedExpiredResponse| {
called_with_message.store(true, Ordering::SeqCst);
});
let payload: Value = serde_json::from_str(r#"{"pushId":"pm_PM2026052812346_pm_market_sell_fail_d4e5f6a7","amount":"10.50","topic":"ETH price.."}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
});
}
#[test]
fn market_sell_order_failed_expired_should_not_fire_after_unsubscribe() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_pm_market_sell_fail", &vars);
let ws_stream = api.market_sell_order_failed_expired().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_clone = called.clone();
ws_stream.on_message(move |_payload: models::MarketSellOrderFailedExpiredResponse| {
called_clone.store(true, Ordering::SeqCst);
});
assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");
ws_stream.unsubscribe().await;
let payload: Value = serde_json::from_str(r#"{"pushId":"pm_PM2026052812346_pm_market_sell_fail_d4e5f6a7","amount":"10.50","topic":"ETH price.."}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
});
}
#[test]
fn market_sell_order_fully_filled_should_execute_successfully() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, _) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders(
"/web3_prediction_pm_market_sell_success",
&vars,
);
let ws_stream = api
.market_sell_order_fully_filled()
.await
.expect("market_sell_order_fully_filled should return a WebsocketStream");
assert!(
streams_base.is_subscribed(&stream).await,
"expected stream '{stream}' to be subscribed"
);
assert_eq!(ws_stream.id, None, "signed streams carry no request id");
});
}
#[test]
fn market_sell_order_fully_filled_should_handle_incoming_message() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_pm_market_sell_success", &vars);
let ws_stream = api.market_sell_order_fully_filled().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_with_message = called.clone();
ws_stream.on_message(move |_payload: models::MarketSellOrderFullyFilledResponse| {
called_with_message.store(true, Ordering::SeqCst);
});
let payload: Value = serde_json::from_str(r#"{"pushId":"pm_PM2026052812346_pm_market_sell_success_c3d4e5f6","amount":"10.50","topic":"ETH price.."}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
});
}
#[test]
fn market_sell_order_fully_filled_should_not_fire_after_unsubscribe() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_pm_market_sell_success", &vars);
let ws_stream = api.market_sell_order_fully_filled().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_clone = called.clone();
ws_stream.on_message(move |_payload: models::MarketSellOrderFullyFilledResponse| {
called_clone.store(true, Ordering::SeqCst);
});
assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");
ws_stream.unsubscribe().await;
let payload: Value = serde_json::from_str(r#"{"pushId":"pm_PM2026052812346_pm_market_sell_success_c3d4e5f6","amount":"10.50","topic":"ETH price.."}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
});
}
#[test]
fn prediction_order_book_by_market_id_should_execute_successfully() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, _) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let params = PredictionOrderBookByMarketIdParams::builder("8859231".to_string())
.build()
.unwrap();
let PredictionOrderBookByMarketIdParams { market_id } = params.clone();
let pairs: &[(&str, Option<String>)] = &[("marketId", Some(market_id.clone()))];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders(
"/web3_prediction_orderbook_<marketId>",
&vars,
);
let ws_stream = api
.prediction_order_book_by_market_id(params)
.await
.expect("prediction_order_book_by_market_id should return a WebsocketStream");
assert!(
streams_base.is_subscribed(&stream).await,
"expected stream '{stream}' to be subscribed"
);
assert_eq!(ws_stream.id, None, "signed streams carry no request id");
});
}
#[test]
fn prediction_order_book_by_market_id_should_handle_incoming_message() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let params = PredictionOrderBookByMarketIdParams::builder("8859231".to_string()).build().unwrap();
let PredictionOrderBookByMarketIdParams {
market_id,
} = params.clone();
let pairs: &[(&str, Option<String>)] = &[
("marketId",
Some(market_id.clone())
),
];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_orderbook_<marketId>", &vars);
let ws_stream = api.prediction_order_book_by_market_id(params).await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_with_message = called.clone();
ws_stream.on_message(move |_payload: models::PredictionOrderBookByMarketIdResponse| {
called_with_message.store(true, Ordering::SeqCst);
});
let payload: Value = serde_json::from_str(r#"{"msgType":"orderbook","marketId":8859231,"updateTimestampMs":1717420800123,"asks":[[0.32,500],[0.33,1200.5]],"bids":[[0.31,800],[0.3,2000]]}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
});
}
#[test]
fn prediction_order_book_by_market_id_should_not_fire_after_unsubscribe() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let params = PredictionOrderBookByMarketIdParams::builder("8859231".to_string()).build().unwrap();
let PredictionOrderBookByMarketIdParams {
market_id,
} = params.clone();
let pairs: &[(&str, Option<String>)] = &[
("marketId",
Some(market_id.clone())
),
];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_orderbook_<marketId>", &vars);
let ws_stream = api.prediction_order_book_by_market_id(params).await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_clone = called.clone();
ws_stream.on_message(move |_payload: models::PredictionOrderBookByMarketIdResponse| {
called_clone.store(true, Ordering::SeqCst);
});
assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");
ws_stream.unsubscribe().await;
let payload: Value = serde_json::from_str(r#"{"msgType":"orderbook","marketId":8859231,"updateTimestampMs":1717420800123,"asks":[[0.32,500],[0.33,1200.5]],"bids":[[0.31,800],[0.3,2000]]}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
});
}
#[test]
fn reward_claim_failed_should_execute_successfully() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, _) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream =
replace_websocket_streams_placeholders("/web3_prediction_pm_claim_fail", &vars);
let ws_stream = api
.reward_claim_failed()
.await
.expect("reward_claim_failed should return a WebsocketStream");
assert!(
streams_base.is_subscribed(&stream).await,
"expected stream '{stream}' to be subscribed"
);
assert_eq!(ws_stream.id, None, "signed streams carry no request id");
});
}
#[test]
fn reward_claim_failed_should_handle_incoming_message() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_pm_claim_fail", &vars);
let ws_stream = api.reward_claim_failed().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_with_message = called.clone();
ws_stream.on_message(move |_payload: models::RewardClaimFailedResponse| {
called_with_message.store(true, Ordering::SeqCst);
});
let payload: Value = serde_json::from_str(r#"{"pushId":"pm_CLAIM_BATCH_20260528002_pm_claim_fail_d0e1f2a3","amount":"10.00","outcome":"Yes"}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
});
}
#[test]
fn reward_claim_failed_should_not_fire_after_unsubscribe() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_pm_claim_fail", &vars);
let ws_stream = api.reward_claim_failed().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_clone = called.clone();
ws_stream.on_message(move |_payload: models::RewardClaimFailedResponse| {
called_clone.store(true, Ordering::SeqCst);
});
assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");
ws_stream.unsubscribe().await;
let payload: Value = serde_json::from_str(r#"{"pushId":"pm_CLAIM_BATCH_20260528002_pm_claim_fail_d0e1f2a3","amount":"10.00","outcome":"Yes"}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
});
}
#[test]
fn reward_claim_succeeded_should_execute_successfully() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, _) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream =
replace_websocket_streams_placeholders("/web3_prediction_pm_claim_success", &vars);
let ws_stream = api
.reward_claim_succeeded()
.await
.expect("reward_claim_succeeded should return a WebsocketStream");
assert!(
streams_base.is_subscribed(&stream).await,
"expected stream '{stream}' to be subscribed"
);
assert_eq!(ws_stream.id, None, "signed streams carry no request id");
});
}
#[test]
fn reward_claim_succeeded_should_handle_incoming_message() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_pm_claim_success", &vars);
let ws_stream = api.reward_claim_succeeded().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_with_message = called.clone();
ws_stream.on_message(move |_payload: models::RewardClaimSucceededResponse| {
called_with_message.store(true, Ordering::SeqCst);
});
let payload: Value = serde_json::from_str(r#"{"pushId":"pm_CLAIM_BATCH_20260528001_pm_claim_success_c9d0e1f2","amount":"50.00"}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(called.load(Ordering::SeqCst), "expected our callback to have been invoked");
});
}
#[test]
fn reward_claim_succeeded_should_not_fire_after_unsubscribe() {
TOKIO_SHARED_RT.block_on(async {
let (streams_base, conn) = make_streams_base().await;
let api = ApiClient::new(streams_base.clone());
let pairs: &[(&str, Option<String>)] = &[];
let vars: HashMap<_, _> = pairs
.iter()
.filter_map(|&(k, ref v)| v.clone().map(|v| (k, v)))
.collect();
let stream = replace_websocket_streams_placeholders("/web3_prediction_pm_claim_success", &vars);
let ws_stream = api.reward_claim_succeeded().await.unwrap();
let called = Arc::new(AtomicBool::new(false));
let called_clone = called.clone();
ws_stream.on_message(move |_payload: models::RewardClaimSucceededResponse| {
called_clone.store(true, Ordering::SeqCst);
});
assert!(streams_base.is_subscribed(&stream).await, "should be subscribed before unsubscribe");
ws_stream.unsubscribe().await;
let payload: Value = serde_json::from_str(r#"{"pushId":"pm_CLAIM_BATCH_20260528001_pm_claim_success_c9d0e1f2","amount":"50.00"}"#).unwrap_or_else(|_| serde_json::json!({}));
let msg = json!({
"type": "DATA",
"topic": stream,
"data": payload.to_string(),
});
streams_base.on_message(msg.to_string(), conn.clone()).await;
yield_now().await;
assert!(!called.load(Ordering::SeqCst), "callback should not be invoked after unsubscribe");
});
}
}