nautilus_betfair/http/
client.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Betfair HTTP client implementation.
17
18use std::{
19    collections::HashMap,
20    num::NonZeroU32,
21    sync::{
22        Arc,
23        atomic::{AtomicU64, Ordering},
24    },
25};
26
27use nautilus_core::{MUTEX_POISONED, string::urlencoding};
28use nautilus_network::{
29    http::{HttpClient, Method},
30    ratelimiter::quota::Quota,
31    retry::{RetryConfig, RetryManager},
32};
33use serde::{Deserialize, Serialize, de::DeserializeOwned};
34use tokio_util::sync::CancellationToken;
35
36use super::{
37    error::BetfairHttpError,
38    models::{LoginResponse, LoginStatus},
39};
40use crate::common::{
41    consts::{
42        BETFAIR_ACCOUNTS_URL, BETFAIR_BETTING_URL, BETFAIR_IDENTITY_LOGIN_URL,
43        BETFAIR_KEEP_ALIVE_URL, BETFAIR_NAVIGATION_URL, BETFAIR_RATE_LIMIT_DEFAULT,
44        BETFAIR_RATE_LIMIT_ORDERS, HEADER_X_APPLICATION, HEADER_X_AUTHENTICATION,
45    },
46    credential::BetfairCredential,
47};
48
49/// Betfair JSON-RPC request envelope.
50#[derive(Debug, Serialize)]
51struct JsonRpcRequest<P: Serialize> {
52    jsonrpc: &'static str,
53    method: String,
54    params: P,
55    id: u64,
56}
57
58/// Betfair JSON-RPC response envelope.
59#[derive(Debug, Deserialize)]
60struct JsonRpcResponse<T> {
61    result: Option<T>,
62    error: Option<JsonRpcError>,
63}
64
65/// JSON-RPC error object.
66#[derive(Debug, Deserialize)]
67struct JsonRpcError {
68    code: i64,
69    message: String,
70}
71
72/// Betfair HTTP client for raw API operations.
73///
74/// Handles session-token authentication, JSON-RPC protocol, form-encoded
75/// identity requests, REST navigation, rate limiting, and retry logic.
76#[derive(Debug)]
77pub struct BetfairHttpClient {
78    client: HttpClient,
79    credential: BetfairCredential,
80    session_token: Arc<tokio::sync::RwLock<Option<String>>>,
81    retry_manager: RetryManager<BetfairHttpError>,
82    cancellation_token: std::sync::Mutex<CancellationToken>,
83    connect_lock: tokio::sync::Mutex<()>,
84    request_id: AtomicU64,
85    url_identity_login: String,
86    url_keep_alive: String,
87    url_betting: String,
88    url_accounts: String,
89    url_navigation: String,
90}
91
92impl BetfairHttpClient {
93    /// Creates a new [`BetfairHttpClient`].
94    ///
95    /// # Errors
96    ///
97    /// Returns an error if the HTTP client cannot be created.
98    pub fn new(
99        credential: BetfairCredential,
100        timeout_secs: Option<u64>,
101        max_retries: Option<u32>,
102        retry_delay_ms: Option<u64>,
103        proxy_url: Option<String>,
104        request_rate_per_second: Option<u32>,
105        order_request_rate_per_second: Option<u32>,
106    ) -> Result<Self, BetfairHttpError> {
107        let retry_config = RetryConfig {
108            max_retries: max_retries.unwrap_or(3),
109            initial_delay_ms: retry_delay_ms.unwrap_or(1000),
110            max_delay_ms: 10_000,
111            backoff_factor: 2.0,
112            jitter_ms: 500,
113            operation_timeout_ms: Some(30_000),
114            immediate_first: false,
115            max_elapsed_ms: Some(120_000),
116        };
117
118        Ok(Self {
119            client: HttpClient::new(
120                HashMap::new(),
121                Vec::new(),
122                Self::rate_limiter_quotas(
123                    request_rate_per_second.unwrap_or(5),
124                    order_request_rate_per_second.unwrap_or(20),
125                )?,
126                Self::default_quota(request_rate_per_second.unwrap_or(5))?,
127                timeout_secs,
128                proxy_url,
129            )
130            .map_err(|e| anyhow::anyhow!("Failed to create HTTP client: {e}"))?,
131            credential,
132            session_token: Arc::new(tokio::sync::RwLock::new(None)),
133            retry_manager: RetryManager::new(retry_config),
134            cancellation_token: std::sync::Mutex::new(CancellationToken::new()),
135            connect_lock: tokio::sync::Mutex::new(()),
136            request_id: AtomicU64::new(1),
137            url_identity_login: BETFAIR_IDENTITY_LOGIN_URL.to_string(),
138            url_keep_alive: BETFAIR_KEEP_ALIVE_URL.to_string(),
139            url_betting: BETFAIR_BETTING_URL.to_string(),
140            url_accounts: BETFAIR_ACCOUNTS_URL.to_string(),
141            url_navigation: BETFAIR_NAVIGATION_URL.to_string(),
142        })
143    }
144
145    /// Overrides the API base URLs (for testing with mock servers).
146    ///
147    /// The keep-alive URL is derived from `identity_login` by replacing the
148    /// path with `/keepAlive`.
149    #[must_use]
150    pub fn with_urls(
151        mut self,
152        identity_login: String,
153        betting: String,
154        accounts: String,
155        navigation: String,
156    ) -> Self {
157        // Derive keep-alive from same host as login
158        if let Some(base) = identity_login.rfind('/') {
159            self.url_keep_alive = format!("{}/keepAlive", &identity_login[..base]);
160        }
161        self.url_identity_login = identity_login;
162        self.url_betting = betting;
163        self.url_accounts = accounts;
164        self.url_navigation = navigation;
165        self
166    }
167
168    /// Returns a clone of the current cancellation token for this client.
169    ///
170    /// `disconnect()` cancels and replaces the token, so callers should fetch
171    /// a fresh clone for each operation rather than holding one long-term.
172    ///
173    /// # Panics
174    ///
175    /// Panics if the internal cancellation-token mutex is poisoned.
176    pub fn cancellation_token(&self) -> CancellationToken {
177        self.cancellation_token
178            .lock()
179            .expect(MUTEX_POISONED)
180            .clone()
181    }
182
183    /// Returns the current session token, if authenticated.
184    pub async fn session_token(&self) -> Option<String> {
185        self.session_token.read().await.clone()
186    }
187
188    /// Returns whether the client has an active session.
189    pub async fn is_connected(&self) -> bool {
190        self.session_token.read().await.is_some()
191    }
192
193    /// Returns the application key.
194    #[must_use]
195    pub fn app_key(&self) -> &str {
196        self.credential.app_key()
197    }
198
199    /// Authenticates with Betfair using interactive (non-cert) login.
200    ///
201    /// Sends credentials to the Identity API and stores the returned
202    /// session token for subsequent requests.
203    ///
204    /// # Errors
205    ///
206    /// Returns an error if the login request fails or authentication
207    /// is rejected.
208    pub async fn connect(&self) -> Result<(), BetfairHttpError> {
209        // Serialise concurrent connect calls so only one login fires; matches
210        // Python's per-instance asyncio.Lock around the same path.
211        let _guard = self.connect_lock.lock().await;
212
213        if self.session_token.read().await.is_some() {
214            log::debug!("Session token exists (already connected), skipping");
215            return Ok(());
216        }
217
218        let form_body = format!(
219            "username={}&password={}",
220            urlencoding::encode(self.credential.username()),
221            urlencoding::encode(self.credential.password()),
222        );
223
224        let resp_bytes = self
225            .send_identity(&self.url_identity_login, form_body.into_bytes())
226            .await?;
227
228        let resp: LoginResponse = serde_json::from_slice(&resp_bytes)?;
229
230        if resp.status == LoginStatus::Success {
231            log::debug!("Betfair login successful");
232            *self.session_token.write().await = Some(resp.token);
233            Ok(())
234        } else {
235            Err(BetfairHttpError::LoginFailed {
236                status: resp.error.unwrap_or_else(|| format!("{:?}", resp.status)),
237            })
238        }
239    }
240
241    /// Resets the session and re-authenticates.
242    ///
243    /// # Errors
244    ///
245    /// Returns an error if re-authentication fails.
246    pub async fn reconnect(&self) -> Result<(), BetfairHttpError> {
247        log::info!("Betfair reconnecting...");
248        *self.session_token.write().await = None;
249        self.connect().await
250    }
251
252    /// Clears the session token, cancels any in-flight retries, and primes a
253    /// fresh cancellation token for the next session.
254    ///
255    /// # Panics
256    ///
257    /// Panics if the internal cancellation-token mutex is poisoned.
258    pub async fn disconnect(&self) {
259        log::info!("Betfair disconnecting...");
260        {
261            let mut guard = self.cancellation_token.lock().expect(MUTEX_POISONED);
262            guard.cancel();
263            *guard = CancellationToken::new();
264        }
265        *self.session_token.write().await = None;
266    }
267
268    /// Sends a keep-alive request to renew the session.
269    ///
270    /// # Errors
271    ///
272    /// Returns an error if the keep-alive request fails.
273    pub async fn keep_alive(&self) -> Result<(), BetfairHttpError> {
274        let resp_bytes = self.send_identity(&self.url_keep_alive, Vec::new()).await?;
275
276        let resp: LoginResponse = serde_json::from_slice(&resp_bytes)?;
277
278        if resp.status == LoginStatus::Success {
279            *self.session_token.write().await = Some(resp.token);
280            Ok(())
281        } else {
282            Err(BetfairHttpError::LoginFailed {
283                status: resp.error.unwrap_or_else(|| format!("{:?}", resp.status)),
284            })
285        }
286    }
287
288    /// Sends a JSON-RPC request to the Betting API.
289    ///
290    /// # Errors
291    ///
292    /// Returns an error if the request fails, authentication is missing,
293    /// or the response contains a JSON-RPC error.
294    pub async fn send_betting<T, P>(&self, method: &str, params: P) -> Result<T, BetfairHttpError>
295    where
296        T: DeserializeOwned,
297        P: Serialize,
298    {
299        self.send_jsonrpc(&self.url_betting, method, params, false)
300            .await
301    }
302
303    /// Sends a JSON-RPC request to the Betting API with order rate limiting.
304    ///
305    /// # Errors
306    ///
307    /// Returns an error if the request fails, authentication is missing,
308    /// or the response contains a JSON-RPC error.
309    pub async fn send_betting_order<T, P>(
310        &self,
311        method: &str,
312        params: P,
313    ) -> Result<T, BetfairHttpError>
314    where
315        T: DeserializeOwned,
316        P: Serialize,
317    {
318        self.send_jsonrpc(&self.url_betting, method, params, true)
319            .await
320    }
321
322    /// Sends a JSON-RPC request to the Accounts API.
323    ///
324    /// # Errors
325    ///
326    /// Returns an error if the request fails, authentication is missing,
327    /// or the response contains a JSON-RPC error.
328    pub async fn send_accounts<T, P>(&self, method: &str, params: P) -> Result<T, BetfairHttpError>
329    where
330        T: DeserializeOwned,
331        P: Serialize,
332    {
333        self.send_jsonrpc(&self.url_accounts, method, params, false)
334            .await
335    }
336
337    /// Sends a GET request to the Navigation API.
338    ///
339    /// # Errors
340    ///
341    /// Returns an error if the request fails or the response cannot be parsed.
342    pub async fn send_navigation<T>(&self) -> Result<T, BetfairHttpError>
343    where
344        T: DeserializeOwned,
345    {
346        let headers = self.build_headers("application/json").await?;
347
348        let resp = self
349            .client
350            .request(
351                Method::GET,
352                self.url_navigation.clone(),
353                None,
354                Some(headers),
355                None,
356                None,
357                Some(vec![BETFAIR_RATE_LIMIT_DEFAULT.to_string()]),
358            )
359            .await
360            .map_err(|e| BetfairHttpError::NetworkError(e.to_string()))?;
361
362        if resp.status.as_u16() != 200 {
363            let body = String::from_utf8_lossy(&resp.body);
364            return Err(BetfairHttpError::UnexpectedStatus {
365                status: resp.status.as_u16(),
366                body: body.to_string(),
367            });
368        }
369
370        serde_json::from_slice(&resp.body).map_err(BetfairHttpError::from)
371    }
372
373    fn make_quota(requests_per_second: u32, label: &str) -> Result<Quota, BetfairHttpError> {
374        let rate = NonZeroU32::new(requests_per_second).ok_or_else(|| {
375            BetfairHttpError::InvalidConfiguration(format!("{label} must be greater than zero"))
376        })?;
377
378        Quota::per_second(rate).ok_or_else(|| {
379            BetfairHttpError::InvalidConfiguration(format!("Invalid {label} quota configuration"))
380        })
381    }
382
383    fn rate_limiter_quotas(
384        request_rate_per_second: u32,
385        order_request_rate_per_second: u32,
386    ) -> Result<Vec<(String, Quota)>, BetfairHttpError> {
387        Ok(vec![
388            (
389                BETFAIR_RATE_LIMIT_DEFAULT.to_string(),
390                Self::make_quota(request_rate_per_second, "request_rate_per_second")?,
391            ),
392            (
393                BETFAIR_RATE_LIMIT_ORDERS.to_string(),
394                Self::make_quota(
395                    order_request_rate_per_second,
396                    "order_request_rate_per_second",
397                )?,
398            ),
399        ])
400    }
401
402    fn default_quota(request_rate_per_second: u32) -> Result<Option<Quota>, BetfairHttpError> {
403        Ok(Some(Self::make_quota(
404            request_rate_per_second,
405            "request_rate_per_second",
406        )?))
407    }
408
409    async fn build_headers(
410        &self,
411        content_type: &str,
412    ) -> Result<HashMap<String, String>, BetfairHttpError> {
413        let token = self
414            .session_token
415            .read()
416            .await
417            .clone()
418            .ok_or(BetfairHttpError::MissingCredentials)?;
419
420        let mut headers = HashMap::new();
421        headers.insert(HEADER_X_AUTHENTICATION.to_string(), token);
422        headers.insert(
423            HEADER_X_APPLICATION.to_string(),
424            self.credential.app_key().to_string(),
425        );
426        headers.insert("Accept".to_string(), "application/json".to_string());
427        headers.insert("Content-Type".to_string(), content_type.to_string());
428        Ok(headers)
429    }
430
431    async fn send_identity(&self, url: &str, body: Vec<u8>) -> Result<Vec<u8>, BetfairHttpError> {
432        let mut headers = HashMap::new();
433        headers.insert("Accept".to_string(), "application/json".to_string());
434        headers.insert(
435            "Content-Type".to_string(),
436            "application/x-www-form-urlencoded".to_string(),
437        );
438        headers.insert(
439            HEADER_X_APPLICATION.to_string(),
440            self.credential.app_key().to_string(),
441        );
442
443        // Add session token if we have one (for keep-alive)
444        if let Some(token) = self.session_token.read().await.as_ref() {
445            headers.insert(HEADER_X_AUTHENTICATION.to_string(), token.clone());
446        }
447
448        let resp = self
449            .client
450            .request(
451                Method::POST,
452                url.to_string(),
453                None,
454                Some(headers),
455                Some(body),
456                None,
457                Some(vec![BETFAIR_RATE_LIMIT_DEFAULT.to_string()]),
458            )
459            .await
460            .map_err(|e| BetfairHttpError::NetworkError(e.to_string()))?;
461
462        if resp.status.as_u16() != 200 {
463            let body = String::from_utf8_lossy(&resp.body);
464            return Err(BetfairHttpError::UnexpectedStatus {
465                status: resp.status.as_u16(),
466                body: body.to_string(),
467            });
468        }
469
470        Ok(resp.body.to_vec())
471    }
472
473    async fn send_jsonrpc<T, P>(
474        &self,
475        base_url: &str,
476        method: &str,
477        params: P,
478        is_order: bool,
479    ) -> Result<T, BetfairHttpError>
480    where
481        T: DeserializeOwned,
482        P: Serialize,
483    {
484        let operation_id = format!("{base_url}#{method}");
485        let params_value = serde_json::to_value(&params)?;
486
487        let operation = || {
488            let method = method.to_string();
489            let params_value = params_value.clone();
490
491            async move {
492                let id = self.request_id.fetch_add(1, Ordering::SeqCst);
493                let request = JsonRpcRequest {
494                    jsonrpc: "2.0",
495                    method: method.clone(),
496                    params: params_value.clone(),
497                    id,
498                };
499
500                let body = serde_json::to_vec(&request)?;
501                let headers = self.build_headers("application/json").await?;
502
503                let rate_keys = if is_order {
504                    vec![BETFAIR_RATE_LIMIT_ORDERS.to_string()]
505                } else {
506                    vec![BETFAIR_RATE_LIMIT_DEFAULT.to_string()]
507                };
508
509                let resp = self
510                    .client
511                    .request(
512                        Method::POST,
513                        base_url.to_string(),
514                        None,
515                        Some(headers),
516                        Some(body),
517                        None,
518                        Some(rate_keys),
519                    )
520                    .await
521                    .map_err(|e| BetfairHttpError::NetworkError(e.to_string()))?;
522
523                let json_value: serde_json::Value = match serde_json::from_slice(&resp.body) {
524                    Ok(json) => json,
525                    Err(_) => {
526                        let error_body = String::from_utf8_lossy(&resp.body);
527                        let preview: String = error_body.chars().take(500).collect();
528                        log::warn!(
529                            "Non-JSON response: method={method}, status={}, body={}",
530                            resp.status.as_u16(),
531                            preview,
532                        );
533                        return Err(BetfairHttpError::UnexpectedStatus {
534                            status: resp.status.as_u16(),
535                            body: error_body.to_string(),
536                        });
537                    }
538                };
539
540                let rpc_resp: JsonRpcResponse<T> =
541                    serde_json::from_value(json_value).map_err(|e| {
542                        log::warn!(
543                            "Failed to deserialize JSON-RPC response: method={method}, error={e}",
544                        );
545                        BetfairHttpError::JsonError(e.to_string())
546                    })?;
547
548                if let Some(result) = rpc_resp.result {
549                    Ok(result)
550                } else if let Some(error) = rpc_resp.error {
551                    Err(BetfairHttpError::BetfairError {
552                        code: error.code,
553                        message: error.message,
554                    })
555                } else {
556                    Err(BetfairHttpError::JsonError(
557                        "Response contains neither result nor error".to_string(),
558                    ))
559                }
560            }
561        };
562
563        let should_retry = |error: &BetfairHttpError| -> bool { error.is_retryable() };
564
565        let create_error = |msg: String| -> BetfairHttpError {
566            if msg == "canceled" {
567                BetfairHttpError::Canceled("Adapter disconnecting or shutting down".to_string())
568            } else {
569                BetfairHttpError::NetworkError(msg)
570            }
571        };
572
573        // Snapshot the current token; `disconnect()` may swap it for a fresh
574        // one mid-flight, but the in-flight retry loop should observe the
575        // pre-disconnect token so a cancel actually unblocks it.
576        let token = self
577            .cancellation_token
578            .lock()
579            .expect(MUTEX_POISONED)
580            .clone();
581
582        let result = self
583            .retry_manager
584            .execute_with_retry_with_cancel(
585                &operation_id,
586                operation,
587                should_retry,
588                create_error,
589                &token,
590            )
591            .await;
592
593        if let Err(ref e) = result
594            && e.is_retryable()
595        {
596            log::error!("Request exhausted retries: method={method}, error={e}");
597        }
598
599        result
600    }
601}
602
603#[cfg(test)]
604mod tests {
605    use rstest::rstest;
606
607    use super::*;
608    use crate::common::consts::{
609        BETFAIR_RATE_LIMIT_DEFAULT, BETFAIR_RATE_LIMIT_ORDERS, METHOD_LIST_MARKET_CATALOGUE,
610    };
611
612    #[rstest]
613    fn test_rate_limiter_quotas_has_expected_keys() {
614        let quotas = BetfairHttpClient::rate_limiter_quotas(5, 20).unwrap();
615        let keys: Vec<&str> = quotas.iter().map(|(k, _)| k.as_str()).collect();
616        assert!(keys.contains(&BETFAIR_RATE_LIMIT_DEFAULT));
617        assert!(keys.contains(&BETFAIR_RATE_LIMIT_ORDERS));
618    }
619
620    #[rstest]
621    fn test_default_quota_is_some() {
622        assert!(BetfairHttpClient::default_quota(5).unwrap().is_some());
623    }
624
625    #[rstest]
626    fn test_rate_limiter_quotas_reject_zero_rate_limit() {
627        let result = BetfairHttpClient::rate_limiter_quotas(0, 20);
628
629        assert!(result.is_err());
630        assert!(
631            result
632                .err()
633                .unwrap()
634                .to_string()
635                .contains("request_rate_per_second")
636        );
637    }
638
639    #[rstest]
640    fn test_json_rpc_request_serialization() {
641        let request = JsonRpcRequest {
642            jsonrpc: "2.0",
643            method: METHOD_LIST_MARKET_CATALOGUE.to_string(),
644            params: serde_json::json!({"filter": {}, "maxResults": 100}),
645            id: 1,
646        };
647
648        let json = serde_json::to_value(&request).unwrap();
649        assert_eq!(json["jsonrpc"], "2.0");
650        assert_eq!(json["method"], "SportsAPING/v1.0/listMarketCatalogue");
651        assert_eq!(json["params"]["maxResults"], 100);
652        assert_eq!(json["id"], 1);
653    }
654
655    #[rstest]
656    fn test_json_rpc_response_success() {
657        let json = r#"{"result": [1, 2, 3], "error": null}"#;
658        let resp: JsonRpcResponse<Vec<i32>> = serde_json::from_str(json).unwrap();
659        assert_eq!(resp.result, Some(vec![1, 2, 3]));
660        assert!(resp.error.is_none());
661    }
662
663    #[rstest]
664    fn test_json_rpc_response_error() {
665        let json = r#"{"result": null, "error": {"code": -32600, "message": "Invalid request"}}"#;
666        let resp: JsonRpcResponse<serde_json::Value> = serde_json::from_str(json).unwrap();
667        assert!(resp.result.is_none());
668        let error = resp.error.unwrap();
669        assert_eq!(error.code, -32600);
670        assert_eq!(error.message, "Invalid request");
671    }
672}
nautilus_betfair/http/client.rs

nautilus_betfair/http/
client.rs
