azeventhubs 0.20.0

An unofficial AMQP 1.0 rust client for Azure Event Hubs
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

use fe2o3_amqp::{session::SessionHandle, link::{SendError, LinkStateError}};
use fe2o3_amqp_management::{MgmtClient, error::Error as ManagementError, Request, Response};
use fe2o3_amqp_types::messaging::FromBody;

use crate::util::sharable::Sharable;

use super::{amqp_connection_scope::AmqpConnectionScope, error::OpenMgmtLinkError};

#[derive(Debug)]
enum State {
    Recovering,
    Connected {
        session: SessionHandle<()>,
        client: MgmtClient,
    }
}

#[derive(Debug)]
pub(crate) struct AmqpManagementLink {
    state: State,
}

impl AmqpManagementLink {
    pub(crate) fn new(session: SessionHandle<()>, client: MgmtClient) -> Self {
        let state = State::Connected {
            session,
            client,
        };
        Self {
            state
        }
    }

    pub(crate) async fn call<Req, Res>(&mut self, request: Req) -> Result<Res, ManagementError> 
    where
        Req: Request<Response = Res>,
        Res: Response,
        Res::Error: Into<ManagementError>,
        for<'de> Res::Body: FromBody<'de> + std::fmt::Debug + Send,
    {
        match &mut self.state {
            State::Recovering => {
                Err(ManagementError::Send(SendError::LinkStateError(LinkStateError::IllegalSessionState)))
            },
            State::Connected { client, .. } => {
                client.call(request).await
            }
        }
    }

    pub(crate) async fn recover(&mut self, scope: &mut AmqpConnectionScope) -> Result<(), OpenMgmtLinkError> {
        log::debug!("Recovering management link");
        if let State::Connected { session, .. } = &self.state {
            if !session.is_ended() {
                log::debug!("Management link session is still open, performing recovery anyway");
            }
        }

        // Close the old session and client
        let old_state = std::mem::replace(&mut self.state, State::Recovering);
        if let State::Connected { mut session, client } = old_state {
            if let Err(err) = client.close().await {
                log::error!("Found error closing old management client during recovery: {:?}", err);
            }
            if let Err(err) = session.end().await {
                log::error!("Found error closing old management session during recovery: {:?}", err);
            }
        }

        let (new_session, new_client) = scope.create_management_link().await?;
        let new_state = State::Connected {
            session: new_session,
            client: new_client,
        };
        let _ = std::mem::replace(&mut self.state, new_state);

        log::debug!("Management link recovered");

        Ok(())
    }
}

impl Sharable<AmqpManagementLink> {
    pub(crate) async fn call<Req, Res>(&mut self, request: Req) -> Result<Res, ManagementError>
    where
        Req: Request<Response = Res>,
        Res: Response,
        Res::Error: Into<ManagementError>,
        for<'de> Res::Body: FromBody<'de> + std::fmt::Debug + Send,
    {
        match self {
            Sharable::Owned(l) => l.call(request).await,
            Sharable::Shared(lock) => {
                let mut lock = lock.write().await;
                lock.call(request).await
            },
            Sharable::None => unreachable!(),
        }
    }
}