17229631411

Committed 26 Aug 2025 06:13AM UTC coverage: 51.84% (-2.1%) from 53.937%

Build # 17229631411

Build Type

push

github

Committed by

web-flow

Commit Message

fix: remove another bottleneck that causes boot time spike (#596)

* fix: remove another bottleneck that causes boot time spike

* chore: add integration test

Run Details

28 of 33 new or added lines in 1 file covered. (84.85%)

4922 existing lines in 74 files now uncovered.

18444 of 35579 relevant lines covered (51.84%)

5545.51 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

54.23

/ext/node/ops/http2.rs

// Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.

use std::borrow::Cow;
use std::cell::RefCell;
use std::collections::HashMap;
use std::rc::Rc;
use std::task::Poll;

use bytes::Bytes;
use deno_core::futures::future::poll_fn;
use deno_core::op2;
use deno_core::serde::Serialize;
use deno_core::AsyncRefCell;
use deno_core::BufView;
use deno_core::ByteString;
use deno_core::CancelFuture;
use deno_core::CancelHandle;
use deno_core::JsBuffer;
use deno_core::OpState;
use deno_core::RcRef;
use deno_core::Resource;
use deno_core::ResourceId;
use deno_net::raw::take_network_stream_resource;
use deno_net::raw::NetworkStream;
use h2;
use h2::Reason;
use h2::RecvStream;
use http;
use http::header::HeaderName;
use http::header::HeaderValue;
use http::request::Parts;
use http::HeaderMap;
use http::Response;
use http::StatusCode;
use url::Url;

pub struct Http2Client {
  pub client: AsyncRefCell<h2::client::SendRequest<BufView>>,
  pub url: Url,
}

impl Resource for Http2Client {
  fn name(&self) -> Cow<str> {
    "http2Client".into()
  }
}

#[derive(Debug)]
pub struct Http2ClientConn {
  pub conn: AsyncRefCell<h2::client::Connection<NetworkStream, BufView>>,
  cancel_handle: CancelHandle,
}

impl Resource for Http2ClientConn {
  fn name(&self) -> Cow<str> {
    "http2ClientConnection".into()
  }

  fn close(self: Rc<Self>) {
    self.cancel_handle.cancel()
  }
}

#[derive(Debug)]
pub struct Http2ClientStream {
  pub response: AsyncRefCell<h2::client::ResponseFuture>,
  pub stream: AsyncRefCell<h2::SendStream<BufView>>,
}

impl Resource for Http2ClientStream {
  fn name(&self) -> Cow<str> {
    "http2ClientStream".into()
  }
}

#[derive(Debug)]
pub struct Http2ClientResponseBody {
  pub body: AsyncRefCell<h2::RecvStream>,
  pub trailers_rx:
    AsyncRefCell<Option<tokio::sync::oneshot::Receiver<Option<HeaderMap>>>>,
  pub trailers_tx:
    AsyncRefCell<Option<tokio::sync::oneshot::Sender<Option<HeaderMap>>>>,
}

impl Resource for Http2ClientResponseBody {
  fn name(&self) -> Cow<str> {
    "http2ClientResponseBody".into()
  }
}

#[derive(Debug)]
pub struct Http2ServerConnection {
  pub conn: AsyncRefCell<h2::server::Connection<NetworkStream, BufView>>,
}

impl Resource for Http2ServerConnection {
  fn name(&self) -> Cow<str> {
    "http2ServerConnection".into()
  }
}

pub struct Http2ServerSendResponse {
  pub send_response: AsyncRefCell<h2::server::SendResponse<BufView>>,
}

impl Resource for Http2ServerSendResponse {
  fn name(&self) -> Cow<str> {
    "http2ServerSendResponse".into()
  }
}

#[derive(Debug, thiserror::Error)]
pub enum Http2Error {
  #[error(transparent)]
  Resource(deno_core::error::AnyError),
  #[error(transparent)]
  UrlParse(#[from] url::ParseError),
  #[error(transparent)]
  H2(#[from] h2::Error),
}

#[op2(async)]
#[serde]
pub async fn op_http2_connect(
  state: Rc<RefCell<OpState>>,
  #[smi] rid: ResourceId,
  #[string] url: String,
) -> Result<(ResourceId, ResourceId), Http2Error> {
  // No permission check necessary because we're using an existing connection
  let network_stream = {
    let mut state = state.borrow_mut();
    take_network_stream_resource(&mut state.resource_table, rid)
      .map_err(Http2Error::Resource)?
  };

  let url = Url::parse(&url)?;

  let (client, conn) =
    h2::client::Builder::new().handshake(network_stream).await?;
  let mut state = state.borrow_mut();
  let client_rid = state.resource_table.add(Http2Client {
    client: AsyncRefCell::new(client),
    url,
  });
  let conn_rid = state.resource_table.add(Http2ClientConn {
    conn: AsyncRefCell::new(conn),
    cancel_handle: CancelHandle::new(),
  });
  Ok((client_rid, conn_rid))
}

#[op2(async)]
#[smi]
pub async fn op_http2_listen(
  state: Rc<RefCell<OpState>>,
  #[smi] rid: ResourceId,
) -> Result<ResourceId, Http2Error> {
  let stream =
    take_network_stream_resource(&mut state.borrow_mut().resource_table, rid)
      .map_err(Http2Error::Resource)?;

  let conn = h2::server::Builder::new().handshake(stream).await?;
  Ok(
    state
      .borrow_mut()
      .resource_table
      .add(Http2ServerConnection {
        conn: AsyncRefCell::new(conn),
      }),
  )
}

#[op2(async)]
#[serde]
pub async fn op_http2_accept(
  state: Rc<RefCell<OpState>>,
  #[smi] rid: ResourceId,
) -> Result<
  Option<(Vec<(ByteString, ByteString)>, ResourceId, ResourceId)>,
  Http2Error,
> {
  let resource = state
    .borrow()
    .resource_table
    .get::<Http2ServerConnection>(rid)
    .map_err(Http2Error::Resource)?;
  let mut conn = RcRef::map(&resource, |r| &r.conn).borrow_mut().await;
  if let Some(res) = conn.accept().await {
    let (req, resp) = res?;
    let (parts, body) = req.into_parts();
    let (trailers_tx, trailers_rx) = tokio::sync::oneshot::channel();
    let stm = state
      .borrow_mut()
      .resource_table
      .add(Http2ClientResponseBody {
        body: AsyncRefCell::new(body),
        trailers_rx: AsyncRefCell::new(Some(trailers_rx)),
        trailers_tx: AsyncRefCell::new(Some(trailers_tx)),
      });

    let Parts {
      uri,
      method,
      headers,
      ..
    } = parts;
    let mut req_headers = Vec::with_capacity(headers.len() + 4);
    req_headers.push((
      ByteString::from(":method"),
      ByteString::from(method.as_str()),
    ));
    req_headers.push((
      ByteString::from(":scheme"),
      ByteString::from(uri.scheme().map(|s| s.as_str()).unwrap_or("http")),
    ));
    req_headers.push((
      ByteString::from(":path"),
      ByteString::from(uri.path_and_query().map(|p| p.as_str()).unwrap_or("")),
    ));
    req_headers.push((
      ByteString::from(":authority"),
      ByteString::from(uri.authority().map(|a| a.as_str()).unwrap_or("")),
    ));
    for (key, val) in headers.iter() {
      req_headers.push((key.as_str().into(), val.as_bytes().into()));
    }

    let resp = state
      .borrow_mut()
      .resource_table
      .add(Http2ServerSendResponse {
        send_response: AsyncRefCell::new(resp),
      });

    Ok(Some((req_headers, stm, resp)))
  } else {
    Ok(None)
  }
}

#[op2(async)]
#[serde]
pub async fn op_http2_send_response(
  state: Rc<RefCell<OpState>>,
  #[smi] rid: ResourceId,
  #[smi] status: u16,
  #[serde] headers: Vec<(ByteString, ByteString)>,
) -> Result<(ResourceId, u32), Http2Error> {
  let resource = state
    .borrow()
    .resource_table
    .get::<Http2ServerSendResponse>(rid)
    .map_err(Http2Error::Resource)?;
  let mut send_response = RcRef::map(resource, |r| &r.send_response)
    .borrow_mut()
    .await;
  let mut response = Response::new(());
  if let Ok(status) = StatusCode::from_u16(status) {
    *response.status_mut() = status;
  }
  for (name, value) in headers {
    response.headers_mut().append(
      HeaderName::from_bytes(&name).unwrap(),
      HeaderValue::from_bytes(&value).unwrap(),
    );
  }

  let stream = send_response.send_response(response, false)?;
  let stream_id = stream.stream_id();

  Ok((rid, stream_id.into()))
}

#[op2(async)]
pub async fn op_http2_poll_client_connection(
  state: Rc<RefCell<OpState>>,
  #[smi] rid: ResourceId,
) -> Result<(), Http2Error> {
  let resource = state
    .borrow()
    .resource_table
    .get::<Http2ClientConn>(rid)
    .map_err(Http2Error::Resource)?;

  let cancel_handle = RcRef::map(resource.clone(), |this| &this.cancel_handle);
  let mut conn = RcRef::map(resource, |this| &this.conn).borrow_mut().await;

  match (&mut *conn).or_cancel(cancel_handle).await {
    Ok(result) => result?,
    Err(_) => {
      // TODO(bartlomieju): probably need a better mechanism for closing the connection

      // cancelled
    }
  }

  Ok(())
}

#[op2(async)]
#[serde]
pub async fn op_http2_client_request(
  state: Rc<RefCell<OpState>>,
  #[smi] client_rid: ResourceId,
  // TODO(bartlomieju): maybe use a vector with fixed layout to save sending
  // 4 strings of keys?
  #[serde] mut pseudo_headers: HashMap<String, String>,
  #[serde] headers: Vec<(ByteString, ByteString)>,
) -> Result<(ResourceId, u32), Http2Error> {
  let resource = state
    .borrow()
    .resource_table
    .get::<Http2Client>(client_rid)
    .map_err(Http2Error::Resource)?;

  let url = resource.url.clone();

  let pseudo_path = pseudo_headers.remove(":path").unwrap_or("/".to_string());
  let pseudo_method = pseudo_headers
    .remove(":method")
    .unwrap_or("GET".to_string());
  // TODO(bartlomieju): handle all pseudo-headers (:authority, :scheme)
  let _pseudo_authority = pseudo_headers
    .remove(":authority")
    .unwrap_or("/".to_string());
  let _pseudo_scheme = pseudo_headers
    .remove(":scheme")
    .unwrap_or("http".to_string());

  let url = url.join(&pseudo_path)?;

  let mut req = http::Request::builder()
    .uri(url.as_str())
    .method(pseudo_method.as_str());

  for (name, value) in headers {
    req.headers_mut().unwrap().append(
      HeaderName::from_bytes(&name).unwrap(),
      HeaderValue::from_bytes(&value).unwrap(),
    );
  }

  let request = req.body(()).unwrap();

  let resource = {
    let state = state.borrow();
    state
      .resource_table
      .get::<Http2Client>(client_rid)
      .map_err(Http2Error::Resource)?
  };
  let mut client = RcRef::map(&resource, |r| &r.client).borrow_mut().await;
  poll_fn(|cx| client.poll_ready(cx)).await?;
  let (response, stream) = client.send_request(request, false).unwrap();
  let stream_id = stream.stream_id();
  let stream_rid = state.borrow_mut().resource_table.add(Http2ClientStream {
    response: AsyncRefCell::new(response),
    stream: AsyncRefCell::new(stream),
  });
  Ok((stream_rid, stream_id.into()))
}

#[op2(async)]
pub async fn op_http2_client_send_data(
  state: Rc<RefCell<OpState>>,
  #[smi] stream_rid: ResourceId,
  #[buffer] data: JsBuffer,
  end_of_stream: bool,
) -> Result<(), Http2Error> {
  let resource = state
    .borrow()
    .resource_table
    .get::<Http2ClientStream>(stream_rid)
    .map_err(Http2Error::Resource)?;
  let mut stream = RcRef::map(&resource, |r| &r.stream).borrow_mut().await;

  stream.send_data(data.to_vec().into(), end_of_stream)?;
  Ok(())
}

#[op2(async)]
pub async fn op_http2_client_reset_stream(
  state: Rc<RefCell<OpState>>,
  #[smi] stream_rid: ResourceId,
  #[smi] code: u32,
) -> Result<(), deno_core::error::AnyError> {
  let resource = state
    .borrow()
    .resource_table
    .get::<Http2ClientStream>(stream_rid)?;
  let mut stream = RcRef::map(&resource, |r| &r.stream).borrow_mut().await;
  stream.send_reset(h2::Reason::from(code));
  Ok(())
}

#[op2(async)]
pub async fn op_http2_client_send_trailers(
  state: Rc<RefCell<OpState>>,
  #[smi] stream_rid: ResourceId,
  #[serde] trailers: Vec<(ByteString, ByteString)>,
) -> Result<(), Http2Error> {
  let resource = state
    .borrow()
    .resource_table
    .get::<Http2ClientStream>(stream_rid)
    .map_err(Http2Error::Resource)?;
  let mut stream = RcRef::map(&resource, |r| &r.stream).borrow_mut().await;

  let mut trailers_map = http::HeaderMap::new();
  for (name, value) in trailers {
    trailers_map.insert(
      HeaderName::from_bytes(&name).unwrap(),
      HeaderValue::from_bytes(&value).unwrap(),
    );
  }

  stream.send_trailers(trailers_map)?;
  Ok(())
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Http2ClientResponse {
  headers: Vec<(ByteString, ByteString)>,
  body_rid: ResourceId,
  status_code: u16,
}

#[op2(async)]
#[serde]
pub async fn op_http2_client_get_response(
  state: Rc<RefCell<OpState>>,
  #[smi] stream_rid: ResourceId,
) -> Result<(Http2ClientResponse, bool), Http2Error> {
  let resource = state
    .borrow()
    .resource_table
    .get::<Http2ClientStream>(stream_rid)
    .map_err(Http2Error::Resource)?;
  let mut response_future =
    RcRef::map(&resource, |r| &r.response).borrow_mut().await;

  let response = (&mut *response_future).await?;

  let (parts, body) = response.into_parts();
  let status = parts.status;
  let mut res_headers = Vec::new();
  for (key, val) in parts.headers.iter() {
    res_headers.push((key.as_str().into(), val.as_bytes().into()));
  }
  let end_stream = body.is_end_stream();

  let (trailers_tx, trailers_rx) = tokio::sync::oneshot::channel();
  let body_rid =
    state
      .borrow_mut()
      .resource_table
      .add(Http2ClientResponseBody {
        body: AsyncRefCell::new(body),
        trailers_rx: AsyncRefCell::new(Some(trailers_rx)),
        trailers_tx: AsyncRefCell::new(Some(trailers_tx)),
      });
  Ok((
    Http2ClientResponse {
      headers: res_headers,
      body_rid,
      status_code: status.into(),
    },
    end_stream,
  ))
}

enum DataOrTrailers {
  Data(Bytes),
  Trailers(HeaderMap),
  Eof,
}

fn poll_data_or_trailers(
  cx: &mut std::task::Context,
  body: &mut RecvStream,
) -> Poll<Result<DataOrTrailers, h2::Error>> {
  if let Poll::Ready(trailers) = body.poll_trailers(cx) {
    if let Some(trailers) = trailers? {
      return Poll::Ready(Ok(DataOrTrailers::Trailers(trailers)));
    } else {
      return Poll::Ready(Ok(DataOrTrailers::Eof));
    }
  }
  if let Poll::Ready(Some(data)) = body.poll_data(cx) {
    let data = data?;
    body.flow_control().release_capacity(data.len())?;
    return Poll::Ready(Ok(DataOrTrailers::Data(data)));
    // If `poll_data` returns `Ready(None)`, poll one more time to check for trailers
  }
  // Return pending here as poll_data will keep the waker
  Poll::Pending
}

#[op2(async)]
#[serde]
pub async fn op_http2_client_get_response_body_chunk(
  state: Rc<RefCell<OpState>>,
  #[smi] body_rid: ResourceId,
) -> Result<(Option<Vec<u8>>, bool, bool), Http2Error> {
  let resource = state
    .borrow()
    .resource_table
    .get::<Http2ClientResponseBody>(body_rid)
    .map_err(Http2Error::Resource)?;
  let mut body = RcRef::map(&resource, |r| &r.body).borrow_mut().await;

  loop {
    let result = poll_fn(|cx| poll_data_or_trailers(cx, &mut body)).await;
    if let Err(err) = result {
      match err.reason() {
        Some(Reason::NO_ERROR) => return Ok((None, true, false)),
        Some(Reason::CANCEL) => return Ok((None, false, true)),
        _ => return Err(err.into()),
      }
    }
    match result.unwrap() {
      DataOrTrailers::Data(data) => {
        return Ok((Some(data.to_vec()), false, false));
      }
      DataOrTrailers::Trailers(trailers) => {
        if let Some(trailers_tx) = RcRef::map(&resource, |r| &r.trailers_tx)
          .borrow_mut()
          .await
          .take()
        {
          _ = trailers_tx.send(Some(trailers));
        };

        continue;
      }
      DataOrTrailers::Eof => {
        RcRef::map(&resource, |r| &r.trailers_tx)
          .borrow_mut()
          .await
          .take();
        return Ok((None, true, false));
      }
    };
  }
}

#[op2(async)]
#[serde]
pub async fn op_http2_client_get_response_trailers(
  state: Rc<RefCell<OpState>>,
  #[smi] body_rid: ResourceId,
) -> Result<Option<Vec<(ByteString, ByteString)>>, deno_core::error::AnyError> {
  let resource = state
    .borrow()
    .resource_table
    .get::<Http2ClientResponseBody>(body_rid)?;
  let trailers = RcRef::map(&resource, |r| &r.trailers_rx)
    .borrow_mut()
    .await
    .take();
  if let Some(trailers) = trailers {
    if let Ok(Some(trailers)) = trailers.await {
      let mut v = Vec::with_capacity(trailers.len());
      for (key, value) in trailers.iter() {
        v.push((
          ByteString::from(key.as_str()),
          ByteString::from(value.as_bytes()),
        ));
      }
      Ok(Some(v))
    } else {
      Ok(None)
    }
  } else {
    Ok(None)
  }
}

1	// Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
2
3	use std::borrow::Cow;
4	use std::cell::RefCell;
5	use std::collections::HashMap;
6	use std::rc::Rc;
7	use std::task::Poll;
8
9	use bytes::Bytes;
10	use deno_core::futures::future::poll_fn;
11	use deno_core::op2;
12	use deno_core::serde::Serialize;
13	use deno_core::AsyncRefCell;
14	use deno_core::BufView;
15	use deno_core::ByteString;
16	use deno_core::CancelFuture;
17	use deno_core::CancelHandle;
18	use deno_core::JsBuffer;
19	use deno_core::OpState;
20	use deno_core::RcRef;
21	use deno_core::Resource;
22	use deno_core::ResourceId;
23	use deno_net::raw::take_network_stream_resource;
24	use deno_net::raw::NetworkStream;
25	use h2;
26	use h2::Reason;
27	use h2::RecvStream;
28	use http;
29	use http::header::HeaderName;
30	use http::header::HeaderValue;
31	use http::request::Parts;
32	use http::HeaderMap;
33	use http::Response;
34	use http::StatusCode;
35	use url::Url;
36
37	pub struct Http2Client {
38	pub client: AsyncRefCell<h2::client::SendRequest<BufView>>,
39	pub url: Url,
40	}
41
42	impl Resource for Http2Client {
UNCOV 43	fn name(&self) -> Cow<str> {	×
44	"http2Client".into()	×
45	}	×
46	}
47
48	#[derive(Debug)]
49	pub struct Http2ClientConn {
50	pub conn: AsyncRefCell<h2::client::Connection<NetworkStream, BufView>>,
51	cancel_handle: CancelHandle,
52	}
53
54	impl Resource for Http2ClientConn {
UNCOV 55	fn name(&self) -> Cow<str> {	×
56	"http2ClientConnection".into()	×
57	}	×
58
59	fn close(self: Rc<Self>) {	1✔
60	self.cancel_handle.cancel()	1✔
61	}	1✔
62	}
63
64	#[derive(Debug)]
65	pub struct Http2ClientStream {
66	pub response: AsyncRefCell<h2::client::ResponseFuture>,
67	pub stream: AsyncRefCell<h2::SendStream<BufView>>,
68	}
69
70	impl Resource for Http2ClientStream {
UNCOV 71	fn name(&self) -> Cow<str> {	×
72	"http2ClientStream".into()	×
73	}	×
74	}
75
76	#[derive(Debug)]
77	pub struct Http2ClientResponseBody {
78	pub body: AsyncRefCell<h2::RecvStream>,
79	pub trailers_rx:
80	AsyncRefCell<Option<tokio::sync::oneshot::Receiver<Option<HeaderMap>>>>,
81	pub trailers_tx:
82	AsyncRefCell<Option<tokio::sync::oneshot::Sender<Option<HeaderMap>>>>,
83	}
84
85	impl Resource for Http2ClientResponseBody {
86	fn name(&self) -> Cow<str> {	×
87	"http2ClientResponseBody".into()	×
UNCOV 88	}	×
89	}
90
91	#[derive(Debug)]
92	pub struct Http2ServerConnection {
93	pub conn: AsyncRefCell<h2::server::Connection<NetworkStream, BufView>>,
94	}
95
96	impl Resource for Http2ServerConnection {
97	fn name(&self) -> Cow<str> {	×
98	"http2ServerConnection".into()	×
UNCOV 99	}	×
100	}
101
102	pub struct Http2ServerSendResponse {
103	pub send_response: AsyncRefCell<h2::server::SendResponse<BufView>>,
104	}
105
106	impl Resource for Http2ServerSendResponse {
107	fn name(&self) -> Cow<str> {	×
108	"http2ServerSendResponse".into()	×
UNCOV 109	}	×
110	}
111
112	#[derive(Debug, thiserror::Error)]
113	pub enum Http2Error {
114	#[error(transparent)]
115	Resource(deno_core::error::AnyError),
116	#[error(transparent)]
117	UrlParse(#[from] url::ParseError),
118	#[error(transparent)]
119	H2(#[from] h2::Error),
120	}
121
122	#[op2(async)]	1,160✔
123	#[serde]
124	pub async fn op_http2_connect(	1✔
125	state: Rc<RefCell<OpState>>,	1✔
126	#[smi] rid: ResourceId,	1✔
127	#[string] url: String,	1✔
128	) -> Result<(ResourceId, ResourceId), Http2Error> {	1✔
129	// No permission check necessary because we're using an existing connection
130	let network_stream = {	1✔
131	let mut state = state.borrow_mut();	1✔
132	take_network_stream_resource(&mut state.resource_table, rid)	1✔
133	.map_err(Http2Error::Resource)?	1✔
134	};
135
136	let url = Url::parse(&url)?;	1✔
137
138	let (client, conn) =	1✔
139	h2::client::Builder::new().handshake(network_stream).await?;	1✔
140	let mut state = state.borrow_mut();	1✔
141	let client_rid = state.resource_table.add(Http2Client {	1✔
142	client: AsyncRefCell::new(client),	1✔
143	url,	1✔
144	});	1✔
145	let conn_rid = state.resource_table.add(Http2ClientConn {	1✔
146	conn: AsyncRefCell::new(conn),	1✔
147	cancel_handle: CancelHandle::new(),	1✔
148	});	1✔
149	Ok((client_rid, conn_rid))	1✔
150	}	1✔
151
152	#[op2(async)]	1,159✔
153	#[smi]
154	pub async fn op_http2_listen(	×
UNCOV 155	state: Rc<RefCell<OpState>>,	×
156	#[smi] rid: ResourceId,	×
UNCOV 157	) -> Result<ResourceId, Http2Error> {	×
158	let stream =	×
159	take_network_stream_resource(&mut state.borrow_mut().resource_table, rid)	×
160	.map_err(Http2Error::Resource)?;	×
161
162	let conn = h2::server::Builder::new().handshake(stream).await?;	×
163	Ok(	×
164	state	×
165	.borrow_mut()	×
166	.resource_table	×
167	.add(Http2ServerConnection {	×
168	conn: AsyncRefCell::new(conn),	×
169	}),	×
170	)	×
171	}	×
172
173	#[op2(async)]	1,159✔
174	#[serde]
175	pub async fn op_http2_accept(	×
176	state: Rc<RefCell<OpState>>,	×
177	#[smi] rid: ResourceId,	×
178	) -> Result<	×
179	Option<(Vec<(ByteString, ByteString)>, ResourceId, ResourceId)>,	×
180	Http2Error,	×
181	> {	×
182	let resource = state	×
183	.borrow()	×
184	.resource_table	×
185	.get::<Http2ServerConnection>(rid)	×
186	.map_err(Http2Error::Resource)?;	×
187	let mut conn = RcRef::map(&resource, \|r\| &r.conn).borrow_mut().await;	×
188	if let Some(res) = conn.accept().await {	×
189	let (req, resp) = res?;	×
190	let (parts, body) = req.into_parts();	×
191	let (trailers_tx, trailers_rx) = tokio::sync::oneshot::channel();	×
192	let stm = state	×
193	.borrow_mut()	×
194	.resource_table	×
195	.add(Http2ClientResponseBody {	×
196	body: AsyncRefCell::new(body),	×
197	trailers_rx: AsyncRefCell::new(Some(trailers_rx)),	×
198	trailers_tx: AsyncRefCell::new(Some(trailers_tx)),	×
199	});	×
200		×
201	let Parts {	×
202	uri,	×
203	method,	×
204	headers,	×
205	..	×
UNCOV 206	} = parts;	×
207	let mut req_headers = Vec::with_capacity(headers.len() + 4);	×
208	req_headers.push((	×
209	ByteString::from(":method"),	×
210	ByteString::from(method.as_str()),	×
211	));	×
212	req_headers.push((	×
213	ByteString::from(":scheme"),	×
214	ByteString::from(uri.scheme().map(\|s\| s.as_str()).unwrap_or("http")),	×
UNCOV 215	));	×
216	req_headers.push((	×
UNCOV 217	ByteString::from(":path"),	×
218	ByteString::from(uri.path_and_query().map(\|p\| p.as_str()).unwrap_or("")),	×
UNCOV 219	));	×
220	req_headers.push((	×
UNCOV 221	ByteString::from(":authority"),	×
222	ByteString::from(uri.authority().map(\|a\| a.as_str()).unwrap_or("")),	×
223	));	×
224	for (key, val) in headers.iter() {	×
225	req_headers.push((key.as_str().into(), val.as_bytes().into()));	×
226	}	×
227
228	let resp = state	×
229	.borrow_mut()	×
230	.resource_table	×
231	.add(Http2ServerSendResponse {	×
232	send_response: AsyncRefCell::new(resp),	×
233	});	×
234		×
235	Ok(Some((req_headers, stm, resp)))	×
236	} else {
237	Ok(None)	×
238	}
239	}	×
240
241	#[op2(async)]	1,159✔
242	#[serde]
243	pub async fn op_http2_send_response(	×
244	state: Rc<RefCell<OpState>>,	×
245	#[smi] rid: ResourceId,	×
UNCOV 246	#[smi] status: u16,	×
247	#[serde] headers: Vec<(ByteString, ByteString)>,	×
248	) -> Result<(ResourceId, u32), Http2Error> {	×
249	let resource = state	×
250	.borrow()	×
251	.resource_table	×
UNCOV 252	.get::<Http2ServerSendResponse>(rid)	×
253	.map_err(Http2Error::Resource)?;	×
254	let mut send_response = RcRef::map(resource, \|r\| &r.send_response)	×
255	.borrow_mut()	×
256	.await;	×
257	let mut response = Response::new(());	×
258	if let Ok(status) = StatusCode::from_u16(status) {	×
UNCOV 259	*response.status_mut() = status;	×
260	}	×
261	for (name, value) in headers {	×
UNCOV 262	response.headers_mut().append(	×
263	HeaderName::from_bytes(&name).unwrap(),	×
264	HeaderValue::from_bytes(&value).unwrap(),	×
265	);	×
266	}	×
267
268	let stream = send_response.send_response(response, false)?;	×
269	let stream_id = stream.stream_id();	×
UNCOV 270		×
UNCOV 271	Ok((rid, stream_id.into()))	×
272	}	×
273
274	#[op2(async)]	1,160✔
275	pub async fn op_http2_poll_client_connection(	1✔
276	state: Rc<RefCell<OpState>>,	1✔
277	#[smi] rid: ResourceId,	1✔
278	) -> Result<(), Http2Error> {	1✔
279	let resource = state	1✔
280	.borrow()	1✔
281	.resource_table	1✔
282	.get::<Http2ClientConn>(rid)	1✔
283	.map_err(Http2Error::Resource)?;	1✔
284
285	let cancel_handle = RcRef::map(resource.clone(), \|this\| &this.cancel_handle);	1✔
286	let mut conn = RcRef::map(resource, \|this\| &this.conn).borrow_mut().await;	1✔
287
288	match (&mut *conn).or_cancel(cancel_handle).await {	5✔
UNCOV 289	Ok(result) => result?,	×
290	Err(_) => {	1✔
291	// TODO(bartlomieju): probably need a better mechanism for closing the connection	1✔
292		1✔
293	// cancelled	1✔
294	}	1✔
295	}
296
297	Ok(())	1✔
298	}	1✔
299
300	#[op2(async)]	1,160✔
301	#[serde]
302	pub async fn op_http2_client_request(	1✔
303	state: Rc<RefCell<OpState>>,	1✔
304	#[smi] client_rid: ResourceId,	1✔
305	// TODO(bartlomieju): maybe use a vector with fixed layout to save sending	1✔
306	// 4 strings of keys?	1✔
307	#[serde] mut pseudo_headers: HashMap<String, String>,	1✔
308	#[serde] headers: Vec<(ByteString, ByteString)>,	1✔
309	) -> Result<(ResourceId, u32), Http2Error> {	1✔
310	let resource = state	1✔
311	.borrow()	1✔
312	.resource_table	1✔
313	.get::<Http2Client>(client_rid)	1✔
314	.map_err(Http2Error::Resource)?;	1✔
315
316	let url = resource.url.clone();	1✔
317		1✔
318	let pseudo_path = pseudo_headers.remove(":path").unwrap_or("/".to_string());	1✔
319	let pseudo_method = pseudo_headers	1✔
320	.remove(":method")	1✔
321	.unwrap_or("GET".to_string());	1✔
322	// TODO(bartlomieju): handle all pseudo-headers (:authority, :scheme)	1✔
323	let _pseudo_authority = pseudo_headers	1✔
324	.remove(":authority")	1✔
325	.unwrap_or("/".to_string());	1✔
326	let _pseudo_scheme = pseudo_headers	1✔
327	.remove(":scheme")	1✔
328	.unwrap_or("http".to_string());	1✔
329
330	let url = url.join(&pseudo_path)?;	1✔
331
332	let mut req = http::Request::builder()	1✔
333	.uri(url.as_str())	1✔
334	.method(pseudo_method.as_str());	1✔
335
336	for (name, value) in headers {	1✔
337	req.headers_mut().unwrap().append(	×
338	HeaderName::from_bytes(&name).unwrap(),	×
339	HeaderValue::from_bytes(&value).unwrap(),	×
340	);	×
341	}	×
342
343	let request = req.body(()).unwrap();	1✔
344
345	let resource = {	1✔
346	let state = state.borrow();	1✔
347	state	1✔
348	.resource_table	1✔
349	.get::<Http2Client>(client_rid)	1✔
350	.map_err(Http2Error::Resource)?	1✔
351	};
352	let mut client = RcRef::map(&resource, \|r\| &r.client).borrow_mut().await;	1✔
353	poll_fn(\|cx\| client.poll_ready(cx)).await?;	1✔
354	let (response, stream) = client.send_request(request, false).unwrap();	1✔
355	let stream_id = stream.stream_id();	1✔
356	let stream_rid = state.borrow_mut().resource_table.add(Http2ClientStream {	1✔
357	response: AsyncRefCell::new(response),	1✔
358	stream: AsyncRefCell::new(stream),	1✔
359	});	1✔
360	Ok((stream_rid, stream_id.into()))	1✔
361	}	1✔
362
363	#[op2(async)]	1,160✔
364	pub async fn op_http2_client_send_data(	1✔
365	state: Rc<RefCell<OpState>>,	1✔
366	#[smi] stream_rid: ResourceId,	1✔
367	#[buffer] data: JsBuffer,	1✔
368	end_of_stream: bool,	1✔
369	) -> Result<(), Http2Error> {	1✔
370	let resource = state	1✔
371	.borrow()	1✔
372	.resource_table	1✔
373	.get::<Http2ClientStream>(stream_rid)	1✔
374	.map_err(Http2Error::Resource)?;	1✔
375	let mut stream = RcRef::map(&resource, \|r\| &r.stream).borrow_mut().await;	1✔
376
377	stream.send_data(data.to_vec().into(), end_of_stream)?;	1✔
378	Ok(())	1✔
379	}	1✔
380
381	#[op2(async)]	1,160✔
382	pub async fn op_http2_client_reset_stream(	1✔
383	state: Rc<RefCell<OpState>>,	1✔
384	#[smi] stream_rid: ResourceId,	1✔
385	#[smi] code: u32,	1✔
386	) -> Result<(), deno_core::error::AnyError> {	1✔
387	let resource = state	1✔
388	.borrow()	1✔
389	.resource_table	1✔
390	.get::<Http2ClientStream>(stream_rid)?;	1✔
UNCOV 391	let mut stream = RcRef::map(&resource, \|r\| &r.stream).borrow_mut().await;	×
UNCOV 392	stream.send_reset(h2::Reason::from(code));	×
UNCOV 393	Ok(())	×
394	}	1✔
395
396	#[op2(async)]	1,159✔
UNCOV 397	pub async fn op_http2_client_send_trailers(	×
UNCOV 398	state: Rc<RefCell<OpState>>,	×
399	#[smi] stream_rid: ResourceId,	×
UNCOV 400	#[serde] trailers: Vec<(ByteString, ByteString)>,	×
401	) -> Result<(), Http2Error> {	×
402	let resource = state	×
403	.borrow()	×
404	.resource_table	×
405	.get::<Http2ClientStream>(stream_rid)	×
406	.map_err(Http2Error::Resource)?;	×
407	let mut stream = RcRef::map(&resource, \|r\| &r.stream).borrow_mut().await;	×
408
409	let mut trailers_map = http::HeaderMap::new();	×
UNCOV 410	for (name, value) in trailers {	×
411	trailers_map.insert(	×
UNCOV 412	HeaderName::from_bytes(&name).unwrap(),	×
413	HeaderValue::from_bytes(&value).unwrap(),	×
414	);	×
415	}	×
416
417	stream.send_trailers(trailers_map)?;	×
418	Ok(())	×
419	}	×
420
421	#[derive(Serialize)]
422	#[serde(rename_all = "camelCase")]
423	pub struct Http2ClientResponse {
424	headers: Vec<(ByteString, ByteString)>,
425	body_rid: ResourceId,
426	status_code: u16,
427	}
428
429	#[op2(async)]	1,160✔
430	#[serde]
431	pub async fn op_http2_client_get_response(	1✔
432	state: Rc<RefCell<OpState>>,	1✔
433	#[smi] stream_rid: ResourceId,	1✔
434	) -> Result<(Http2ClientResponse, bool), Http2Error> {	1✔
435	let resource = state	1✔
436	.borrow()	1✔
437	.resource_table	1✔
438	.get::<Http2ClientStream>(stream_rid)	1✔
439	.map_err(Http2Error::Resource)?;	1✔
440	let mut response_future =	1✔
441	RcRef::map(&resource, \|r\| &r.response).borrow_mut().await;	1✔
442
443	let response = (&mut *response_future).await?;	2✔
444
445	let (parts, body) = response.into_parts();	1✔
446	let status = parts.status;	1✔
447	let mut res_headers = Vec::new();	1✔
448	for (key, val) in parts.headers.iter() {	7✔
449	res_headers.push((key.as_str().into(), val.as_bytes().into()));	7✔
450	}	7✔
451	let end_stream = body.is_end_stream();	1✔
452		1✔
453	let (trailers_tx, trailers_rx) = tokio::sync::oneshot::channel();	1✔
454	let body_rid =	1✔
455	state	1✔
456	.borrow_mut()	1✔
457	.resource_table	1✔
458	.add(Http2ClientResponseBody {	1✔
459	body: AsyncRefCell::new(body),	1✔
460	trailers_rx: AsyncRefCell::new(Some(trailers_rx)),	1✔
461	trailers_tx: AsyncRefCell::new(Some(trailers_tx)),	1✔
462	});	1✔
463	Ok((	1✔
464	Http2ClientResponse {	1✔
465	headers: res_headers,	1✔
466	body_rid,	1✔
467	status_code: status.into(),	1✔
468	},	1✔
469	end_stream,	1✔
470	))	1✔
471	}	1✔
472
473	enum DataOrTrailers {
474	Data(Bytes),
475	Trailers(HeaderMap),
476	Eof,
477	}
478
479	fn poll_data_or_trailers(	2✔
480	cx: &mut std::task::Context,	2✔
481	body: &mut RecvStream,	2✔
482	) -> Poll<Result<DataOrTrailers, h2::Error>> {	2✔
483	if let Poll::Ready(trailers) = body.poll_trailers(cx) {	2✔
484	if let Some(trailers) = trailers? {	1✔
485	return Poll::Ready(Ok(DataOrTrailers::Trailers(trailers)));	×
486	} else {
487	return Poll::Ready(Ok(DataOrTrailers::Eof));	1✔
488	}
489	}	1✔
490	if let Poll::Ready(Some(data)) = body.poll_data(cx) {	1✔
491	let data = data?;	1✔
492	body.flow_control().release_capacity(data.len())?;	1✔
493	return Poll::Ready(Ok(DataOrTrailers::Data(data)));	1✔
494	// If `poll_data` returns `Ready(None)`, poll one more time to check for trailers
495	}	×
UNCOV 496	// Return pending here as poll_data will keep the waker	×
497	Poll::Pending	×
498	}	2✔
499
500	#[op2(async)]	1,161✔
501	#[serde]
502	pub async fn op_http2_client_get_response_body_chunk(	2✔
503	state: Rc<RefCell<OpState>>,	2✔
504	#[smi] body_rid: ResourceId,	2✔
505	) -> Result<(Option<Vec<u8>>, bool, bool), Http2Error> {	2✔
506	let resource = state	2✔
507	.borrow()	2✔
508	.resource_table	2✔
509	.get::<Http2ClientResponseBody>(body_rid)	2✔
510	.map_err(Http2Error::Resource)?;	2✔
511	let mut body = RcRef::map(&resource, \|r\| &r.body).borrow_mut().await;	2✔
512
513	loop {
514	let result = poll_fn(\|cx\| poll_data_or_trailers(cx, &mut body)).await;	2✔
515	if let Err(err) = result {	2✔
UNCOV 516	match err.reason() {	×
517	Some(Reason::NO_ERROR) => return Ok((None, true, false)),	×
UNCOV 518	Some(Reason::CANCEL) => return Ok((None, false, true)),	×
519	_ => return Err(err.into()),	×
520	}
521	}	2✔
522	match result.unwrap() {	2✔
523	DataOrTrailers::Data(data) => {	1✔
524	return Ok((Some(data.to_vec()), false, false));	1✔
525	}
526	DataOrTrailers::Trailers(trailers) => {	×
527	if let Some(trailers_tx) = RcRef::map(&resource, \|r\| &r.trailers_tx)	×
528	.borrow_mut()	×
529	.await	×
530	.take()	×
531	{	×
532	_ = trailers_tx.send(Some(trailers));	×
533	};	×
534
535	continue;	×
536	}
537	DataOrTrailers::Eof => {
538	RcRef::map(&resource, \|r\| &r.trailers_tx)	1✔
539	.borrow_mut()	1✔
540	.await	×
541	.take();	1✔
542	return Ok((None, true, false));	1✔
543	}
544	};
545	}
546	}	2✔
547
548	#[op2(async)]	1,160✔
549	#[serde]
550	pub async fn op_http2_client_get_response_trailers(	1✔
551	state: Rc<RefCell<OpState>>,	1✔
552	#[smi] body_rid: ResourceId,	1✔
553	) -> Result<Option<Vec<(ByteString, ByteString)>>, deno_core::error::AnyError> {	1✔
554	let resource = state	1✔
555	.borrow()	1✔
556	.resource_table	1✔
557	.get::<Http2ClientResponseBody>(body_rid)?;	1✔
558	let trailers = RcRef::map(&resource, \|r\| &r.trailers_rx)	1✔
559	.borrow_mut()	1✔
UNCOV 560	.await	×
561	.take();	1✔
562	if let Some(trailers) = trailers {	1✔
563	if let Ok(Some(trailers)) = trailers.await {	1✔
UNCOV 564	let mut v = Vec::with_capacity(trailers.len());	×
UNCOV 565	for (key, value) in trailers.iter() {	×
UNCOV 566	v.push((	×
UNCOV 567	ByteString::from(key.as_str()),	×
UNCOV 568	ByteString::from(value.as_bytes()),	×
UNCOV 569	));	×
UNCOV 570	}	×
UNCOV 571	Ok(Some(v))	×
572	} else {
573	Ok(None)	1✔
574	}
575	} else {
UNCOV 576	Ok(None)	×
577	}
578	}	1✔

supabase / edge-runtime / 17229631411

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous