amq-protocol-tcp 10.6.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
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
145
146
147
148
149
150
151
152
153

#![deny(missing_docs, missing_debug_implementations, unsafe_code)]
#![warn(unreachable_pub, unused_qualifications, unused_lifetimes)]
#![warn(
    clippy::must_use_candidate,
    clippy::unwrap_in_result,
    clippy::panic_in_result_fn
)]
#![allow(clippy::result_large_err)]

//! TCP/TLS connection helpers for AMQP URIs.
//!
//! Provides [`AMQPUriTcpExt`], a trait that extends [`AMQPUri`] with a
//! [`connect`](AMQPUriTcpExt::connect) method that opens a
//! [`TcpStream`](tcp_stream::TcpStream) — with or without TLS — according to
//! the URI scheme and the active runtime/TLS feature flags.
//!
//! # Feature flags
//!
//! ## Async runtime (pick exactly one)
//!
//! | Flag | Notes |
//! |------|-------|
//! | `tokio` *(default)* | Requires a running Tokio runtime |
//! | `smol` | Uses the smol executor |
//! | `async-global-executor` | Uses async-global-executor |
//!
//! ## TLS backend (pick at most one; `rustls` is the default)
//!
//! | Flag | Notes |
//! |------|-------|
//! | `rustls` *(default)* | TLS via rustls |
//! | `native-tls` | TLS via the platform's native library |
//! | `openssl` | TLS via OpenSSL |
//!
//! ## Rustls certificate store (only when `rustls` is active)
//!
//! | Flag | Notes |
//! |------|-------|
//! | `rustls-platform-verifier` *(default)* | Uses the platform trust store |
//! | `rustls-native-certs` | Loads native root certificates |
//! | `rustls-webpki-roots-certs` | Uses the webpki bundled root set |
//!
//! ## Rustls crypto provider (at least one must be enabled)
//!
//! | Flag | Notes |
//! |------|-------|
//! | `rustls--aws_lc_rs` *(default)* | Uses aws-lc-rs |
//! | `rustls--ring` | Uses ring (more portable) |

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)
        }
    }
}