gel-stream 0.4.5

A library for streaming data between clients and servers.
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

use std::marker::PhantomData;

use crate::common::resolver::Resolver;
use crate::common::tokio_stream::TokioStream;
use crate::Target;
use crate::{ConnectionError, ResolvedTarget, Ssl, StreamUpgrade, TlsDriver, UpgradableStream};

type Connection<S, D> = UpgradableStream<S, D>;

#[derive(derive_more::Debug, Clone)]
enum ConnectorInner {
    #[debug("{:?}", _0)]
    Unresolved(Target, Resolver),
    #[debug("{:?}", _0)]
    Resolved(ResolvedTarget),
}

/// A connector can be used to connect multiple times to the same target.
#[derive(derive_more::Debug, Clone)]
#[allow(private_bounds)]
pub struct Connector<D: TlsDriver = Ssl> {
    target: ConnectorInner,
    #[debug(skip)]
    driver: PhantomData<D>,
    ignore_missing_close_notify: bool,
    #[cfg(feature = "keepalive")]
    keepalive: Option<std::time::Duration>,
}

impl Connector<Ssl> {
    /// Create a new connector with the given target and default resolver.
    pub fn new(target: Target) -> Result<Self, std::io::Error> {
        Self::new_explicit(target)
    }

    /// Create a new connector with the given resolved target.
    pub fn new_resolved(target: ResolvedTarget) -> Self {
        Self::new_explicit_resolved(target)
    }

    /// Create a new connector with the given target and resolver.
    pub fn new_with_resolver(target: Target, resolver: Resolver) -> Self {
        Self::new_explicit_with_resolver(target, resolver)
    }
}

#[allow(private_bounds)]
impl<D: TlsDriver> Connector<D> {
    /// Create a new connector with the given TLS driver and default resolver.
    pub fn new_explicit(target: Target) -> Result<Self, std::io::Error> {
        Ok(Self {
            target: ConnectorInner::Unresolved(target, Resolver::new()?),
            driver: PhantomData,
            ignore_missing_close_notify: false,
            #[cfg(feature = "keepalive")]
            keepalive: None,
        })
    }

    /// Create a new connector with the given TLS driver and resolved target.
    pub fn new_explicit_resolved(target: ResolvedTarget) -> Self {
        Self {
            target: ConnectorInner::Resolved(target),
            driver: PhantomData,
            ignore_missing_close_notify: false,
            #[cfg(feature = "keepalive")]
            keepalive: None,
        }
    }

    /// Create a new connector with the given TLS driver and resolver.
    pub fn new_explicit_with_resolver(target: Target, resolver: Resolver) -> Self {
        Self {
            target: ConnectorInner::Unresolved(target, resolver),
            driver: PhantomData,
            ignore_missing_close_notify: false,
            #[cfg(feature = "keepalive")]
            keepalive: None,
        }
    }

    /// Set a keepalive for the connection. This is only supported for TCP
    /// connections and will be ignored for unix sockets.
    #[cfg(feature = "keepalive")]
    pub fn set_keepalive(&mut self, keepalive: Option<std::time::Duration>) {
        self.keepalive = keepalive;
    }

    /// For TLS connections, ignore a hard close where the socket was closed
    /// before receiving CLOSE_NOTIFY.
    ///
    /// This may result in vulnerability to truncation attacks for protocols
    /// that do not include an implicit length, but may also result in spurious
    /// failures on Windows where sockets may be closed before the CLOSE_NOTIFY
    /// is received.
    pub fn ignore_missing_tls_close_notify(&mut self) {
        self.ignore_missing_close_notify = true;
    }

    /// Connect to the target.
    pub async fn connect(&self) -> Result<Connection<TokioStream, D>, ConnectionError> {
        let target = match &self.target {
            ConnectorInner::Unresolved(target, resolver) => {
                resolver.resolve_remote(target.maybe_resolved()).await?
            }
            ConnectorInner::Resolved(target) => target.clone(),
        };
        let stream = target.connect().await?;

        #[cfg(feature = "keepalive")]
        if let Some(keepalive) = self.keepalive {
            if target.is_tcp() {
                stream.set_keepalive(Some(keepalive))?;
            }
        }

        if let ConnectorInner::Unresolved(target, _) = &self.target {
            if let Some(ssl) = target.maybe_ssl() {
                let ssl = D::init_client(ssl, target.name())?;
                let mut stm = UpgradableStream::new_client(stream, Some(ssl));
                if self.ignore_missing_close_notify {
                    stm.ignore_missing_close_notify();
                }
                if !target.is_starttls() {
                    stm = stm.secure_upgrade().await?;
                }
                Ok(stm)
            } else {
                Ok(UpgradableStream::new_client(stream, None))
            }
        } else {
            Ok(UpgradableStream::new_client(stream, None))
        }
    }
}