forked/reqwest-impersonate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Sync Response::text() & Response::json() call their async version

Browse Source
This commit is contained in:
messense
2019-06-23 16:30:07 +08:00
committed by Sean McArthur
parent c5f2bf6c32
commit 1e7c9fe507
3 changed files with 33 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/async_impl/decoder.rs
View File

@@ -142,14 +142,6 @@ impl Decoder {
Decoder::plain_text(body) Decoder::plain_text(body)
} }
} }
pub(crate) fn content_length(&self) -> Option<u64> {
match self.inner {
Inner::PlainText(ref body) => body.content_length(),
_ => None,
}
}
} }
impl Stream for Decoder { impl Stream for Decoder {

4
src/async_impl/response.rs
View File

@@ -271,7 +271,7 @@ impl<T: Into<Body>> From<http::Response<T>> for Response {
} }
/// A JSON object. /// A JSON object.
pub struct Json<T> { struct Json<T> {
concat: Concat2<Decoder>, concat: Concat2<Decoder>,
_marker: PhantomData<T>, _marker: PhantomData<T>,
} }
@@ -294,7 +294,7 @@ impl<T> fmt::Debug for Json<T> {
} }
#[derive(Debug)] #[derive(Debug)]
pub struct Text { struct Text {
concat: Concat2<Decoder>, concat: Concat2<Decoder>,
encoding: &'static Encoding, encoding: &'static Encoding,
} }

95
src/response.rs
View File

@@ -3,14 +3,10 @@ use std::fmt;
use std::io::{self, Read}; use std::io::{self, Read};
use std::net::SocketAddr; use std::net::SocketAddr;
use std::time::Duration; use std::time::Duration;
use std::borrow::Cow;
use encoding_rs::{Encoding, UTF_8};
use futures::{Async, Poll, Stream}; use futures::{Async, Poll, Stream};
use http; use http;
use mime::Mime;
use serde::de::DeserializeOwned; use serde::de::DeserializeOwned;
use serde_json;
use cookie; use cookie;
use client::KeepCoreThreadAlive; use client::KeepCoreThreadAlive;
@@ -20,8 +16,8 @@ use {async_impl, StatusCode, Url, Version, wait};
/// A Response to a submitted `Request`. /// A Response to a submitted `Request`.
pub struct Response { pub struct Response {
inner: async_impl::Response, inner: async_impl::Response,
body: async_impl::ReadableChunks<WaitBody>, body: Option<async_impl::ReadableChunks<WaitBody>>,
content_length: Option<u64>, timeout: Option<Duration>,
_thread_handle: KeepCoreThreadAlive, _thread_handle: KeepCoreThreadAlive,
} }
@@ -32,17 +28,11 @@ impl fmt::Debug for Response {
} }
impl Response { impl Response {
pub(crate) fn new(mut res: async_impl::Response, timeout: Option<Duration>, thread: KeepCoreThreadAlive) -> Response { pub(crate) fn new(res: async_impl::Response, timeout: Option<Duration>, thread: KeepCoreThreadAlive) -> Response {
let body = mem::replace(res.body_mut(), async_impl::Decoder::empty());
let len = body.content_length();
let body = async_impl::ReadableChunks::new(WaitBody {
inner: wait::stream(body, timeout)
});
Response { Response {
inner: res, inner: res,
body: body, body: None,
content_length: len, timeout,
_thread_handle: thread, _thread_handle: thread,
} }
} }
@@ -212,21 +202,13 @@ impl Response {
/// [`serde_json::from_reader`]: https://docs.serde.rs/serde_json/fn.from_reader.html /// [`serde_json::from_reader`]: https://docs.serde.rs/serde_json/fn.from_reader.html
#[inline] #[inline]
pub fn json<T: DeserializeOwned>(&mut self) -> ::Result<T> { pub fn json<T: DeserializeOwned>(&mut self) -> ::Result<T> {
// There's 2 ways we could implement this: wait::timeout(self.inner.json(), self.timeout).map_err(|e| {
// match e {
// 1. Just using from_reader(self), making use of our blocking read adapter wait::Waited::TimedOut => ::error::timedout(None),
// 2. Just use self.inner.json().wait() wait::Waited::Executor(e) => ::error::from(e),
// wait::Waited::Inner(e) => e,
// Doing 1 is pretty easy, but it means we have the `serde_json` code }
// in more than one place, doing basically the same thing. })
//
// Doing 2 would mean `serde_json` is only in one place, but we'd
// need to update the sync Response to lazily make a blocking read
// adapter, so that our `inner` could possibly still have the original
// body.
//
// Went for easier for now, just to get it working.
serde_json::from_reader(self).map_err(::error::from)
} }
/// Get the response text. /// Get the response text.
@@ -278,38 +260,13 @@ impl Response {
/// This consumes the body. Trying to read more, or use of `response.json()` /// This consumes the body. Trying to read more, or use of `response.json()`
/// will return empty values. /// will return empty values.
pub fn text_with_charset(&mut self, default_encoding: &str) -> ::Result<String> { pub fn text_with_charset(&mut self, default_encoding: &str) -> ::Result<String> {
let len = self.content_length.unwrap_or(0); wait::timeout(self.inner.text_with_charset(default_encoding), self.timeout).map_err(|e| {
let mut content = Vec::with_capacity(len as usize); match e {
self.read_to_end(&mut content).map_err(::error::from)?; wait::Waited::TimedOut => ::error::timedout(None),
let content_type = self.headers().get(::header::CONTENT_TYPE) wait::Waited::Executor(e) => ::error::from(e),
.and_then(|value| { wait::Waited::Inner(e) => e,
value.to_str().ok()
})
.and_then(|value| {
value.parse::<Mime>().ok()
});
let encoding_name = content_type
.as_ref()
.and_then(|mime| {
mime
.get_param("charset")
.map(|charset| charset.as_str())
})
.unwrap_or(default_encoding);
let encoding = Encoding::for_label(encoding_name.as_bytes()).unwrap_or(UTF_8);
// a block because of borrow checker
{
let (text, _, _) = encoding.decode(&content);
match text {
Cow::Owned(s) => return Ok(s),
_ => (),
} }
} })
unsafe {
// decoding returned Cow::Borrowed, meaning these bytes
// are already valid utf8
Ok(String::from_utf8_unchecked(content))
}
} }
/// Copy the response body into a writer. /// Copy the response body into a writer.
@@ -357,12 +314,12 @@ impl Response {
/// ``` /// ```
#[inline] #[inline]
pub fn error_for_status(self) -> ::Result<Self> { pub fn error_for_status(self) -> ::Result<Self> {
let Response { body, content_length, inner, _thread_handle } = self; let Response { body, inner, timeout, _thread_handle } = self;
inner.error_for_status().map(move |inner| { inner.error_for_status().map(move |inner| {
Response { Response {
inner, inner,
body, body,
content_length, timeout,
_thread_handle, _thread_handle,
} }
}) })
@@ -393,7 +350,17 @@ impl Response {
impl Read for Response { impl Read for Response {
#[inline] #[inline]
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> { fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
self.body.read(buf) if self.body.is_none() {
let body = mem::replace(self.inner.body_mut(), async_impl::Decoder::empty());
let body = async_impl::ReadableChunks::new(WaitBody {
inner: wait::stream(body, self.timeout)
});
self.body = Some(body);
}
let mut body = self.body.take().unwrap();
let bytes = body.read(buf);
self.body = Some(body);
bytes
} }
} }