[go: up one dir, main page]

cyper-core 0.1.0-beta.1

Core adaptors and services for compio running on hyper.
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

use std::{
    future::Future,
    io,
    net::SocketAddr,
    pin::Pin,
    task::{ready, Context, Poll},
};

use compio::net::{TcpListener, TcpStream};
use compio_tls::TlsStream;
use hyper::server::accept::Accept;

use crate::HttpStream;

type LocalPinBoxFuture<T> = Pin<Box<dyn Future<Output = T>>>;

/// A TCP acceptor.
pub struct Acceptor {
    listener: TcpListener,
    fut: Option<LocalPinBoxFuture<io::Result<(TcpStream, SocketAddr)>>>,
}

impl Acceptor {
    /// Create [`Acceptor`] binding to provided socket address.
    pub async fn bind(addr: &SocketAddr) -> io::Result<Self> {
        let listener = TcpListener::bind(addr).await?;
        Ok(Self::from_listener(listener))
    }

    /// Create [`Acceptor`] from an existing [`compio::net::TcpListener`].
    pub fn from_listener(listener: TcpListener) -> Self {
        Self {
            listener,
            fut: None,
        }
    }

    fn poll_accept_impl(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<io::Result<TcpStream>> {
        let listener: &'static TcpListener = unsafe { &*(&self.listener as *const _) };
        if let Some(mut fut) = self.fut.take() {
            match fut.as_mut().poll(cx) {
                Poll::Pending => {
                    self.fut = Some(fut);
                    Poll::Pending
                }
                Poll::Ready(res) => Poll::Ready(res.map(|(s, _)| s)),
            }
        } else {
            self.fut = Some(Box::pin(listener.accept()));
            cx.waker().wake_by_ref();
            Poll::Pending
        }
    }
}

impl Accept for Acceptor {
    type Conn = HttpStream;
    type Error = io::Error;

    fn poll_accept(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Self::Conn, Self::Error>>> {
        let res = ready!(self.poll_accept_impl(cx));
        Poll::Ready(Some(res.map(HttpStream::from_tcp)))
    }
}

/// A TLS acceptor.
pub struct TlsAcceptor {
    tcp_acceptor: Acceptor,
    tls_acceptor: compio_tls::TlsAcceptor,
    fut: Option<LocalPinBoxFuture<io::Result<TlsStream<TcpStream>>>>,
}

impl TlsAcceptor {
    /// Create [`TlsAcceptor`] from an existing [`compio::net::TcpListener`] and
    /// [`compio_tls::TlsAcceptor`].
    pub fn from_listener(tcp_listener: TcpListener, tls_acceptor: compio_tls::TlsAcceptor) -> Self {
        Self {
            tcp_acceptor: Acceptor::from_listener(tcp_listener),
            tls_acceptor,
            fut: None,
        }
    }
}

impl Accept for TlsAcceptor {
    type Conn = HttpStream;
    type Error = io::Error;

    fn poll_accept(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Self::Conn, Self::Error>>> {
        let acceptor: &'static compio_tls::TlsAcceptor =
            unsafe { &*(&self.tls_acceptor as *const _) };
        if let Some(mut fut) = self.fut.take() {
            match fut.as_mut().poll(cx) {
                Poll::Pending => {
                    self.fut = Some(fut);
                    Poll::Pending
                }
                Poll::Ready(res) => Poll::Ready(Some(res.map(HttpStream::from_tls))),
            }
        } else {
            let tcp_acceptor = Pin::new(&mut self.tcp_acceptor);
            let res = ready!(tcp_acceptor.poll_accept_impl(cx));
            match res {
                Ok(stream) => {
                    self.fut = Some(Box::pin(acceptor.accept(stream)));
                    cx.waker().wake_by_ref();
                    Poll::Pending
                }
                Err(e) => Poll::Ready(Some(Err(e))),
            }
        }
    }
}