polyedge 1.0.0

Official Rust SDK for PolyEdge ultra-low latency Polymarket mempool trade streaming and real-time on-chain analytics API.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154

use futures_util::StreamExt;
use polyedge::{PolyEdgeStream, StreamError, StreamOptions};
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;

#[tokio::test]
async fn test_rust_stream_receive_and_heartbeat() {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let port = listener.local_addr().unwrap().port();

    tokio::spawn(async move {
        if let Ok((mut socket, _)) = listener.accept().await {
            let mut buf = [0u8; 1024];
            let _ = socket.read(&mut buf).await;

            let response = "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nCache-Control: no-cache\r\n\r\n: heartbeat\n\nid: 999\nevent: tx\ndata: {\"tx_hash\":\"0xrust_test\",\"timestamp\":\"2026-10-09T00:00:00Z\",\"market\":{\"id\":1,\"question\":\"\",\"slug\":\"\",\"event_slug\":\"\",\"series_slug\":\"\",\"condition_id\":\"\",\"neg_risk\":false,\"sports_market_type\":\"\",\"start_date\":\"\",\"tags_slug\":[],\"outcomes\":[\"Yes\",\"No\"],\"token_ids\":[\"1\",\"2\"]},\"taker\":{\"fee\":\"0\",\"order\":{\"order_hash\":\"\",\"shares\":\"1\",\"signature_type\":1,\"timestamp\":\"\",\"usdc\":\"1\"},\"outcome\":\"Yes\",\"shares\":\"1\",\"side\":\"BUY\",\"token_ids_index\":0,\"usdc\":\"1\",\"user\":{\"address\":\"0x1\",\"name\":\"\"}}}\n\n";
            let _ = socket.write_all(response.as_bytes()).await;
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
    });

    let stream = PolyEdgeStream::new(
        "test-stream",
        "test_key",
        Some(StreamOptions {
            stream_base_url: Some(format!("http://127.0.0.1:{}", port)),
            heartbeat_timeout: Some(Duration::from_secs(5)),
            initial_backoff: Some(Duration::from_millis(50)),
            max_backoff: Some(Duration::from_millis(200)),
        }),
    );

    let mut stream_iter = Box::pin(stream.subscribe());
    let item = stream_iter.next().await;
    assert!(item.is_some());
    let tx_res = item.unwrap();
    assert!(tx_res.is_ok());
    let tx = tx_res.unwrap();
    assert_eq!(tx.tx_hash.as_deref(), Some("0xrust_test"));
}

#[tokio::test]
async fn test_rust_stream_fatal_401() {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let port = listener.local_addr().unwrap().port();

    tokio::spawn(async move {
        if let Ok((mut socket, _)) = listener.accept().await {
            let mut buf = [0u8; 1024];
            let _ = socket.read(&mut buf).await;

            let response = "HTTP/1.1 401 Unauthorized\r\nContent-Type: application/json\r\nContent-Length: 25\r\n\r\n{\"error\":\"Unauthorized\"}";
            let _ = socket.write_all(response.as_bytes()).await;
        }
    });

    let stream = PolyEdgeStream::new(
        "fatal-stream",
        "bad_key",
        Some(StreamOptions {
            stream_base_url: Some(format!("http://127.0.0.1:{}", port)),
            heartbeat_timeout: Some(Duration::from_secs(5)),
            initial_backoff: Some(Duration::from_millis(50)),
            max_backoff: Some(Duration::from_millis(200)),
        }),
    );

    let mut stream_iter = Box::pin(stream.subscribe());
    let item = stream_iter.next().await;
    assert!(item.is_some());
    let err = item.unwrap().unwrap_err();
    assert!(matches!(err, StreamError::Unauthorized));
    assert!(err.is_fatal());
}

#[tokio::test]
async fn test_rust_stream_watchdog_and_resume() {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let port = listener.local_addr().unwrap().port();

    let conns = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
    let last_event_id_received = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));

    let conns_clone = conns.clone();
    let last_event_id_clone = last_event_id_received.clone();

    tokio::spawn(async move {
        while let Ok((mut socket, _)) = listener.accept().await {
            let conn_num = conns_clone.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            let mut buf = [0u8; 2048];
            let n = socket.read(&mut buf).await.unwrap_or(0);
            let req_str = String::from_utf8_lossy(&buf[..n]);

            let last_id = req_str
                .lines()
                .find(|l| l.to_lowercase().starts_with("last-event-id:"))
                .map(|l| l[14..].trim().to_string());
            last_event_id_clone.lock().unwrap().push(last_id);

            if conn_num == 0 {
                // First connection: send message with id 444, then stall to trigger watchdog
                let res = "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nCache-Control: no-cache\r\n\r\nid: 444\nevent: tx\ndata: {\"tx_hash\":\"0xfirst\",\"timestamp\":\"\",\"market\":{\"id\":1,\"question\":\"\",\"slug\":\"\",\"event_slug\":\"\",\"series_slug\":\"\",\"condition_id\":\"\",\"neg_risk\":false,\"sports_market_type\":\"\",\"start_date\":\"\",\"tags_slug\":[],\"outcomes\":[\"Yes\",\"No\"],\"token_ids\":[\"1\",\"2\"]},\"taker\":{\"fee\":\"0\",\"order\":{\"order_hash\":\"\",\"shares\":\"1\",\"signature_type\":1,\"timestamp\":\"\",\"usdc\":\"1\"},\"outcome\":\"Yes\",\"shares\":\"1\",\"side\":\"BUY\",\"token_ids_index\":0,\"usdc\":\"1\",\"user\":{\"address\":\"0x1\",\"name\":\"\"}}}\n\n";
                let _ = socket.write_all(res.as_bytes()).await;
                tokio::time::sleep(Duration::from_millis(1500)).await;
            } else {
                // Second connection: send heartbeat and second tx
                let res = "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nCache-Control: no-cache\r\n\r\n: heartbeat\n\nid: 445\nevent: tx\ndata: {\"tx_hash\":\"0xsecond\",\"timestamp\":\"\",\"market\":{\"id\":1,\"question\":\"\",\"slug\":\"\",\"event_slug\":\"\",\"series_slug\":\"\",\"condition_id\":\"\",\"neg_risk\":false,\"sports_market_type\":\"\",\"start_date\":\"\",\"tags_slug\":[],\"outcomes\":[\"Yes\",\"No\"],\"token_ids\":[\"1\",\"2\"]},\"taker\":{\"fee\":\"0\",\"order\":{\"order_hash\":\"\",\"shares\":\"1\",\"signature_type\":1,\"timestamp\":\"\",\"usdc\":\"1\"},\"outcome\":\"Yes\",\"shares\":\"1\",\"side\":\"BUY\",\"token_ids_index\":0,\"usdc\":\"1\",\"user\":{\"address\":\"0x1\",\"name\":\"\"}}}\n\n";
                let _ = socket.write_all(res.as_bytes()).await;
            }
        }
    });

    let stream = PolyEdgeStream::new(
        "watchdog-stream",
        "test_key",
        Some(StreamOptions {
            stream_base_url: Some(format!("http://127.0.0.1:{}", port)),
            heartbeat_timeout: Some(Duration::from_millis(300)), // 300ms watchdog
            initial_backoff: Some(Duration::from_millis(50)),
            max_backoff: Some(Duration::from_millis(200)),
        }),
    );

    let mut stream_iter = Box::pin(stream.subscribe());
    let mut received = Vec::new();

    while let Some(item) = stream_iter.next().await {
        match item {
            Ok(tx) => {
                received.push(tx.tx_hash);
                if received.len() == 2 {
                    break;
                }
            }
            Err(StreamError::WatchdogTimeout(_)) => {
                // Expected watchdog event on first disconnect
            }
            Err(e) => {
                eprintln!("Non-fatal stream notice: {:?}", e);
            }
        }
    }

    assert_eq!(received.len(), 2);
    assert_eq!(received[0].as_deref(), Some("0xfirst"));
    assert_eq!(received[1].as_deref(), Some("0xsecond"));

    let history = last_event_id_received.lock().unwrap();
    assert!(history.len() >= 2);
    assert_eq!(history[0], None);
    assert_eq!(history[1], Some("444".to_string()));
}