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

//! Have you ever wondered if you could run [hyper](https://docs.rs/hyper) on
//! [fahrenheit](https://docs.rs/fahrenheit/)?
//! I bet you haven't, but yes, you can (but please don't).
//!
//! ## Example:
//! ```
//! use fahrenheit;
//! use hyper::{Client, Uri};
//! use hyper_fahrenheit::{Connector, FahrenheitExecutor};
//!
//! fahrenheit::run(async move {
//!   let client: Client<Connector, hyper::Body> = Client::builder()
//!       .executor(FahrenheitExecutor)
//!       .build(Connector);
//!   let res = client
//!       .get(Uri::from_static("http://httpbin.org/ip"))
//!       .await
//!       .unwrap();
//!   println!("status: {}", res.status());
//!   let buf = hyper::body::to_bytes(res).await.unwrap();
//!   println!("body: {:?}", buf);
//! ```

use futures_io::{AsyncRead, AsyncWrite};
use futures_util::future::BoxFuture;
use hyper::rt::Executor;
use hyper::{
    client::connect::{Connected, Connection},
    service::Service,
    Uri,
};
use std::io::Error;
use std::net::ToSocketAddrs;
use std::pin::Pin;
use std::task::{Context, Poll};

/// Wraps fahrenheit's TcpStream for hyper's pleasure.
pub struct AsyncTcpStream(fahrenheit::AsyncTcpStream);

impl AsyncTcpStream {
    pub fn connect<A: ToSocketAddrs>(addr: A) -> Result<AsyncTcpStream, std::io::Error> {
        Ok(AsyncTcpStream(fahrenheit::AsyncTcpStream::connect(addr)?))
    }
}

// Hyper needs this.
impl tokio::io::AsyncRead for AsyncTcpStream {
    fn poll_read(
        self: Pin<&mut Self>,
        ctx: &mut Context,
        buf: &mut [u8],
    ) -> Poll<Result<usize, Error>> {
        let this = Pin::into_inner(self);
        AsyncRead::poll_read(Pin::new(&mut this.0), ctx, buf)
    }
}

impl tokio::io::AsyncWrite for AsyncTcpStream {
    fn poll_write(
        self: Pin<&mut Self>,
        ctx: &mut Context,
        buf: &[u8],
    ) -> Poll<Result<usize, Error>> {
        let this = Pin::into_inner(self);
        AsyncWrite::poll_write(Pin::new(&mut this.0), ctx, buf)
    }

    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>> {
        let this = Pin::into_inner(self);
        AsyncWrite::poll_flush(Pin::new(&mut this.0), cx)
    }
    fn poll_shutdown(
        self: Pin<&mut Self>,
        _cx: &mut Context<'_>,
    ) -> Poll<Result<(), std::io::Error>> {
        Poll::Ready(Ok(()))
    }
}

#[derive(Clone, Copy, Debug, Default)]
pub struct Connector;

impl Service<Uri> for Connector {
    type Response = AsyncTcpStream;
    type Error = std::io::Error;
    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;

    fn call(&mut self, req: Uri) -> Self::Future {
        let fut = async move {
            let addr = format!("{}:{}", req.host().unwrap(), req.port_u16().unwrap_or(80));
            AsyncTcpStream::connect(addr)
        };

        Box::pin(fut)
    }

    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        Poll::Ready(Ok(()))
    }
}

impl Connection for AsyncTcpStream {
    fn connected(&self) -> Connected {
        Connected::new()
    }
}

// Wraps fahrenheit as hyper's Executor.
pub struct FahrenheitExecutor;

impl<Fut> Executor<Fut> for FahrenheitExecutor
where
    Fut: Send + std::future::Future<Output = ()> + 'static,
{
    fn execute(&self, fut: Fut) {
        fahrenheit::spawn(fut);
    }
}