_ Git - rust-lightning/blob - lightning-block-sync/src/http.rs

   1 //! Simple HTTP implementation which supports both async and traditional execution environments
   2 //! with minimal dependencies. This is used as the basis for REST and RPC clients.
   3
   4 use chunked_transfer;
   5 use serde_json;
   6
   7 use std::convert::TryFrom;
   8 #[cfg(not(feature = "tokio"))]
   9 use std::io::Write;
  10 use std::net::ToSocketAddrs;
  11 use std::time::Duration;
  12
  13 #[cfg(feature = "tokio")]
  14 use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt};
  15 #[cfg(feature = "tokio")]
  16 use tokio::net::TcpStream;
  17
  18 #[cfg(not(feature = "tokio"))]
  19 use std::io::BufRead;
  20 use std::io::Read;
  21 #[cfg(not(feature = "tokio"))]
  22 use std::net::TcpStream;
  23
  24 /// Timeout for operations on TCP streams.
  25 const TCP_STREAM_TIMEOUT: Duration = Duration::from_secs(5);
  26
  27 /// Maximum HTTP message header size in bytes.
  28 const MAX_HTTP_MESSAGE_HEADER_SIZE: usize = 8192;
  29
  30 /// Maximum HTTP message body size in bytes. Enough for a hex-encoded block in JSON format and any
  31 /// overhead for HTTP chunked transfer encoding.
  32 const MAX_HTTP_MESSAGE_BODY_SIZE: usize = 2 * 4_000_000 + 32_000;
  33
  34 /// Endpoint for interacting with an HTTP-based API.
  35 #[derive(Debug)]
  36 pub struct HttpEndpoint {
  37         host: String,
  38         port: Option<u16>,
  39         path: String,
  40 }
  41
  42 impl HttpEndpoint {
  43         /// Creates an endpoint for the given host and default HTTP port.
  44         pub fn for_host(host: String) -> Self {
  45                 Self {
  46                         host,
  47                         port: None,
  48                         path: String::from("/"),
  49                 }
  50         }
  51
  52         /// Specifies a port to use with the endpoint.
  53         pub fn with_port(mut self, port: u16) -> Self {
  54                 self.port = Some(port);
  55                 self
  56         }
  57
  58         /// Specifies a path to use with the endpoint.
  59         pub fn with_path(mut self, path: String) -> Self {
  60                 self.path = path;
  61                 self
  62         }
  63
  64         /// Returns the endpoint host.
  65         pub fn host(&self) -> &str {
  66                 &self.host
  67         }
  68
  69         /// Returns the endpoint port.
  70         pub fn port(&self) -> u16 {
  71                 match self.port {
  72                         None => 80,
  73                         Some(port) => port,
  74                 }
  75         }
  76
  77         /// Returns the endpoint path.
  78         pub fn path(&self) -> &str {
  79                 &self.path
  80         }
  81 }
  82
  83 impl<'a> std::net::ToSocketAddrs for &'a HttpEndpoint {
  84         type Iter = <(&'a str, u16) as std::net::ToSocketAddrs>::Iter;
  85
  86         fn to_socket_addrs(&self) -> std::io::Result<Self::Iter> {
  87                 (self.host(), self.port()).to_socket_addrs()
  88         }
  89 }
  90
  91 /// Client for making HTTP requests.
  92 pub(crate) struct HttpClient {
  93         stream: TcpStream,
  94 }
  95
  96 impl HttpClient {
  97         /// Opens a connection to an HTTP endpoint.
  98         pub fn connect<E: ToSocketAddrs>(endpoint: E) -> std::io::Result<Self> {
  99                 let address = match endpoint.to_socket_addrs()?.next() {
 100                         None => {
 101                                 return Err(std::io::Error::new(std::io::ErrorKind::InvalidInput, "could not resolve to any addresses"));
 102                         },
 103                         Some(address) => address,
 104                 };
 105                 let stream = std::net::TcpStream::connect_timeout(&address, TCP_STREAM_TIMEOUT)?;
 106                 stream.set_read_timeout(Some(TCP_STREAM_TIMEOUT))?;
 107                 stream.set_write_timeout(Some(TCP_STREAM_TIMEOUT))?;
 108
 109                 #[cfg(feature = "tokio")]
 110                 let stream = {
 111                         stream.set_nonblocking(true)?;
 112                         TcpStream::from_std(stream)?
 113                 };
 114
 115                 Ok(Self { stream })
 116         }
 117
 118         /// Sends a `GET` request for a resource identified by `uri` at the `host`.
 119         ///
 120         /// Returns the response body in `F` format.
 121         #[allow(dead_code)]
 122         pub async fn get<F>(&mut self, uri: &str, host: &str) -> std::io::Result<F>
 123         where F: TryFrom<Vec<u8>, Error = std::io::Error> {
 124                 let request = format!(
 125                         "GET {} HTTP/1.1\r\n\
 126                          Host: {}\r\n\
 127                          Connection: keep-alive\r\n\
 128                          \r\n", uri, host);
 129                 let response_body = self.send_request_with_retry(&request).await?;
 130                 F::try_from(response_body)
 131         }
 132
 133         /// Sends a `POST` request for a resource identified by `uri` at the `host` using the given HTTP
 134         /// authentication credentials.
 135         ///
 136         /// The request body consists of the provided JSON `content`. Returns the response body in `F`
 137         /// format.
 138         #[allow(dead_code)]
 139         pub async fn post<F>(&mut self, uri: &str, host: &str, auth: &str, content: serde_json::Value) -> std::io::Result<F>
 140         where F: TryFrom<Vec<u8>, Error = std::io::Error> {
 141                 let content = content.to_string();
 142                 let request = format!(
 143                         "POST {} HTTP/1.1\r\n\
 144                          Host: {}\r\n\
 145                          Authorization: {}\r\n\
 146                          Connection: keep-alive\r\n\
 147                          Content-Type: application/json\r\n\
 148                          Content-Length: {}\r\n\
 149                          \r\n\
 150                          {}", uri, host, auth, content.len(), content);
 151                 let response_body = self.send_request_with_retry(&request).await?;
 152                 F::try_from(response_body)
 153         }
 154
 155         /// Sends an HTTP request message and reads the response, returning its body. Attempts to
 156         /// reconnect and retry if the connection has been closed.
 157         async fn send_request_with_retry(&mut self, request: &str) -> std::io::Result<Vec<u8>> {
 158                 let endpoint = self.stream.peer_addr().unwrap();
 159                 match self.send_request(request).await {
 160                         Ok(bytes) => Ok(bytes),
 161                         Err(e) => match e.kind() {
 162                                 std::io::ErrorKind::ConnectionReset |
 163                                 std::io::ErrorKind::ConnectionAborted |
 164                                 std::io::ErrorKind::UnexpectedEof => {
 165                                         // Reconnect if the connection was closed. This may happen if the server's
 166                                         // keep-alive limits are reached.
 167                                         *self = Self::connect(endpoint)?;
 168                                         self.send_request(request).await
 169                                 },
 170                                 _ => Err(e),
 171                         },
 172                 }
 173         }
 174
 175         /// Sends an HTTP request message and reads the response, returning its body.
 176         async fn send_request(&mut self, request: &str) -> std::io::Result<Vec<u8>> {
 177                 self.write_request(request).await?;
 178                 self.read_response().await
 179         }
 180
 181         /// Writes an HTTP request message.
 182         async fn write_request(&mut self, request: &str) -> std::io::Result<()> {
 183                 #[cfg(feature = "tokio")]
 184                 {
 185                         self.stream.write_all(request.as_bytes()).await?;
 186                         self.stream.flush().await
 187                 }
 188                 #[cfg(not(feature = "tokio"))]
 189                 {
 190                         self.stream.write_all(request.as_bytes())?;
 191                         self.stream.flush()
 192                 }
 193         }
 194
 195         /// Reads an HTTP response message.
 196         async fn read_response(&mut self) -> std::io::Result<Vec<u8>> {
 197                 #[cfg(feature = "tokio")]
 198                 let stream = self.stream.split().0;
 199                 #[cfg(not(feature = "tokio"))]
 200                 let stream = std::io::Read::by_ref(&mut self.stream);
 201
 202                 let limited_stream = stream.take(MAX_HTTP_MESSAGE_HEADER_SIZE as u64);
 203
 204                 #[cfg(feature = "tokio")]
 205                 let mut reader = tokio::io::BufReader::new(limited_stream);
 206                 #[cfg(not(feature = "tokio"))]
 207                 let mut reader = std::io::BufReader::new(limited_stream);
 208
 209                 macro_rules! read_line { () => { {
 210                         let mut line = String::new();
 211                         #[cfg(feature = "tokio")]
 212                         let bytes_read = reader.read_line(&mut line).await?;
 213                         #[cfg(not(feature = "tokio"))]
 214                         let bytes_read = reader.read_line(&mut line)?;
 215
 216                         match bytes_read {
 217                                 0 => None,
 218                                 _ => {
 219                                         // Remove trailing CRLF
 220                                         if line.ends_with('\n') { line.pop(); if line.ends_with('\r') { line.pop(); } }
 221                                         Some(line)
 222                                 },
 223                         }
 224                 } } }
 225
 226                 // Read and parse status line
 227                 let status_line = read_line!()
 228                         .ok_or(std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "no status line"))?;
 229                 let status = HttpStatus::parse(&status_line)?;
 230
 231                 // Read and parse relevant headers
 232                 let mut message_length = HttpMessageLength::Empty;
 233                 loop {
 234                         let line = read_line!()
 235                                 .ok_or(std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "no headers"))?;
 236                         if line.is_empty() { break; }
 237
 238                         let header = HttpHeader::parse(&line)?;
 239                         if header.has_name("Content-Length") {
 240                                 let length = header.value.parse()
 241                                         .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
 242                                 if let HttpMessageLength::Empty = message_length {
 243                                         message_length = HttpMessageLength::ContentLength(length);
 244                                 }
 245                                 continue;
 246                         }
 247
 248                         if header.has_name("Transfer-Encoding") {
 249                                 message_length = HttpMessageLength::TransferEncoding(header.value.into());
 250                                 continue;
 251                         }
 252                 }
 253
 254                 if !status.is_ok() {
 255                         // TODO: Handle 3xx redirection responses.
 256                         return Err(std::io::Error::new(std::io::ErrorKind::NotFound, "not found"));
 257                 }
 258
 259                 // Read message body
 260                 let read_limit = MAX_HTTP_MESSAGE_BODY_SIZE - reader.buffer().len();
 261                 reader.get_mut().set_limit(read_limit as u64);
 262                 match message_length {
 263                         HttpMessageLength::Empty => { Ok(Vec::new()) },
 264                         HttpMessageLength::ContentLength(length) => {
 265                                 if length == 0 || length > MAX_HTTP_MESSAGE_BODY_SIZE {
 266                                         Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "out of range"))
 267                                 } else {
 268                                         let mut content = vec![0; length];
 269                                         #[cfg(feature = "tokio")]
 270                                         reader.read_exact(&mut content[..]).await?;
 271                                         #[cfg(not(feature = "tokio"))]
 272                                         reader.read_exact(&mut content[..])?;
 273                                         Ok(content)
 274                                 }
 275                         },
 276                         HttpMessageLength::TransferEncoding(coding) => {
 277                                 if !coding.eq_ignore_ascii_case("chunked") {
 278                                         Err(std::io::Error::new(
 279                                                         std::io::ErrorKind::InvalidInput, "unsupported transfer coding"))
 280                                 } else {
 281                                         let mut content = Vec::new();
 282                                         #[cfg(feature = "tokio")]
 283                                         {
 284                                                 // Since chunked_transfer doesn't have an async interface, only use it to
 285                                                 // determine the size of each chunk to read.
 286                                                 //
 287                                                 // TODO: Replace with an async interface when available.
 288                                                 // https://github.com/frewsxcv/rust-chunked-transfer/issues/7
 289                                                 loop {
 290                                                         // Read the chunk header which contains the chunk size.
 291                                                         let mut chunk_header = String::new();
 292                                                         reader.read_line(&mut chunk_header).await?;
 293                                                         if chunk_header == "0\r\n" {
 294                                                                 // Read the terminator chunk since the decoder consumes the CRLF
 295                                                                 // immediately when this chunk is encountered.
 296                                                                 reader.read_line(&mut chunk_header).await?;
 297                                                         }
 298
 299                                                         // Decode the chunk header to obtain the chunk size.
 300                                                         let mut buffer = Vec::new();
 301                                                         let mut decoder = chunked_transfer::Decoder::new(chunk_header.as_bytes());
 302                                                         decoder.read_to_end(&mut buffer)?;
 303
 304                                                         // Read the chunk body.
 305                                                         let chunk_size = match decoder.remaining_chunks_size() {
 306                                                                 None => break,
 307                                                                 Some(chunk_size) => chunk_size,
 308                                                         };
 309                                                         let chunk_offset = content.len();
 310                                                         content.resize(chunk_offset + chunk_size + "\r\n".len(), 0);
 311                                                         reader.read_exact(&mut content[chunk_offset..]).await?;
 312                                                         content.resize(chunk_offset + chunk_size, 0);
 313                                                 }
 314                                                 Ok(content)
 315                                         }
 316                                         #[cfg(not(feature = "tokio"))]
 317                                         {
 318                                                 let mut decoder = chunked_transfer::Decoder::new(reader);
 319                                                 decoder.read_to_end(&mut content)?;
 320                                                 Ok(content)
 321                                         }
 322                                 }
 323                         },
 324                 }
 325         }
 326 }
 327
 328 /// HTTP response status code as defined by [RFC 7231].
 329 ///
 330 /// [RFC 7231]: https://tools.ietf.org/html/rfc7231#section-6
 331 struct HttpStatus<'a> {
 332         code: &'a str,
 333 }
 334
 335 impl<'a> HttpStatus<'a> {
 336         /// Parses an HTTP status line as defined by [RFC 7230].
 337         ///
 338         /// [RFC 7230]: https://tools.ietf.org/html/rfc7230#section-3.1.2
 339         fn parse(line: &'a String) -> std::io::Result<HttpStatus<'a>> {
 340                 let mut tokens = line.splitn(3, ' ');
 341
 342                 let http_version = tokens.next()
 343                         .ok_or(std::io::Error::new(std::io::ErrorKind::InvalidData, "no HTTP-Version"))?;
 344                 if !http_version.eq_ignore_ascii_case("HTTP/1.1") &&
 345                         !http_version.eq_ignore_ascii_case("HTTP/1.0") {
 346                         return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "invalid HTTP-Version"));
 347                 }
 348
 349                 let code = tokens.next()
 350                         .ok_or(std::io::Error::new(std::io::ErrorKind::InvalidData, "no Status-Code"))?;
 351                 if code.len() != 3 || !code.chars().all(|c| c.is_ascii_digit()) {
 352                         return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "invalid Status-Code"));
 353                 }
 354
 355                 let _reason = tokens.next()
 356                         .ok_or(std::io::Error::new(std::io::ErrorKind::InvalidData, "no Reason-Phrase"))?;
 357
 358                 Ok(Self { code })
 359         }
 360
 361         /// Returns whether the status is successful (i.e., 2xx status class).
 362         fn is_ok(&self) -> bool {
 363                 self.code.starts_with('2')
 364         }
 365 }
 366
 367 /// HTTP response header as defined by [RFC 7231].
 368 ///
 369 /// [RFC 7231]: https://tools.ietf.org/html/rfc7231#section-7
 370 struct HttpHeader<'a> {
 371         name: &'a str,
 372         value: &'a str,
 373 }
 374
 375 impl<'a> HttpHeader<'a> {
 376         /// Parses an HTTP header field as defined by [RFC 7230].
 377         ///
 378         /// [RFC 7230]: https://tools.ietf.org/html/rfc7230#section-3.2
 379         fn parse(line: &'a String) -> std::io::Result<HttpHeader<'a>> {
 380                 let mut tokens = line.splitn(2, ':');
 381                 let name = tokens.next()
 382                         .ok_or(std::io::Error::new(std::io::ErrorKind::InvalidData, "no header name"))?;
 383                 let value = tokens.next()
 384                         .ok_or(std::io::Error::new(std::io::ErrorKind::InvalidData, "no header value"))?
 385                         .trim_start();
 386                 Ok(Self { name, value })
 387         }
 388
 389         /// Returns whether the header field has the given name.
 390         fn has_name(&self, name: &str) -> bool {
 391                 self.name.eq_ignore_ascii_case(name)
 392         }
 393 }
 394
 395 /// HTTP message body length as defined by [RFC 7230].
 396 ///
 397 /// [RFC 7230]: https://tools.ietf.org/html/rfc7230#section-3.3.3
 398 enum HttpMessageLength {
 399         Empty,
 400         ContentLength(usize),
 401         TransferEncoding(String),
 402 }
 403
 404 /// An HTTP response body in binary format.
 405 pub struct BinaryResponse(pub Vec<u8>);
 406
 407 /// An HTTP response body in JSON format.
 408 pub struct JsonResponse(pub serde_json::Value);
 409
 410 /// Interprets bytes from an HTTP response body as binary data.
 411 impl TryFrom<Vec<u8>> for BinaryResponse {
 412         type Error = std::io::Error;
 413
 414         fn try_from(bytes: Vec<u8>) -> std::io::Result<Self> {
 415                 Ok(BinaryResponse(bytes))
 416         }
 417 }
 418
 419 /// Interprets bytes from an HTTP response body as a JSON value.
 420 impl TryFrom<Vec<u8>> for JsonResponse {
 421         type Error = std::io::Error;
 422
 423         fn try_from(bytes: Vec<u8>) -> std::io::Result<Self> {
 424                 Ok(JsonResponse(serde_json::from_slice(&bytes)?))
 425         }
 426 }
 427
 428 #[cfg(test)]
 429 mod endpoint_tests {
 430         use super::HttpEndpoint;
 431
 432         #[test]
 433         fn with_default_port() {
 434                 let endpoint = HttpEndpoint::for_host("foo.com".into());
 435                 assert_eq!(endpoint.host(), "foo.com");
 436                 assert_eq!(endpoint.port(), 80);
 437         }
 438
 439         #[test]
 440         fn with_custom_port() {
 441                 let endpoint = HttpEndpoint::for_host("foo.com".into()).with_port(8080);
 442                 assert_eq!(endpoint.host(), "foo.com");
 443                 assert_eq!(endpoint.port(), 8080);
 444         }
 445
 446         #[test]
 447         fn with_uri_path() {
 448                 let endpoint = HttpEndpoint::for_host("foo.com".into()).with_path("/path".into());
 449                 assert_eq!(endpoint.host(), "foo.com");
 450                 assert_eq!(endpoint.path(), "/path");
 451         }
 452
 453         #[test]
 454         fn without_uri_path() {
 455                 let endpoint = HttpEndpoint::for_host("foo.com".into());
 456                 assert_eq!(endpoint.host(), "foo.com");
 457                 assert_eq!(endpoint.path(), "/");
 458         }
 459
 460         #[test]
 461         fn convert_to_socket_addrs() {
 462                 let endpoint = HttpEndpoint::for_host("foo.com".into());
 463                 let host = endpoint.host();
 464                 let port = endpoint.port();
 465
 466                 use std::net::ToSocketAddrs;
 467                 match (&endpoint).to_socket_addrs() {
 468                         Err(e) => panic!("Unexpected error: {:?}", e),
 469                         Ok(mut socket_addrs) => {
 470                                 match socket_addrs.next() {
 471                                         None => panic!("Expected socket address"),
 472                                         Some(addr) => {
 473                                                 assert_eq!(addr, (host, port).to_socket_addrs().unwrap().next().unwrap());
 474                                                 assert!(socket_addrs.next().is_none());
 475                                         }
 476                                 }
 477                         }
 478                 }
 479         }
 480 }
 481
 482 #[cfg(test)]
 483 pub(crate) mod client_tests {
 484         use super::*;
 485         use std::io::BufRead;
 486         use std::io::Write;
 487
 488         /// Server for handling HTTP client requests with a stock response.
 489         pub struct HttpServer {
 490                 address: std::net::SocketAddr,
 491                 handler: std::thread::JoinHandle<()>,
 492                 shutdown: std::sync::Arc<std::sync::atomic::AtomicBool>,
 493         }
 494
 495         /// Body of HTTP response messages.
 496         pub enum MessageBody<T: ToString> {
 497                 Empty,
 498                 Content(T),
 499                 ChunkedContent(T),
 500         }
 501
 502         impl HttpServer {
 503                 pub fn responding_with_ok<T: ToString>(body: MessageBody<T>) -> Self {
 504                         let response = match body {
 505                                 MessageBody::Empty => "HTTP/1.1 200 OK\r\n\r\n".to_string(),
 506                                 MessageBody::Content(body) => {
 507                                         let body = body.to_string();
 508                                         format!(
 509                                                 "HTTP/1.1 200 OK\r\n\
 510                                                  Content-Length: {}\r\n\
 511                                                  \r\n\
 512                                                  {}", body.len(), body)
 513                                 },
 514                                 MessageBody::ChunkedContent(body) => {
 515                                         let mut chuncked_body = Vec::new();
 516                                         {
 517                                                 use chunked_transfer::Encoder;
 518                                                 let mut encoder = Encoder::with_chunks_size(&mut chuncked_body, 8);
 519                                                 encoder.write_all(body.to_string().as_bytes()).unwrap();
 520                                         }
 521                                         format!(
 522                                                 "HTTP/1.1 200 OK\r\n\
 523                                                  Transfer-Encoding: chunked\r\n\
 524                                                  \r\n\
 525                                                  {}", String::from_utf8(chuncked_body).unwrap())
 526                                 },
 527                         };
 528                         HttpServer::responding_with(response)
 529                 }
 530
 531                 pub fn responding_with_not_found() -> Self {
 532                         let response = "HTTP/1.1 404 Not Found\r\n\r\n".to_string();
 533                         HttpServer::responding_with(response)
 534                 }
 535
 536                 fn responding_with(response: String) -> Self {
 537                         let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
 538                         let address = listener.local_addr().unwrap();
 539
 540                         let shutdown = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
 541                         let shutdown_signaled = std::sync::Arc::clone(&shutdown);
 542                         let handler = std::thread::spawn(move || {
 543                                 for stream in listener.incoming() {
 544                                         let mut stream = stream.unwrap();
 545                                         stream.set_write_timeout(Some(TCP_STREAM_TIMEOUT)).unwrap();
 546
 547                                         let lines_read = std::io::BufReader::new(&stream)
 548                                                 .lines()
 549                                                 .take_while(|line| !line.as_ref().unwrap().is_empty())
 550                                                 .count();
 551                                         if lines_read == 0 { continue; }
 552
 553                                         for chunk in response.as_bytes().chunks(16) {
 554                                                 if shutdown_signaled.load(std::sync::atomic::Ordering::SeqCst) {
 555                                                         return;
 556                                                 } else {
 557                                                         if let Err(_) = stream.write(chunk) { break; }
 558                                                         if let Err(_) = stream.flush() { break; }
 559                                                 }
 560                                         }
 561                                 }
 562                         });
 563
 564                         Self { address, handler, shutdown }
 565                 }
 566
 567                 fn shutdown(self) {
 568                         self.shutdown.store(true, std::sync::atomic::Ordering::SeqCst);
 569                         self.handler.join().unwrap();
 570                 }
 571
 572                 pub fn endpoint(&self) -> HttpEndpoint {
 573                         HttpEndpoint::for_host(self.address.ip().to_string()).with_port(self.address.port())
 574                 }
 575         }
 576
 577         #[test]
 578         fn connect_to_unresolvable_host() {
 579                 match HttpClient::connect(("example.invalid", 80)) {
 580                         Err(e) => assert_eq!(e.kind(), std::io::ErrorKind::Other),
 581                         Ok(_) => panic!("Expected error"),
 582                 }
 583         }
 584
 585         #[test]
 586         fn connect_with_no_socket_address() {
 587                 match HttpClient::connect(&vec![][..]) {
 588                         Err(e) => assert_eq!(e.kind(), std::io::ErrorKind::InvalidInput),
 589                         Ok(_) => panic!("Expected error"),
 590                 }
 591         }
 592
 593         #[test]
 594         fn connect_with_unknown_server() {
 595                 match HttpClient::connect(("::", 80)) {
 596                         #[cfg(target_os = "windows")]
 597                         Err(e) => assert_eq!(e.kind(), std::io::ErrorKind::AddrNotAvailable),
 598                         #[cfg(not(target_os = "windows"))]
 599                         Err(e) => assert_eq!(e.kind(), std::io::ErrorKind::ConnectionRefused),
 600                         Ok(_) => panic!("Expected error"),
 601                 }
 602         }
 603
 604         #[tokio::test]
 605         async fn connect_with_valid_endpoint() {
 606                 let server = HttpServer::responding_with_ok::<String>(MessageBody::Empty);
 607
 608                 match HttpClient::connect(&server.endpoint()) {
 609                         Err(e) => panic!("Unexpected error: {:?}", e),
 610                         Ok(_) => {},
 611                 }
 612         }
 613
 614         #[tokio::test]
 615         async fn read_empty_message() {
 616                 let server = HttpServer::responding_with("".to_string());
 617
 618                 let mut client = HttpClient::connect(&server.endpoint()).unwrap();
 619                 match client.get::<BinaryResponse>("/foo", "foo.com").await {
 620                         Err(e) => {
 621                                 assert_eq!(e.kind(), std::io::ErrorKind::UnexpectedEof);
 622                                 assert_eq!(e.get_ref().unwrap().to_string(), "no status line");
 623                         },
 624                         Ok(_) => panic!("Expected error"),
 625                 }
 626         }
 627
 628         #[tokio::test]
 629         async fn read_incomplete_message() {
 630                 let server = HttpServer::responding_with("HTTP/1.1 200 OK".to_string());
 631
 632                 let mut client = HttpClient::connect(&server.endpoint()).unwrap();
 633                 match client.get::<BinaryResponse>("/foo", "foo.com").await {
 634                         Err(e) => {
 635                                 assert_eq!(e.kind(), std::io::ErrorKind::UnexpectedEof);
 636                                 assert_eq!(e.get_ref().unwrap().to_string(), "no headers");
 637                         },
 638                         Ok(_) => panic!("Expected error"),
 639                 }
 640         }
 641
 642         #[tokio::test]
 643         async fn read_too_large_message_headers() {
 644                 let response = format!(
 645                         "HTTP/1.1 302 Found\r\n\
 646                          Location: {}\r\n\
 647                          \r\n", "Z".repeat(MAX_HTTP_MESSAGE_HEADER_SIZE));
 648                 let server = HttpServer::responding_with(response);
 649
 650                 let mut client = HttpClient::connect(&server.endpoint()).unwrap();
 651                 match client.get::<BinaryResponse>("/foo", "foo.com").await {
 652                         Err(e) => {
 653                                 assert_eq!(e.kind(), std::io::ErrorKind::UnexpectedEof);
 654                                 assert_eq!(e.get_ref().unwrap().to_string(), "no headers");
 655                         },
 656                         Ok(_) => panic!("Expected error"),
 657                 }
 658         }
 659
 660         #[tokio::test]
 661         async fn read_too_large_message_body() {
 662                 let body = "Z".repeat(MAX_HTTP_MESSAGE_BODY_SIZE + 1);
 663                 let server = HttpServer::responding_with_ok::<String>(MessageBody::Content(body));
 664
 665                 let mut client = HttpClient::connect(&server.endpoint()).unwrap();
 666                 match client.get::<BinaryResponse>("/foo", "foo.com").await {
 667                         Err(e) => {
 668                                 assert_eq!(e.kind(), std::io::ErrorKind::InvalidData);
 669                                 assert_eq!(e.get_ref().unwrap().to_string(), "out of range");
 670                         },
 671                         Ok(_) => panic!("Expected error"),
 672                 }
 673                 server.shutdown();
 674         }
 675
 676         #[tokio::test]
 677         async fn read_message_with_unsupported_transfer_coding() {
 678                 let response = String::from(
 679                         "HTTP/1.1 200 OK\r\n\
 680                          Transfer-Encoding: gzip\r\n\
 681                          \r\n\
 682                          foobar");
 683                 let server = HttpServer::responding_with(response);
 684
 685                 let mut client = HttpClient::connect(&server.endpoint()).unwrap();
 686                 match client.get::<BinaryResponse>("/foo", "foo.com").await {
 687                         Err(e) => {
 688                                 assert_eq!(e.kind(), std::io::ErrorKind::InvalidInput);
 689                                 assert_eq!(e.get_ref().unwrap().to_string(), "unsupported transfer coding");
 690                         },
 691                         Ok(_) => panic!("Expected error"),
 692                 }
 693         }
 694
 695         #[tokio::test]
 696         async fn read_empty_message_body() {
 697                 let server = HttpServer::responding_with_ok::<String>(MessageBody::Empty);
 698
 699                 let mut client = HttpClient::connect(&server.endpoint()).unwrap();
 700                 match client.get::<BinaryResponse>("/foo", "foo.com").await {
 701                         Err(e) => panic!("Unexpected error: {:?}", e),
 702                         Ok(bytes) => assert_eq!(bytes.0, Vec::<u8>::new()),
 703                 }
 704         }
 705
 706         #[tokio::test]
 707         async fn read_message_body_with_length() {
 708                 let body = "foo bar baz qux".repeat(32);
 709                 let content = MessageBody::Content(body.clone());
 710                 let server = HttpServer::responding_with_ok::<String>(content);
 711
 712                 let mut client = HttpClient::connect(&server.endpoint()).unwrap();
 713                 match client.get::<BinaryResponse>("/foo", "foo.com").await {
 714                         Err(e) => panic!("Unexpected error: {:?}", e),
 715                         Ok(bytes) => assert_eq!(bytes.0, body.as_bytes()),
 716                 }
 717         }
 718
 719         #[tokio::test]
 720         async fn read_chunked_message_body() {
 721                 let body = "foo bar baz qux".repeat(32);
 722                 let chunked_content = MessageBody::ChunkedContent(body.clone());
 723                 let server = HttpServer::responding_with_ok::<String>(chunked_content);
 724
 725                 let mut client = HttpClient::connect(&server.endpoint()).unwrap();
 726                 match client.get::<BinaryResponse>("/foo", "foo.com").await {
 727                         Err(e) => panic!("Unexpected error: {:?}", e),
 728                         Ok(bytes) => assert_eq!(bytes.0, body.as_bytes()),
 729                 }
 730         }
 731
 732         #[tokio::test]
 733         async fn reconnect_closed_connection() {
 734                 let server = HttpServer::responding_with_ok::<String>(MessageBody::Empty);
 735
 736                 let mut client = HttpClient::connect(&server.endpoint()).unwrap();
 737                 assert!(client.get::<BinaryResponse>("/foo", "foo.com").await.is_ok());
 738                 match client.get::<BinaryResponse>("/foo", "foo.com").await {
 739                         Err(e) => panic!("Unexpected error: {:?}", e),
 740                         Ok(bytes) => assert_eq!(bytes.0, Vec::<u8>::new()),
 741                 }
 742         }
 743
 744         #[test]
 745         fn from_bytes_into_binary_response() {
 746                 let bytes = b"foo";
 747                 match BinaryResponse::try_from(bytes.to_vec()) {
 748                         Err(e) => panic!("Unexpected error: {:?}", e),
 749                         Ok(response) => assert_eq!(&response.0, bytes),
 750                 }
 751         }
 752
 753         #[test]
 754         fn from_invalid_bytes_into_json_response() {
 755                 let json = serde_json::json!({ "result": 42 });
 756                 match JsonResponse::try_from(json.to_string().as_bytes()[..5].to_vec()) {
 757                         Err(_) => {},
 758                         Ok(_) => panic!("Expected error"),
 759                 }
 760         }
 761
 762         #[test]
 763         fn from_valid_bytes_into_json_response() {
 764                 let json = serde_json::json!({ "result": 42 });
 765                 match JsonResponse::try_from(json.to_string().as_bytes().to_vec()) {
 766                         Err(e) => panic!("Unexpected error: {:?}", e),
 767                         Ok(response) => assert_eq!(response.0, json),
 768                 }
 769         }
 770 }