lapin 3.7.2

AMQP client library
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

use crate::{
    ErrorKind, Result,
    reactor::FullReactor,
    recovery_config::RecoveryConfig,
    types::{AMQPValue, FieldTable, LongString},
};
use backon::ExponentialBuilder;
use executor_trait::FullExecutor;
use std::sync::Arc;

#[derive(Clone)]
pub struct ConnectionProperties {
    pub locale: String,
    pub client_properties: FieldTable,
    pub executor: Option<Arc<dyn FullExecutor + Send + Sync>>,
    pub reactor: Option<Arc<dyn FullReactor + Send + Sync>>,
    pub recovery_config: Option<RecoveryConfig>,
    pub backoff: ExponentialBuilder,
}

impl Default for ConnectionProperties {
    fn default() -> Self {
        Self {
            locale: "en_US".into(),
            client_properties: FieldTable::default(),
            executor: None,
            reactor: None,
            recovery_config: None,
            backoff: ExponentialBuilder::default().with_max_times(0 /* no retry by default */),
        }
    }
}

impl ConnectionProperties {
    #[must_use]
    pub fn with_connection_name(mut self, connection_name: LongString) -> Self {
        self.client_properties.insert(
            "connection_name".into(),
            AMQPValue::LongString(connection_name),
        );
        self
    }

    #[must_use]
    pub fn with_executor<E: FullExecutor + Send + Sync + 'static>(mut self, executor: E) -> Self {
        self.executor = Some(Arc::new(executor));
        self
    }

    #[must_use]
    pub fn with_reactor<R: FullReactor + Send + Sync + 'static>(mut self, reactor: R) -> Self {
        self.reactor = Some(Arc::new(reactor));
        self
    }

    #[must_use]
    #[cfg(feature = "unstable")]
    pub fn with_experimental_recovery_config(mut self, config: RecoveryConfig) -> Self {
        self.recovery_config = Some(config);
        self
    }

    #[must_use]
    pub fn with_backoff(mut self, backoff: ExponentialBuilder) -> Self {
        self.backoff = backoff;
        self
    }

    #[must_use]
    pub fn configure_backoff(mut self, conf: impl Fn(&mut ExponentialBuilder)) -> Self {
        conf(&mut self.backoff);
        self
    }

    pub(crate) fn executor(&self) -> Result<Arc<dyn FullExecutor + Send + Sync>> {
        if let Some(executor) = self.executor.clone() {
            return Ok(executor);
        }

        #[cfg(feature = "default-runtime")]
        {
            return Ok(Arc::new(async_global_executor_trait::AsyncGlobalExecutor));
        }

        #[allow(unreachable_code)]
        Err(ErrorKind::NoConfiguredExecutor.into())
    }

    pub(crate) fn reactor(&self) -> Result<Arc<dyn FullReactor + Send + Sync>> {
        if let Some(reactor) = self.reactor.clone() {
            return Ok(reactor);
        }

        #[cfg(feature = "default-runtime")]
        {
            return Ok(Arc::new(async_reactor_trait::AsyncIo));
        }

        #[allow(unreachable_code)]
        Err(ErrorKind::NoConfiguredReactor.into())
    }
}