From 7f8bf2537db0ae596a2c1baabd4011a190666ca6 Mon Sep 17 00:00:00 2001 From: Matt Mastracci Date: Thu, 3 Aug 2023 14:27:25 -0600 Subject: refactor(ext/fetch): refactor fetch to use new write_error method (#20029) This is a prerequisite for fast streams work -- this particular resource used a custom `mpsc`-style stream, and this work will allow us to unify it with the streams in `ext/http` in time. Instead of using Option as an internal semaphore for "correctly completed EOF", we allow code to propagate errors into the channel which can be picked up by downstream sinks like Hyper. EOF is signalled using a more standard sender drop. --- ext/node/ops/http.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) (limited to 'ext/node/ops') diff --git a/ext/node/ops/http.rs b/ext/node/ops/http.rs index cc7dbf522..2a4d31f50 100644 --- a/ext/node/ops/http.rs +++ b/ext/node/ops/http.rs @@ -10,12 +10,12 @@ use deno_core::CancelFuture; use deno_core::CancelHandle; use deno_core::OpState; use deno_fetch::get_or_create_client_from_state; +use deno_fetch::FetchBodyStream; use deno_fetch::FetchCancelHandle; use deno_fetch::FetchRequestBodyResource; use deno_fetch::FetchRequestResource; use deno_fetch::FetchReturn; use deno_fetch::HttpClientResource; -use deno_fetch::MpscByteStream; use reqwest::header::HeaderMap; use reqwest::header::HeaderName; use reqwest::header::HeaderValue; @@ -64,12 +64,12 @@ where let request_body_rid = if has_body { // If no body is passed, we return a writer for streaming the body. - let (stream, tx) = MpscByteStream::new(); + let (tx, stream) = tokio::sync::mpsc::channel(1); - request = request.body(Body::wrap_stream(stream)); + request = request.body(Body::wrap_stream(FetchBodyStream(stream))); let request_body_rid = state.resource_table.add(FetchRequestBodyResource { - body: AsyncRefCell::new(tx), + body: AsyncRefCell::new(Some(tx)), cancel: CancelHandle::default(), }); -- cgit v1.2.3