mongodb 3.3.0

The official MongoDB driver for Rust
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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144

//! Contains the events and functionality for monitoring internal `Client` behavior.

pub mod cmap;
pub mod command;
pub mod sdam;

use std::sync::Arc;

use futures_core::future::BoxFuture;

use crate::event::command::CommandEvent;

use self::{cmap::CmapEvent, sdam::SdamEvent};

/// A destination for events.  Allows implicit conversion via [`From`] for concrete types for
/// convenience with [`crate::options::ClientOptions`] construction:
///
/// ```rust
/// # use mongodb::options::ClientOptions;
/// # fn example() {
/// let (tx, mut rx) = tokio::sync::mpsc::channel(100);
/// tokio::spawn(async move {
///     while let Some(ev) = rx.recv().await {
///         println!("{:?}", ev);
///     }
/// });
/// let options = ClientOptions::builder()
///                 .command_event_handler(tx)
///                 .build();
/// # }
/// ```
///
/// or explicit construction for `Fn` traits:
///
/// ```rust
/// # use mongodb::options::ClientOptions;
/// # use mongodb::event::EventHandler;
/// # fn example() {
/// let options = ClientOptions::builder()
///                 .command_event_handler(EventHandler::callback(|ev| println!("{:?}", ev)))
///                 .build();
/// # }
/// ```
#[derive(Clone)]
#[non_exhaustive]
pub enum EventHandler<T> {
    /// A callback.
    Callback(Arc<dyn Fn(T) + Sync + Send>),
    /// An async callback.
    AsyncCallback(Arc<dyn Fn(T) -> BoxFuture<'static, ()> + Sync + Send>),
    /// A `tokio` channel sender.
    TokioMpsc(tokio::sync::mpsc::Sender<T>),
}

impl<T> std::fmt::Debug for EventHandler<T> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_tuple("EventHandler").finish()
    }
}

impl<T> From<tokio::sync::mpsc::Sender<T>> for EventHandler<T> {
    fn from(value: tokio::sync::mpsc::Sender<T>) -> Self {
        Self::TokioMpsc(value)
    }
}

#[allow(deprecated)]
impl<T: crate::event::command::CommandEventHandler + 'static> From<Arc<T>>
    for EventHandler<CommandEvent>
{
    fn from(value: Arc<T>) -> Self {
        Self::callback(move |ev| match ev {
            CommandEvent::Started(e) => value.handle_command_started_event(e),
            CommandEvent::Succeeded(e) => value.handle_command_succeeded_event(e),
            CommandEvent::Failed(e) => value.handle_command_failed_event(e),
        })
    }
}

#[allow(deprecated)]
impl<T: crate::event::cmap::CmapEventHandler + 'static> From<Arc<T>> for EventHandler<CmapEvent> {
    fn from(value: Arc<T>) -> Self {
        use CmapEvent::*;
        Self::callback(move |ev| match ev {
            PoolCreated(ev) => value.handle_pool_created_event(ev),
            PoolReady(ev) => value.handle_pool_ready_event(ev),
            PoolCleared(ev) => value.handle_pool_cleared_event(ev),
            PoolClosed(ev) => value.handle_pool_closed_event(ev),
            ConnectionCreated(ev) => value.handle_connection_created_event(ev),
            ConnectionReady(ev) => value.handle_connection_ready_event(ev),
            ConnectionClosed(ev) => value.handle_connection_closed_event(ev),
            ConnectionCheckoutStarted(ev) => value.handle_connection_checkout_started_event(ev),
            ConnectionCheckoutFailed(ev) => value.handle_connection_checkout_failed_event(ev),
            ConnectionCheckedOut(ev) => value.handle_connection_checked_out_event(ev),
            ConnectionCheckedIn(ev) => value.handle_connection_checked_in_event(ev),
        })
    }
}

#[allow(deprecated)]
impl<T: crate::event::sdam::SdamEventHandler + 'static> From<Arc<T>> for EventHandler<SdamEvent> {
    fn from(value: Arc<T>) -> Self {
        use SdamEvent::*;
        Self::callback(move |ev| match ev {
            ServerDescriptionChanged(ev) => value.handle_server_description_changed_event(*ev),
            ServerOpening(ev) => value.handle_server_opening_event(ev),
            ServerClosed(ev) => value.handle_server_closed_event(ev),
            TopologyDescriptionChanged(ev) => value.handle_topology_description_changed_event(*ev),
            TopologyOpening(ev) => value.handle_topology_opening_event(ev),
            TopologyClosed(ev) => value.handle_topology_closed_event(ev),
            ServerHeartbeatStarted(ev) => value.handle_server_heartbeat_started_event(ev),
            ServerHeartbeatSucceeded(ev) => value.handle_server_heartbeat_succeeded_event(ev),
            ServerHeartbeatFailed(ev) => value.handle_server_heartbeat_failed_event(ev),
        })
    }
}

impl<T: Send + Sync + 'static> EventHandler<T> {
    /// Construct a new event handler with a callback.
    pub fn callback(f: impl Fn(T) + Send + Sync + 'static) -> Self {
        Self::Callback(Arc::new(f))
    }

    /// Construct a new event handler with an async callback.
    pub fn async_callback(f: impl Fn(T) -> BoxFuture<'static, ()> + Send + Sync + 'static) -> Self {
        Self::AsyncCallback(Arc::new(f))
    }

    pub(crate) fn handle(&self, event: T) {
        match self {
            // TODO RUST-1731 Use tokio's spawn_blocking
            Self::Callback(cb) => (cb)(event),
            Self::AsyncCallback(cb) => {
                crate::runtime::spawn((cb)(event));
            }
            Self::TokioMpsc(sender) => {
                let sender = sender.clone();
                crate::runtime::spawn(async move {
                    let _ = sender.send(event).await;
                });
            }
        }
    }
}