nstreams-rabbit 0.1.0

RabbitMQ queue and stream backends for nstreams
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

use std::time::Duration;

use thiserror::Error;

#[derive(Debug, Error)]
pub enum ConfigError {
    #[error("invalid amqp url: {0}")]
    InvalidUrl(String),
}

/// Connection settings parsed from an AMQP URL for RabbitMQ queue and stream clients.
#[derive(Debug, Clone)]
pub struct RabbitConfig {
    pub host: String,
    pub amqp_port: u16,
    pub stream_port: u16,
    pub user: String,
    pub password: String,
    pub vhost: String,
}

impl RabbitConfig {
    pub fn from_amqp_url(amqp_url: &str) -> Result<Self, ConfigError> {
        let url = amqp_url
            .strip_prefix("amqp://")
            .or_else(|| amqp_url.strip_prefix("amqps://"))
            .ok_or_else(|| ConfigError::InvalidUrl(amqp_url.to_string()))?;

        let (credentials, location) = url
            .split_once('@')
            .ok_or_else(|| ConfigError::InvalidUrl(amqp_url.to_string()))?;

        let (user, password) = credentials
            .split_once(':')
            .ok_or_else(|| ConfigError::InvalidUrl(amqp_url.to_string()))?;

        let (host_port, vhost) = location
            .split_once('/')
            .unwrap_or((location, ""));

        let (host, amqp_port) = match host_port.split_once(':') {
            Some((host, port)) => (
                host.to_string(),
                port.parse()
                    .map_err(|_| ConfigError::InvalidUrl(amqp_url.to_string()))?,
            ),
            None => (host_port.to_string(), 5672),
        };

        let vhost = if vhost.is_empty() {
            "/".to_string()
        } else {
            urlencoding::decode(vhost)
                .map(|value| value.into_owned())
                .unwrap_or_else(|_| vhost.to_string())
        };

        Ok(Self {
            host,
            amqp_port,
            stream_port: 5552,
            user: user.to_string(),
            password: password.to_string(),
            vhost,
        })
    }

    pub fn stream_max_age() -> Duration {
        Duration::from_secs(183 * 24 * 60 * 60)
    }
}