Struct BrokerLimiter 

pub struct BrokerLimiter { /* private fields */ }

Shared multi-bucket rate limiter: one instance per broker user.

Every request a connector sends first calls acquire with the name of the method. The limiter spends one token from each bucket on the method’s route, or spends nothing and refuses. It never waits and never sends anything. The caller decides whether to wait for retry_after and try again.

Give the same Arc<BrokerLimiter> to every connector of one broker user (market data and user data together). The broker counts calls per user, so two limiters for one user would each let through the full rate and together exceed it. A different user (another token) needs its own limiter.

The limiter is thread-safe. Its internal lock is held only for a short, non-blocking update.

Examples

use std::collections::HashMap;
use std::time::{Duration, Instant};
use brokapi_core::Error;
use brokapi_runtime::{BrokerLimiter, BrokerLimits, BucketCfg, ConnQuota};

// One bucket of 2 tokens that refills 1 token per second. PlaceOrder spends from it.
let limits = BrokerLimits {
    buckets: HashMap::from([(
        "orders".to_string(),
        BucketCfg { rate: 1, per: Duration::from_secs(1), burst: 2 },
    )]),
    route: HashMap::from([("PlaceOrder".to_string(), vec!["orders".to_string()])]),
    conns: ConnQuota::default(),
};
let limiter = BrokerLimiter::new(&limits);

let t0 = Instant::now();
assert!(limiter.acquire("PlaceOrder", t0).is_ok());
assert!(limiter.acquire("PlaceOrder", t0).is_ok());
// The burst is spent: the third call is refused, nothing was sent.
match limiter.acquire("PlaceOrder", t0) {
    Err(Error::RateLimited { retry_after }) => assert_eq!(retry_after, Duration::from_secs(1)),
    other => panic!("expected RateLimited, got {other:?}"),
}
// A second later one token is back.
assert!(limiter.acquire("PlaceOrder", t0 + Duration::from_secs(1)).is_ok());

Implementations

impl BrokerLimiter

pub fn new(cfg: &BrokerLimits) -> Arc<Self>

Creates a limiter from a policy. Every bucket starts full.

The limiter copies the policy, so later changes to cfg have no effect on it. Returns an Arc because the limiter is meant to be shared between connectors.

Examples

use brokapi_runtime::{BrokerLimiter, BrokerLimits};

let limits = BrokerLimits::from_toml_str(raw)?;
let limiter = BrokerLimiter::new(&limits);
// Clone the Arc for each connector of the same broker user.
let for_market_data = std::sync::Arc::clone(&limiter);
assert_eq!(for_market_data.conn_quota().max_conns, 4);

pub fn conn_quota(&self) -> ConnQuota

The connection quota from the policy this limiter was built with.

The limiter does not enforce it, see ConnQuota.

Examples

use std::collections::HashMap;
use brokapi_runtime::{BrokerLimiter, BrokerLimits, ConnQuota};

let limits = BrokerLimits {
    buckets: HashMap::new(),
    route: HashMap::new(),
    conns: ConnQuota { max_subs_per_conn: 7, ..ConnQuota::default() },
};
assert_eq!(BrokerLimiter::new(&limits).conn_quota().max_subs_per_conn, 7);

pub fn usage(&self, now: Instant) -> Vec<(BucketId, u32, u32)>

How much of each bucket is left at now: (bucket id, tokens left, capacity), sorted by id. Nothing is spent. For a program that shows how close a client is to its limits.

Examples

use std::collections::HashMap;
use std::time::{Duration, Instant};
use brokapi_runtime::{BrokerLimiter, BrokerLimits, BucketCfg, ConnQuota};

let limits = BrokerLimits {
    buckets: HashMap::from([(
        "orders".to_string(),
        BucketCfg { rate: 1, per: Duration::from_secs(1), burst: 5 },
    )]),
    route: HashMap::from([("*".to_string(), vec!["orders".to_string()])]),
    conns: ConnQuota::default(),
};
let limiter = BrokerLimiter::new(&limits);
let t0 = Instant::now();
limiter.acquire("PlaceOrder", t0).unwrap();
let usage = limiter.usage(t0);
assert_eq!(usage[0].0, "orders");
assert_eq!((usage[0].1, usage[0].2), (4, 5));

pub fn acquire(&self, method: &str, now: Instant) -> Result<(), Error>

Spends one token from every bucket on the route of method, all or nothing.

The route is the method’s own entry, else the DEFAULT_ROUTE entry, else empty (then the call is free). Buckets are taken in sorted id order. If any bucket is empty, the tokens already taken from the earlier buckets are given back (up to each bucket’s burst) and the call is refused. A refused call therefore costs nothing.

now is the current time. Pass Instant::now() in production. Tests pass a made-up instant to move time without sleeping. Time must not go backwards: an instant earlier than the last one seen refills nothing.

The retry_after of a refusal is the time the first empty bucket (in sorted order) needs to get one token back. Another bucket on the route may need longer, so a retry after retry_after can be refused again.

A refusal is local: nothing was sent to the broker, so repeating the call later cannot create a duplicate order.

Errors

• [brokapi_core::Error::RateLimited] when a bucket on the route has less than one token.
• [brokapi_core::Error::Config] when the route names a bucket that is not defined. No token is spent in this case.

Examples

use std::collections::HashMap;
use std::time::{Duration, Instant};
use brokapi_core::Error;
use brokapi_runtime::{BrokerLimiter, BrokerLimits, BucketCfg, ConnQuota};

let one_per_sec = |burst| BucketCfg { rate: 1, per: Duration::from_secs(1), burst };
let limits = BrokerLimits {
    buckets: HashMap::from([
        ("global".to_string(), one_per_sec(10)),
        ("orders".to_string(), one_per_sec(1)),
    ]),
    route: HashMap::from([
        ("PlaceOrder".to_string(), vec!["global".to_string(), "orders".to_string()]),
        ("GetOrder".to_string(), vec!["global".to_string()]),
        ("Broken".to_string(), vec!["nope".to_string()]),
    ]),
    conns: ConnQuota::default(),
};
let limiter = BrokerLimiter::new(&limits);
let t0 = Instant::now();

assert!(limiter.acquire("PlaceOrder", t0).is_ok());
// "orders" is empty now. The refusal gives back the token taken from "global".
assert!(matches!(limiter.acquire("PlaceOrder", t0), Err(Error::RateLimited { .. })));
// A method with no route is free.
assert!(limiter.acquire("Unlisted", t0).is_ok());
// A route to a bucket that does not exist is a configuration error.
assert!(matches!(limiter.acquire("Broken", t0), Err(Error::Config(_))));