jaeb 0.3.12

simple snapshot-driven event bus
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
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127

use std::time::Duration;

use crate::error::{ConfigError, EventBusError};
use crate::types::{BusConfig, SubscriptionPolicy};

use super::EventBus;

/// Builder for constructing an [`EventBus`] with custom configuration.
///
/// Obtain an instance via [`EventBus::builder()`]. All settings have sensible
/// defaults (see individual methods), so calling [`build`](Self::build) on a
/// freshly created builder is valid and yields a ready-to-use bus.
///
/// # Errors
///
/// [`build`](Self::build) returns [`EventBusError::InvalidConfig`] when:
/// - `buffer_size` was set to `0`.
/// - `max_concurrent_async` was set to `0`.
pub struct EventBusBuilder {
    config: BusConfig,
}

impl std::fmt::Debug for EventBusBuilder {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("EventBusBuilder").field("config", &self.config).finish()
    }
}

impl EventBusBuilder {
    pub(super) fn new() -> Self {
        Self {
            config: BusConfig::default(),
        }
    }

    /// Set the internal channel buffer capacity.
    ///
    /// This controls the maximum number of in-flight publish permits at any
    /// given time. When the buffer is full, [`EventBus::publish`] will wait
    /// until space becomes available, and [`EventBus::try_publish`] will
    /// return [`EventBusError::ChannelFull`] immediately.
    ///
    /// **Default:** `256`.
    ///
    /// # Errors
    ///
    /// [`build`](Self::build) will return an error if `size` is `0`.
    pub fn buffer_size(mut self, size: usize) -> Self {
        self.config.buffer_size = size;
        self
    }

    /// Set a per-invocation timeout for async handler tasks.
    ///
    /// If an async handler does not complete within this duration it is
    /// cancelled and treated as a failure (subject to the listener's
    /// [`SubscriptionPolicy`]). Sync handlers are not affected.
    ///
    /// **Default:** no timeout (handlers may run indefinitely).
    pub fn handler_timeout(mut self, timeout: Duration) -> Self {
        self.config.handler_timeout = Some(timeout);
        self
    }

    /// Set the maximum number of async handler tasks that may run concurrently.
    ///
    /// When this limit is reached, new async dispatches wait until a running
    /// task finishes. Sync handlers are not counted toward this limit.
    ///
    /// **Default:** unlimited (`None`).
    ///
    /// # Errors
    ///
    /// [`build`](Self::build) will return an error if `max` is `0`.
    pub fn max_concurrent_async(mut self, max: usize) -> Self {
        self.config.max_concurrent_async = Some(max);
        self
    }

    /// Set the fallback [`SubscriptionPolicy`] applied to every new subscription
    /// that does not specify its own policy.
    ///
    /// This policy is overridden on a per-subscription basis by
    /// [`EventBus::subscribe_with_policy`] and friends.
    ///
    /// **Default:** [`SubscriptionPolicy::default()`] — priority `0`, no
    /// retries, dead-letter enabled.
    pub fn default_subscription_policy(mut self, policy: SubscriptionPolicy) -> Self {
        self.config.default_subscription_policy = policy;
        self
    }

    #[deprecated(since = "0.3.3", note = "renamed to default_subscription_policy")]
    pub fn default_failure_policy(self, policy: SubscriptionPolicy) -> Self {
        self.default_subscription_policy(policy)
    }

    /// Set a deadline for draining in-flight async tasks during shutdown.
    ///
    /// If tasks do not complete within this duration after [`EventBus::shutdown`]
    /// is called they are aborted and shutdown returns
    /// [`EventBusError::ShutdownTimeout`].
    ///
    /// **Default:** no timeout (shutdown waits indefinitely for tasks to
    /// finish).
    pub fn shutdown_timeout(mut self, timeout: Duration) -> Self {
        self.config.shutdown_timeout = Some(timeout);
        self
    }

    /// Consume the builder and construct the [`EventBus`].
    ///
    /// # Errors
    ///
    /// Returns [`EventBusError::InvalidConfig`] wrapping a [`ConfigError`] if:
    /// - `buffer_size` is `0` ([`ConfigError::ZeroBufferSize`]).
    /// - `max_concurrent_async` is `0` ([`ConfigError::ZeroConcurrency`]).
    pub fn build(self) -> Result<EventBus, EventBusError> {
        if self.config.buffer_size == 0 {
            return Err(ConfigError::ZeroBufferSize.into());
        }
        if self.config.max_concurrent_async == Some(0) {
            return Err(ConfigError::ZeroConcurrency.into());
        }
        Ok(EventBus::from_config(self.config))
    }
}