amq-protocol-tcp 10.2.0

AMQP URI TCP connection handling
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

#![deny(missing_docs)]
#![allow(clippy::result_large_err)]

//! # AMQP URI TCP connection handling
//!
//! amq-protocol-tcp is a library aiming at providing tools to help
//! connecting to an AMQP URI

use amq_protocol_uri::{AMQPScheme, AMQPUri};
use async_rs::{Runtime, traits::*};
use cfg_if::cfg_if;
use std::{io, time::Duration};
use tracing::trace;

/// Re-export TcpStream
pub use tcp_stream::{
    AsyncTcpStream, HandshakeError, HandshakeResult, Identity, MidHandshakeTlsStream,
    OwnedIdentity, OwnedTLSConfig, TLSConfig, TcpStream,
};

#[cfg(feature = "native-tls")]
pub use tcp_stream::NativeTlsConnector;

#[cfg(feature = "openssl")]
pub use tcp_stream::OpensslConnector;

#[cfg(feature = "rustls-common")]
pub use tcp_stream::{RustlsConnector, RustlsConnectorConfig};

/// Trait providing a method to connect to a TcpStream
pub trait AMQPUriTcpExt {
    /// connect to a TcpStream
    fn connect(&self) -> HandshakeResult
    where
        Self: Sized,
    {
        self.connect_with_config(TLSConfig::default())
    }

    /// connect to a TcpStream with the given configuration
    fn connect_with_config(&self, config: TLSConfig<'_, '_, '_>) -> HandshakeResult;

    /// connect to a TcpStream
    fn connect_async<RK: RuntimeKit + Send + Sync>(
        &self,
        runtime: &Runtime<RK>,
    ) -> impl Future<Output = io::Result<AsyncTcpStream<<RK as Reactor>::TcpStream>>>
    where
        Self: Sized,
    {
        self.connect_with_config_async(TLSConfig::default(), runtime)
    }

    /// connect to a TcpStream with the given configuration
    fn connect_with_config_async<RK: RuntimeKit + Send + Sync>(
        &self,
        config: TLSConfig<'_, '_, '_>,
        runtime: &Runtime<RK>,
    ) -> impl Future<Output = io::Result<AsyncTcpStream<<RK as Reactor>::TcpStream>>>
    where
        Self: Sized;
}

impl AMQPUriTcpExt for AMQPUri {
    fn connect_with_config(&self, config: TLSConfig<'_, '_, '_>) -> HandshakeResult {
        cfg_if! {
            if #[cfg(feature = "hickory-dns")] {
                let uri = async_rs::HickoryToSocketAddrs::new(self.authority.host.clone(), self.authority.port);
            } else {
                let uri = (self.authority.host.as_str(), self.authority.port);
            }
        }
        trace!(uri = ?uri, "Connecting.");
        let stream = if let Some(timeout) = self.query.connection_timeout {
            TcpStream::connect_timeout(uri, Duration::from_millis(timeout))
        } else {
            TcpStream::connect(uri)
        }?;
        let stream = match self.scheme {
            AMQPScheme::AMQP => stream,
            AMQPScheme::AMQPS => stream.into_tls(&self.authority.host, config)?,
        };
        stream.set_nonblocking(true)?;
        Ok(stream)
    }

    fn connect_with_config_async<RK: RuntimeKit + Send + Sync>(
        &self,
        config: TLSConfig<'_, '_, '_>,
        runtime: &Runtime<RK>,
    ) -> impl Future<Output = io::Result<AsyncTcpStream<<RK as Reactor>::TcpStream>>>
    where
        Self: Sized,
    {
        cfg_if! {
            if #[cfg(feature = "hickory-dns")] {
                let uri = async_rs::HickoryToSocketAddrs::new(self.authority.host.clone(), self.authority.port);
            } else {
                let uri = runtime.to_socket_addrs((self.authority.host.clone(), self.authority.port));
            }
        }
        trace!(uri = ?uri, "Connecting.");
        async move {
            let stream = AsyncTcpStream::connect(runtime, uri).await?;
            let stream = match self.scheme {
                AMQPScheme::AMQP => stream,
                AMQPScheme::AMQPS => stream.into_tls(&self.authority.host, config).await?,
            };
            Ok(stream)
        }
    }
}