nitox 0.1.9

Nitox is a `tokio`-based client for NATS.
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

use codec::OpCodec;
use futures::prelude::*;
use native_tls::TlsConnector as NativeTlsConnector;
use protocol::Op;
use std::net::SocketAddr;
use tokio_codec::{Decoder, Framed};
use tokio_tcp::TcpStream;
use tokio_tls::{TlsConnector, TlsStream};

use error::NatsError;

/// Inner raw stream enum over TCP and TLS/TCP
#[derive(Debug)]
pub(crate) enum NatsConnectionInner {
    /// Raw TCP Stream framed connection
    Tcp(Box<Framed<TcpStream, OpCodec>>),
    /// TLS over TCP Stream framed connection
    Tls(Box<Framed<TlsStream<TcpStream>, OpCodec>>),
}

impl NatsConnectionInner {
    /// Connects to a TCP socket
    pub(crate) fn connect_tcp(addr: &SocketAddr) -> impl Future<Item = TcpStream, Error = NatsError> {
        debug!(target: "nitox", "Connecting to {} through TCP", addr);
        TcpStream::connect(addr).from_err()
    }

    /// Upgrades an existing TCP socket to TLS over TCP
    pub(crate) fn upgrade_tcp_to_tls(
        host: &str,
        socket: TcpStream,
    ) -> impl Future<Item = TlsStream<TcpStream>, Error = NatsError> {
        let tls_connector = NativeTlsConnector::builder().build().unwrap();
        let tls_stream: TlsConnector = tls_connector.into();
        debug!(target: "nitox", "Connecting to {} through TLS over TCP", host);
        tls_stream.connect(&host, socket).from_err()
    }
}

impl From<TcpStream> for NatsConnectionInner {
    fn from(socket: TcpStream) -> Self {
        NatsConnectionInner::Tcp(Box::new(OpCodec::default().framed(socket)))
    }
}

impl From<TlsStream<TcpStream>> for NatsConnectionInner {
    fn from(socket: TlsStream<TcpStream>) -> Self {
        NatsConnectionInner::Tls(Box::new(OpCodec::default().framed(socket)))
    }
}

impl Sink for NatsConnectionInner {
    type SinkError = NatsError;
    type SinkItem = Op;

    fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
        match self {
            NatsConnectionInner::Tcp(framed) => framed.start_send(item),
            NatsConnectionInner::Tls(framed) => framed.start_send(item),
        }
    }

    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
        match self {
            NatsConnectionInner::Tcp(framed) => framed.poll_complete(),
            NatsConnectionInner::Tls(framed) => framed.poll_complete(),
        }
    }
}

impl Stream for NatsConnectionInner {
    type Error = NatsError;
    type Item = Op;

    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
        match self {
            NatsConnectionInner::Tcp(framed) => framed.poll(),
            NatsConnectionInner::Tls(framed) => framed.poll(),
        }
    }
}