use hyper instead of reqwest

5 files changed, 49 insertions, 44 deletions

diff --git a/ipfs-api/Cargo.toml b/ipfs-api/Cargo.toml
index abf8189..2bb7a7e 100644
--- a/ipfs-api/Cargo.toml
+++ b/ipfs-api/Cargo.toml
@@ -8,7 +8,7 @@ authors           = ["Ferris Tseng <ferristseng@fastmail.fm>"]
 bytes             = "0.4"
 error-chain       = "0.11"
 futures           = "0.1"
-reqwest           = { version = "0.8", features = ["unstable"] }
+hyper             = "0.11"
 serde             = "1.0"
 serde_derive      = "1.0"
 serde_json        = "1.0"
diff --git a/ipfs-api/src/client.rs b/ipfs-api/src/client.rs
index e0363d4..501e726 100644
--- a/ipfs-api/src/client.rs
+++ b/ipfs-api/src/client.rs
@@ -10,9 +10,9 @@ use futures::Stream;
 use futures::future::{Future, IntoFuture};
 use read::{JsonLineDecoder, StreamReader};
 use request::{self, ApiRequest};
-use reqwest::{self, multipart, Method, StatusCode, Url};
-use reqwest::unstable::async::{self, Client, ClientBuilder};
-use response::{self, ErrorKind, Error};
+use response::{self, Error, ErrorKind};
+use hyper::{self, Body, Chunk, Request, Uri, Method, StatusCode};
+use hyper::client::{Client, HttpConnector};
 use serde::{Deserialize, Serialize};
 use serde_json;
 use std::io::Read;
@@ -33,8 +33,8 @@ type AsyncStreamResponse<T> = Box<Stream<Item = T, Error = Error>>;
 /// Asynchronous Ipfs client.
 ///
 pub struct IpfsClient {
-    base: Url,
-    client: Client,
+    base: Uri,
+    client: Client<HttpConnector, Body>,
 }
 
 impl IpfsClient {
@@ -45,12 +45,12 @@ impl IpfsClient {
         handle: &Handle,
         host: &str,
         port: u16,
-    ) -> Result<IpfsClient, Box<::std::error::Error>> {
+    ) -> Result<IpfsClient, hyper::error::UriError> {
         let base_path = IpfsClient::build_base_path(host, port)?;
 
         Ok(IpfsClient {
             base: base_path,
-            client: ClientBuilder::new().build(handle)?,
+            client: Client::new(handle),
         })
     }
 
@@ -62,13 +62,13 @@ impl IpfsClient {
 
     /// Builds the base url path for the Ipfs api.
     ///
-    fn build_base_path(host: &str, port: u16) -> Result<Url, reqwest::UrlError> {
+    fn build_base_path(host: &str, port: u16) -> Result<Uri, hyper::error::UriError> {
         format!("http://{}:{}/api/v0", host, port).parse()
     }
 
     /// Builds the url for an api call.
     ///
-    fn build_base_request<Req>(&self, req: &Req) -> Result<async::RequestBuilder, Error>
+    fn build_base_request<Req>(&self, req: &Req) -> Result<Request, Error>
     where
         Req: ApiRequest + Serialize,
     {
@@ -79,15 +79,15 @@ impl IpfsClient {
             ::serde_urlencoded::to_string(req)?
         );
 
-        url.parse::<Url>()
-            .map(|url| self.client.request(Method::Get, url))
+        url.parse::<Uri>()
+            .map(|url| Request::new(Method::Get, url))
             .map_err(From::from)
     }
 
     /// Builds an Api error from a response body.
     ///
     #[inline]
-    fn build_error_from_body(chunk: async::Chunk) -> Error {
+    fn build_error_from_body(chunk: Chunk) -> Error {
         match serde_json::from_slice(&chunk) {
             Ok(e) => ErrorKind::Api(e).into(),
             Err(_) => {
@@ -102,7 +102,7 @@ impl IpfsClient {
     /// Processes a response that expects a json encoded body, returning an
     /// error or a deserialized json response.
     ///
-    fn process_json_response<Res>(status: StatusCode, chunk: async::Chunk) -> Result<Res, Error>
+    fn process_json_response<Res>(status: StatusCode, chunk: Chunk) -> Result<Res, Error>
     where
         for<'de> Res: 'static + Deserialize<'de>,
     {
@@ -117,12 +117,13 @@ impl IpfsClient {
     /// Methods prefixed with `send_` work on a raw reqwest `RequestBuilder`
     /// instance.
     ///
-    fn send_request(mut req: async::RequestBuilder) -> AsyncResponse<(StatusCode, async::Chunk)> {
-        let res = req.send()
+    fn send_request(&self, req: Request) -> AsyncResponse<(StatusCode, Chunk)> {
+        let res = self.client
+            .request(req)
             .and_then(|res| {
                 let status = res.status();
 
-                res.into_body().concat2().map(move |chunk| (status, chunk))
+                res.body().concat2().map(move |chunk| (status, chunk))
             })
             .from_err();
 
@@ -134,12 +135,15 @@ impl IpfsClient {
     /// Methods prefixed with `send_` work on a raw reqwest `RequestBuilder`
     /// instance.
     ///
-    fn send_request_json<Res>(req: async::RequestBuilder) -> AsyncResponse<Res>
+    fn send_request_json<Res>(&self, req: Request) -> AsyncResponse<Res>
     where
         for<'de> Res: 'static + Deserialize<'de>,
     {
-        let res = IpfsClient::send_request(req).into_future().and_then(
-            move |(status, chunk)| IpfsClient::process_json_response(status, chunk),
+        let res = self.send_request(req).into_future().and_then(
+            move |(status,
+                   chunk)| {
+                IpfsClient::process_json_response(status, chunk)
+            },
         );
 
         Box::new(res)
@@ -147,13 +151,14 @@ impl IpfsClient {
 
     /// Generates a request, and returns the unprocessed response future.
     ///
-    fn request_raw<Req>(&self, req: &Req) -> AsyncResponse<(StatusCode, async::Chunk)>
+    fn request_raw<Req>(&self, req: &Req) -> AsyncResponse<(StatusCode, Chunk)>
     where
         Req: ApiRequest + Serialize,
     {
-        let res = self.build_base_request(req).into_future().and_then(|req| {
-            IpfsClient::send_request(req)
-        });
+        let res = self.build_base_request(req)
+            .map(|req| self.send_request(req))
+            .into_future()
+            .flatten();
 
         Box::new(res)
     }
@@ -166,9 +171,10 @@ impl IpfsClient {
         Req: ApiRequest + Serialize,
         for<'de> Res: 'static + Deserialize<'de>,
     {
-        let res = self.build_base_request(req).into_future().and_then(|req| {
-            IpfsClient::send_request_json(req)
-        });
+        let res = self.build_base_request(req)
+            .map(|req| self.send_request_json(req))
+            .into_future()
+            .flatten();
 
         Box::new(res)
     }
@@ -182,12 +188,10 @@ impl IpfsClient {
         for<'de> Res: 'static + Deserialize<'de>,
         R: 'static + Read + Send,
     {
-        let res = self.build_base_request(req).into_future().and_then(
-            move |req| {
-                let form = multipart::Form::new().part("file", multipart::Part::reader(data));
-                IpfsClient::send_request_json(req)
-            },
-        );
+        let res = self.build_base_request(req)
+            .map(|req| self.send_request_json(req))
+            .into_future()
+            .flatten();
 
         Box::new(res)
     }
@@ -252,11 +256,12 @@ impl IpfsClient {
         for<'de> Res: 'static + Deserialize<'de>,
     {
         let res = self.build_base_request(req)
+            .map(|req| self.client.request(req).from_err())
             .into_future()
-            .and_then(|mut req| req.send().from_err())
+            .flatten()
             .map(|res| {
                 FramedRead::new(
-                    StreamReader::new(res.into_body().from_err()),
+                    StreamReader::new(res.body().from_err()),
                     JsonLineDecoder::new(),
                 )
             })
diff --git a/ipfs-api/src/lib.rs b/ipfs-api/src/lib.rs
index b3840f5..c273990 100644
--- a/ipfs-api/src/lib.rs
+++ b/ipfs-api/src/lib.rs
@@ -10,7 +10,7 @@ extern crate bytes;
 #[macro_use]
 extern crate error_chain;
 extern crate futures;
-extern crate reqwest;
+extern crate hyper;
 extern crate serde;
 #[macro_use]
 extern crate serde_derive;
diff --git a/ipfs-api/src/read.rs b/ipfs-api/src/read.rs
index 567b823..0a009b5 100644
--- a/ipfs-api/src/read.rs
+++ b/ipfs-api/src/read.rs
@@ -8,7 +8,7 @@
 
 use bytes::BytesMut;
 use futures::{Async, Stream};
-use reqwest::unstable::async;
+use hyper::Chunk;
 use response::Error;
 use serde::Deserialize;
 use serde_json;
@@ -62,7 +62,7 @@ where
 enum ReadState {
     /// A chunk is ready to be read from.
     ///
-    Ready(async::Chunk, usize),
+    Ready(Chunk, usize),
 
     /// The next chunk isn't ready yet.
     ///
@@ -79,7 +79,7 @@ pub struct StreamReader<S> {
 
 impl<S> StreamReader<S>
 where
-    S: Stream<Item = async::Chunk, Error = Error>,
+    S: Stream<Item = Chunk, Error = Error>,
 {
     #[inline]
     pub fn new(stream: S) -> StreamReader<S> {
@@ -92,7 +92,7 @@ where
 
 impl<S> Read for StreamReader<S>
 where
-    S: Stream<Item = async::Chunk, Error = Error>,
+    S: Stream<Item = Chunk, Error = Error>,
 {
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
         loop {
@@ -146,6 +146,6 @@ where
 
 impl<S> AsyncRead for StreamReader<S>
 where
-    S: Stream<Item = async::Chunk, Error = Error>,
+    S: Stream<Item = Chunk, Error = Error>,
 {
 }
diff --git a/ipfs-api/src/response/error.rs b/ipfs-api/src/response/error.rs
index be841d8..b3d9503 100644
--- a/ipfs-api/src/response/error.rs
+++ b/ipfs-api/src/response/error.rs
@@ -6,7 +6,7 @@
 // copied, modified, or distributed except according to those terms.
 //
 
-use reqwest;
+use hyper;
 use serde_json;
 use serde_urlencoded;
 use std::string::FromUtf8Error;
@@ -22,10 +22,10 @@ pub struct ApiError {
 
 error_chain! {
     foreign_links {
-        Http(reqwest::Error);
+        Http(hyper::error::Error);
         Parse(serde_json::Error);
         ParseUtf8(FromUtf8Error);
-        Url(reqwest::UrlError);
+        Url(hyper::error::UriError);
         Io(::std::io::Error);
         EncodeUrl(serde_urlencoded::ser::Error);
     }