Skip to content

Commit 8b79d1d

Browse files
psg-19psg-19
committed
feat(async): enabled async-minreq client
1 parent 713b9e6 commit 8b79d1d

File tree

2 files changed

+261
-90
lines changed

2 files changed

+261
-90
lines changed

‎src/async.rs‎

Lines changed: 205 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -9,11 +9,7 @@
99
// You may not use this file except in accordance with one or both of these
1010
// licenses.
1111

12-
//! Esplora by way of `reqwest` HTTP client.
13-
14-
use std::collections::HashMap;
15-
use std::marker::PhantomData;
16-
use std::str::FromStr;
12+
//! Esplora by way of `reqwest` or `async-minreq` HTTP client.
1713
1814
use bitcoin::consensus::{deserialize, serialize, Decodable, Encodable};
1915
use bitcoin::hashes::{sha256, Hash};
@@ -22,32 +18,44 @@ use bitcoin::Address;
2218
use bitcoin::{
2319
block::Header as BlockHeader, Block, BlockHash, MerkleBlock, Script, Transaction, Txid,
2420
};
25-
26-
#[allow(unused_imports)]
27-
use log::{debug, error, info, trace};
28-
29-
use reqwest::{header, Client, Response};
21+
use std::collections::HashMap;
22+
use std::marker::PhantomData;
23+
use std::str::FromStr;
3024

3125
use crate::api::AddressStats;
3226
use crate::{
3327
BlockStatus, BlockSummary, Builder, Error, MerkleProof, OutputStatus, Tx, TxStatus,
3428
BASE_BACKOFF_MILLIS, RETRYABLE_ERROR_CODES,
3529
};
30+
#[cfg(all(feature = "async-minreq", not(feature = "async")))]
31+
use async_minreq::{Method, Request, Response};
32+
#[cfg(feature = "async")]
33+
use reqwest::{header, Client, Response};
34+
35+
#[allow(unused_imports)]
36+
use log::{debug, error, info, trace};
37+
38+
#[cfg(all(feature = "async-minreq", not(feature = "async")))]
39+
/// Valid HTTP code
40+
const VALID_HTTP_CODE: i32 = 299;
3641

3742
#[derive(Debug, Clone)]
3843
pub struct AsyncClient<S = DefaultSleeper> {
3944
/// The URL of the Esplora Server.
4045
url: String,
41-
/// The inner [`reqwest::Client`] to make HTTP requests.
42-
client: Client,
43-
/// Number of times to retry a request
46+
/// Number of times to retry a request.
4447
max_retries: usize,
45-
46-
/// Marker for the type of sleeper used
48+
#[cfg(feature = "async")]
49+
client: Client,
50+
#[cfg(all(feature = "async-minreq", not(feature = "async")))]
51+
/// Default headers (applied to every request).
52+
headers: HashMap<String, String>,
53+
/// Marker for the sleeper.
4754
marker: PhantomData<S>,
4855
}
4956

5057
impl<S: Sleeper> AsyncClient<S> {
58+
#[cfg(feature = "async")]
5159
/// Build an async client from a builder
5260
pub fn from_builder(builder: Builder) -> Result<Self, Error> {
5361
let mut client_builder = Client::builder();
@@ -82,6 +90,18 @@ impl<S: Sleeper> AsyncClient<S> {
8290
})
8391
}
8492

93+
#[cfg(all(feature = "async-minreq", not(feature = "async")))]
94+
/// Build an async client from a builder
95+
pub fn from_builder(builder: Builder) -> Result<Self, Error> {
96+
Ok(AsyncClient {
97+
url: builder.base_url,
98+
max_retries: builder.max_retries,
99+
headers: builder.headers,
100+
marker: PhantomData,
101+
})
102+
}
103+
104+
#[cfg(feature = "async")]
85105
pub fn from_client(url: String, client: Client) -> Self {
86106
AsyncClient {
87107
url,
@@ -90,6 +110,15 @@ impl<S: Sleeper> AsyncClient<S> {
90110
marker: PhantomData,
91111
}
92112
}
113+
#[cfg(all(feature = "async-minreq", not(feature = "async")))]
114+
pub fn from_client(url: String, headers: HashMap<String, String>) -> Self {
115+
AsyncClient {
116+
url,
117+
headers,
118+
max_retries: crate::DEFAULT_MAX_RETRIES,
119+
marker: PhantomData,
120+
}
121+
}
93122

94123
/// Make an HTTP GET request to given URL, deserializing to any `T` that
95124
/// implement [`bitcoin::consensus::Decodable`].
@@ -106,14 +135,32 @@ impl<S: Sleeper> AsyncClient<S> {
106135
let url = format!("{}{}", self.url, path);
107136
let response = self.get_with_retry(&url).await?;
108137

109-
if !response.status().is_success() {
110-
return Err(Error::HttpResponse {
111-
status: response.status().as_u16(),
112-
message: response.text().await?,
113-
});
138+
#[cfg(feature = "async")]
139+
{
140+
if !response.status().is_success() {
141+
return Err(Error::HttpResponse {
142+
status: response.status().as_u16(),
143+
message: response.text().await?,
144+
});
145+
}
146+
147+
Ok(deserialize::<T>(&response.bytes().await?)?)
114148
}
115149

116-
Ok(deserialize::<T>(&response.bytes().await?)?)
150+
#[cfg(all(feature = "async-minreq", not(feature = "async")))]
151+
{
152+
if response.status_code > VALID_HTTP_CODE {
153+
return Err(Error::HttpResponse {
154+
status: response.status_code as u16,
155+
message: match response.as_str() {
156+
Ok(resp) => resp.to_string(),
157+
Err(_) => return Err(Error::InvalidResponse),
158+
},
159+
});
160+
}
161+
162+
return Ok(deserialize::<T>(response.as_bytes())?);
163+
}
117164
}
118165

119166
/// Make an HTTP GET request to given URL, deserializing to `Option<T>`.
@@ -146,14 +193,31 @@ impl<S: Sleeper> AsyncClient<S> {
146193
let url = format!("{}{}", self.url, path);
147194
let response = self.get_with_retry(&url).await?;
148195

149-
if !response.status().is_success() {
150-
return Err(Error::HttpResponse {
151-
status: response.status().as_u16(),
152-
message: response.text().await?,
153-
});
196+
#[cfg(feature = "async")]
197+
{
198+
if !response.status().is_success() {
199+
return Err(Error::HttpResponse {
200+
status: response.status().as_u16(),
201+
message: response.text().await?,
202+
});
203+
}
204+
205+
response.json::<T>().await.map_err(Error::Reqwest)
154206
}
207+
#[cfg(all(feature = "async-minreq", not(feature = "async")))]
208+
{
209+
if response.status_code > VALID_HTTP_CODE {
210+
return Err(Error::HttpResponse {
211+
status: response.status_code as u16,
212+
message: match response.as_str() {
213+
Ok(resp) => resp.to_string(),
214+
Err(_) => return Err(Error::InvalidResponse),
215+
},
216+
});
217+
}
155218

156-
response.json::<T>().await.map_err(Error::Reqwest)
219+
return response.json().map_err(Error::AsyncMinreq);
220+
}
157221
}
158222

159223
/// Make an HTTP GET request to given URL, deserializing to `Option<T>`.
@@ -188,15 +252,38 @@ impl<S: Sleeper> AsyncClient<S> {
188252
let url = format!("{}{}", self.url, path);
189253
let response = self.get_with_retry(&url).await?;
190254

191-
if !response.status().is_success() {
192-
return Err(Error::HttpResponse {
193-
status: response.status().as_u16(),
194-
message: response.text().await?,
195-
});
255+
#[cfg(feature = "async")]
256+
{
257+
if !response.status().is_success() {
258+
return Err(Error::HttpResponse {
259+
status: response.status().as_u16(),
260+
message: response.text().await?,
261+
});
262+
}
263+
264+
let hex_str = response.text().await?;
265+
Ok(deserialize(&Vec::from_hex(&hex_str)?)?)
196266
}
197267

198-
let hex_str = response.text().await?;
199-
Ok(deserialize(&Vec::from_hex(&hex_str)?)?)
268+
#[cfg(all(feature = "async-minreq", not(feature = "async")))]
269+
{
270+
if response.status_code > VALID_HTTP_CODE {
271+
return Err(Error::HttpResponse {
272+
status: response.status_code as u16,
273+
message: match response.as_str() {
274+
Ok(resp) => resp.to_string(),
275+
Err(_) => return Err(Error::InvalidResponse),
276+
},
277+
});
278+
}
279+
280+
let hex_str = match response.as_str() {
281+
Ok(resp) => resp.to_string(),
282+
Err(_) => return Err(Error::InvalidResponse),
283+
};
284+
285+
return Ok(deserialize(&Vec::from_hex(&hex_str)?)?);
286+
}
200287
}
201288

202289
/// Make an HTTP GET request to given URL, deserializing to `Option<T>`.
@@ -225,14 +312,35 @@ impl<S: Sleeper> AsyncClient<S> {
225312
let url = format!("{}{}", self.url, path);
226313
let response = self.get_with_retry(&url).await?;
227314

228-
if !response.status().is_success() {
229-
return Err(Error::HttpResponse {
230-
status: response.status().as_u16(),
231-
message: response.text().await?,
232-
});
315+
#[cfg(feature = "async")]
316+
{
317+
if !response.status().is_success() {
318+
return Err(Error::HttpResponse {
319+
status: response.status().as_u16(),
320+
message: response.text().await?,
321+
});
322+
}
323+
324+
Ok(response.text().await?)
233325
}
234326

235-
Ok(response.text().await?)
327+
#[cfg(all(feature = "async-minreq", not(feature = "async")))]
328+
{
329+
if response.status_code > VALID_HTTP_CODE {
330+
return Err(Error::HttpResponse {
331+
status: response.status_code as u16,
332+
message: match response.as_str() {
333+
Ok(resp) => resp.to_string(),
334+
Err(_) => return Err(Error::InvalidResponse),
335+
},
336+
});
337+
}
338+
339+
return Ok(match response.as_str() {
340+
Ok(resp) => resp.to_string(),
341+
Err(_) => return Err(Error::InvalidResponse),
342+
});
343+
}
236344
}
237345

238346
/// Make an HTTP GET request to given URL, deserializing to `Option<T>`.
@@ -263,15 +371,36 @@ impl<S: Sleeper> AsyncClient<S> {
263371
let url = format!("{}{}", self.url, path);
264372
let body = serialize::<T>(&body).to_lower_hex_string();
265373

266-
let response = self.client.post(url).body(body).send().await?;
374+
#[cfg(feature = "async")]
375+
{
376+
let response = self.client.post(url).body(body).send().await?;
267377

268-
if !response.status().is_success() {
269-
return Err(Error::HttpResponse {
270-
status: response.status().as_u16(),
271-
message: response.text().await?,
272-
});
273-
}
378+
if !response.status().is_success() {
379+
return Err(Error::HttpResponse {
380+
status: response.status().as_u16(),
381+
message: response.text().await?,
382+
});
383+
}
384+
385+
#[cfg(all(feature = "async-minreq", not(feature = "async")))]
386+
{
387+
let mut request = Request::new(Method::Post, &url).with_body(body);
388+
for (key, value) in &self.headers {
389+
request = request.with_header(key, value);
390+
}
274391

392+
let response = request.send().await.map_err(Error::AsyncMinreq)?;
393+
if response.status_code > VALID_HTTP_CODE {
394+
return Err(Error::HttpResponse {
395+
status: response.status_code as u16,
396+
message: match response.as_str() {
397+
Ok(resp) => resp.to_string(),
398+
Err(_) => return Err(Error::InvalidResponse),
399+
},
400+
});
401+
}
402+
}
403+
}
275404
Ok(())
276405
}
277406

@@ -454,6 +583,7 @@ impl<S: Sleeper> AsyncClient<S> {
454583
&self.url
455584
}
456585

586+
#[cfg(feature = "async")]
457587
/// Get the underlying [`Client`].
458588
pub fn client(&self) -> &Client {
459589
&self.client
@@ -465,6 +595,7 @@ impl<S: Sleeper> AsyncClient<S> {
465595
let mut delay = BASE_BACKOFF_MILLIS;
466596
let mut attempts = 0;
467597

598+
#[cfg(feature = "async")]
468599
loop {
469600
match self.client.get(url).send().await? {
470601
resp if attempts < self.max_retries && is_status_retryable(resp.status()) => {
@@ -475,13 +606,40 @@ impl<S: Sleeper> AsyncClient<S> {
475606
resp => return Ok(resp),
476607
}
477608
}
609+
610+
#[cfg(all(feature = "async-minreq", not(feature = "async")))]
611+
{
612+
loop {
613+
let mut request = Request::new(Method::Get, url);
614+
for (key, value) in &self.headers {
615+
request = request.with_header(key, value);
616+
}
617+
618+
match request.send().await? {
619+
resp if attempts < self.max_retries
620+
&& is_status_retryable(resp.status_code) =>
621+
{
622+
S::sleep(delay).await;
623+
attempts += 1;
624+
delay *= 2;
625+
}
626+
resp => return Ok(resp),
627+
}
628+
}
629+
}
478630
}
479631
}
480632

633+
#[cfg(feature = "async")]
481634
fn is_status_retryable(status: reqwest::StatusCode) -> bool {
482635
RETRYABLE_ERROR_CODES.contains(&status.as_u16())
483636
}
484637

638+
#[cfg(all(feature = "async-minreq", not(feature = "async")))]
639+
fn is_status_retryable(status: i32) -> bool {
640+
RETRYABLE_ERROR_CODES.contains(&(status as u16))
641+
}
642+
485643
pub trait Sleeper: 'static {
486644
type Sleep: std::future::Future<Output = ()>;
487645
fn sleep(dur: std::time::Duration) -> Self::Sleep;

0 commit comments

Comments
 (0)