client.rs 6.54 KB
Newer Older
1
// Copyright 2019-2021 Parity Technologies (UK) Ltd.
David's avatar
David committed
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
//
// Permission is hereby granted, free of charge, to any
// person obtaining a copy of this software and associated
// documentation files (the "Software"), to deal in the
// Software without restriction, including without
// limitation the rights to use, copy, modify, merge,
// publish, distribute, sublicense, and/or sell copies of
// the Software, and to permit persons to whom the Software
// is furnished to do so, subject to the following
// conditions:
//
// The above copyright notice and this permission notice
// shall be included in all copies or substantial portions
// of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
// ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
// TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
// PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
// SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
// CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
// OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
// IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.

Niklas Adolfsson's avatar
Niklas Adolfsson committed
27
use crate::transport::HttpTransportClient;
David's avatar
David committed
28
29
use crate::types::{
	traits::Client,
David's avatar
David committed
30
	v2::{Id, NotificationSer, ParamsSer, RequestSer, Response, RpcError},
31
	Error, RequestIdGuard, TEN_MB_SIZE_BYTES,
32
};
33
use async_trait::async_trait;
34
use fnv::FnvHashMap;
35
use serde::de::DeserializeOwned;
David's avatar
David committed
36
use std::time::Duration;
Niklas Adolfsson's avatar
Niklas Adolfsson committed
37

Niklas Adolfsson's avatar
Niklas Adolfsson committed
38
39
40
41
/// Http Client Builder.
#[derive(Debug)]
pub struct HttpClientBuilder {
	max_request_body_size: u32,
David's avatar
David committed
42
	request_timeout: Duration,
43
	max_concurrent_requests: usize,
Niklas Adolfsson's avatar
Niklas Adolfsson committed
44
45
46
47
48
49
50
51
52
}

impl HttpClientBuilder {
	/// Sets the maximum size of a request body in bytes (default is 10 MiB).
	pub fn max_request_body_size(mut self, size: u32) -> Self {
		self.max_request_body_size = size;
		self
	}

David's avatar
David committed
53
54
55
56
57
58
	/// Set request timeout (default is 60 seconds).
	pub fn request_timeout(mut self, timeout: Duration) -> Self {
		self.request_timeout = timeout;
		self
	}

59
60
61
62
63
64
	/// Set max concurrent requests.
	pub fn max_concurrent_requests(mut self, max: usize) -> Self {
		self.max_concurrent_requests = max;
		self
	}

Niklas Adolfsson's avatar
Niklas Adolfsson committed
65
66
	/// Build the HTTP client with target to connect to.
	pub fn build(self, target: impl AsRef<str>) -> Result<HttpClient, Error> {
67
		let transport =
68
			HttpTransportClient::new(target, self.max_request_body_size).map_err(|e| Error::Transport(e.into()))?;
69
70
71
72
73
		Ok(HttpClient {
			transport,
			id_guard: RequestIdGuard::new(self.max_concurrent_requests),
			request_timeout: self.request_timeout,
		})
Niklas Adolfsson's avatar
Niklas Adolfsson committed
74
75
76
77
78
	}
}

impl Default for HttpClientBuilder {
	fn default() -> Self {
79
80
81
82
83
		Self {
			max_request_body_size: TEN_MB_SIZE_BYTES,
			request_timeout: Duration::from_secs(60),
			max_concurrent_requests: 256,
		}
Niklas Adolfsson's avatar
Niklas Adolfsson committed
84
85
86
	}
}

Niklas Adolfsson's avatar
Niklas Adolfsson committed
87
/// JSON-RPC HTTP Client that provides functionality to perform method calls and notifications.
88
#[derive(Debug)]
Niklas Adolfsson's avatar
Niklas Adolfsson committed
89
90
91
pub struct HttpClient {
	/// HTTP transport client.
	transport: HttpTransportClient,
David's avatar
David committed
92
93
	/// Request timeout. Defaults to 60sec.
	request_timeout: Duration,
94
95
	/// Request ID manager.
	id_guard: RequestIdGuard,
Niklas Adolfsson's avatar
Niklas Adolfsson committed
96
97
}

98
99
#[async_trait]
impl Client for HttpClient {
100
	async fn notification<'a>(&self, method: &'a str, params: Option<ParamsSer<'a>>) -> Result<(), Error> {
David's avatar
David committed
101
		let notif = NotificationSer::new(method, params);
David's avatar
David committed
102
		let fut = self.transport.send(serde_json::to_string(&notif).map_err(Error::ParseError)?);
103
		match tokio::time::timeout(self.request_timeout, fut).await {
David's avatar
David committed
104
105
			Ok(Ok(ok)) => Ok(ok),
			Err(_) => Err(Error::RequestTimeout),
106
			Ok(Err(e)) => Err(Error::Transport(e.into())),
David's avatar
David committed
107
		}
Niklas Adolfsson's avatar
Niklas Adolfsson committed
108
109
110
	}

	/// Perform a request towards the server.
111
	async fn request<'a, R>(&self, method: &'a str, params: Option<ParamsSer<'a>>) -> Result<R, Error>
112
	where
113
		R: DeserializeOwned,
114
	{
115
116
		// NOTE: the IDs wrap on overflow which is intended.
		let id = self.id_guard.next_request_id()?;
David's avatar
David committed
117
		let request = RequestSer::new(Id::Number(id), method, params);
118

119
120
121
122
		let fut = self.transport.send_and_read_body(serde_json::to_string(&request).map_err(|e| {
			self.id_guard.reclaim_request_id();
			Error::ParseError(e)
		})?);
123
		let body = match tokio::time::timeout(self.request_timeout, fut).await {
David's avatar
David committed
124
			Ok(Ok(body)) => body,
125
126
127
128
129
130
131
132
			Err(_e) => {
				self.id_guard.reclaim_request_id();
				return Err(Error::RequestTimeout);
			}
			Ok(Err(e)) => {
				self.id_guard.reclaim_request_id();
				return Err(Error::Transport(e.into()));
			}
David's avatar
David committed
133
		};
Niklas Adolfsson's avatar
Niklas Adolfsson committed
134

135
		self.id_guard.reclaim_request_id();
David's avatar
David committed
136
		let response: Response<_> = match serde_json::from_slice(&body) {
137
138
			Ok(response) => response,
			Err(_) => {
David's avatar
David committed
139
				let err: RpcError = serde_json::from_slice(&body).map_err(Error::ParseError)?;
140
				return Err(Error::Request(err.to_string()));
141
142
143
			}
		};

144
		let response_id = response.id.as_number().copied().ok_or(Error::InvalidRequestId)?;
145
146
147
148
149
150

		if response_id == id {
			Ok(response.result)
		} else {
			Err(Error::InvalidRequestId)
		}
Niklas Adolfsson's avatar
Niklas Adolfsson committed
151
	}
152

153
	async fn batch_request<'a, R>(&self, batch: Vec<(&'a str, Option<ParamsSer<'a>>)>) -> Result<Vec<R>, Error>
154
	where
155
		R: DeserializeOwned + Default + Clone,
156
	{
157
		let mut batch_request = Vec::with_capacity(batch.len());
158
159
160
161
		// NOTE(niklasad1): `ID` is not necessarily monotonically increasing.
		let mut ordered_requests = Vec::with_capacity(batch.len());
		let mut request_set = FnvHashMap::with_capacity_and_hasher(batch.len(), Default::default());

162
		let ids = self.id_guard.next_request_ids(batch.len())?;
163
		for (pos, (method, params)) in batch.into_iter().enumerate() {
164
165
166
			batch_request.push(RequestSer::new(Id::Number(ids[pos]), method, params));
			ordered_requests.push(ids[pos]);
			request_set.insert(ids[pos], pos);
167
168
		}

169
170
171
172
		let fut = self.transport.send_and_read_body(serde_json::to_string(&batch_request).map_err(|e| {
			self.id_guard.reclaim_request_id();
			Error::ParseError(e)
		})?);
David's avatar
David committed
173

174
		let body = match tokio::time::timeout(self.request_timeout, fut).await {
David's avatar
David committed
175
176
			Ok(Ok(body)) => body,
			Err(_e) => return Err(Error::RequestTimeout),
177
			Ok(Err(e)) => return Err(Error::Transport(e.into())),
David's avatar
David committed
178
		};
179

David's avatar
David committed
180
		let rps: Vec<Response<_>> = match serde_json::from_slice(&body) {
181
182
			Ok(response) => response,
			Err(_) => {
183
184
185
186
				let err: RpcError = serde_json::from_slice(&body).map_err(|e| {
					self.id_guard.reclaim_request_id();
					Error::ParseError(e)
				})?;
187
				return Err(Error::Request(err.to_string()));
188
			}
189
190
191
192
193
		};

		// NOTE: `R::default` is placeholder and will be replaced in loop below.
		let mut responses = vec![R::default(); ordered_requests.len()];
		for rp in rps {
194
			let response_id = rp.id.as_number().copied().ok_or(Error::InvalidRequestId)?;
195
196
197
198
199
200
201
202
203
			let pos = match request_set.get(&response_id) {
				Some(pos) => *pos,
				None => return Err(Error::InvalidRequestId),
			};
			responses[pos] = rp.result
		}
		Ok(responses)
	}
}