of_adapters 0.1.0

Provider adapters and market-data abstraction for the Orderflow engine
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

use crate::{AdapterConfig, AdapterError, AdapterResult};

#[derive(Debug, Clone)]
pub struct CqgConfig {
    pub endpoint: String,
    pub private_label: String,
    pub client_id: String,
    pub username: String,
    pub password: String,
    pub ping_interval_secs: u64,
    pub heartbeat_timeout_secs: u64,
    pub reconnect_min_ms: u64,
    pub reconnect_max_ms: u64,
    pub max_inflight_requests: u32,
}

impl CqgConfig {
    pub fn from_adapter_config(cfg: &AdapterConfig) -> AdapterResult<Self> {
        let endpoint = cfg
            .endpoint
            .clone()
            .ok_or(AdapterError::NotConfigured("missing cqg endpoint"))?;
        if !endpoint.starts_with("wss://")
            && !endpoint.starts_with("ws://")
            && !endpoint.starts_with("mock://")
        {
            return Err(AdapterError::NotConfigured(
                "cqg endpoint must use wss://, ws://, or mock://",
            ));
        }

        let creds = cfg.credentials.as_ref().ok_or(AdapterError::NotConfigured(
            "missing cqg credentials reference",
        ))?;
        let username = read_env(&creds.key_id_env)?;
        let password = read_env(&creds.secret_env)?;

        let client_id = cfg
            .app_name
            .clone()
            .unwrap_or_else(|| "orderflow".to_string());

        Ok(Self {
            endpoint,
            private_label: "WebAPITest".to_string(),
            client_id,
            username,
            password,
            ping_interval_secs: 15,
            heartbeat_timeout_secs: 45,
            reconnect_min_ms: 250,
            reconnect_max_ms: 10_000,
            max_inflight_requests: 1024,
        })
    }

    pub fn validate_runtime(&self) -> AdapterResult<()> {
        if self.reconnect_min_ms > self.reconnect_max_ms {
            return Err(AdapterError::NotConfigured(
                "reconnect_min_ms must be <= reconnect_max_ms",
            ));
        }
        if self.max_inflight_requests == 0 {
            return Err(AdapterError::NotConfigured(
                "max_inflight_requests must be > 0",
            ));
        }
        if self.heartbeat_timeout_secs < self.ping_interval_secs {
            return Err(AdapterError::NotConfigured(
                "heartbeat_timeout_secs must be >= ping_interval_secs",
            ));
        }
        Ok(())
    }
}

fn read_env(refs: &str) -> AdapterResult<String> {
    if refs.trim().is_empty() {
        return Err(AdapterError::NotConfigured("empty env reference"));
    }
    let value = std::env::var(refs)
        .map_err(|_| AdapterError::NotConfigured("required cqg env var missing"))?;
    if value.trim().is_empty() {
        return Err(AdapterError::NotConfigured("required cqg env var empty"));
    }
    Ok(value)
}