fix(ext/node): fix grpc error_handling example (#23755)

gRPC depends only on the END_STREAM flag to emit "trailers" event which
is responsible to propagate the errors correctly. This patch uses
Body::is_end_stream() to determine if a stream will end and set the
END_STREAM flag.

Co-authored-by: Bartek Iwańczuk <biwanczuk@gmail.com>

ext/node/ops/http2.rs

@@ -423,7 +423,7 @@ pub struct Http2ClientResponse {
 pub async fn op_http2_client_get_response(
   state: Rc<RefCell<OpState>>,
   #[smi] stream_rid: ResourceId,
-) -> Result<Http2ClientResponse, AnyError> {
+) -> Result<(Http2ClientResponse, bool), AnyError> {
   let resource = state
     .borrow()
     .resource_table
@@ -439,6 +439,7 @@ pub async fn op_http2_client_get_response(
   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 =
@@ -450,11 +451,14 @@ pub async fn op_http2_client_get_response(
         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(),
-  })
+  Ok((
+    Http2ClientResponse {
+      headers: res_headers,
+      body_rid,
+      status_code: status.into(),
+    },
+    end_stream,
+  ))
 }
 
 enum DataOrTrailers {

ext/node/polyfills/http2.ts

@@ -822,7 +822,7 @@ export class ClientHttp2Stream extends Duplex {
         session[kDenoClientRid],
         this.#rid,
       );
-      const response = await op_http2_client_get_response(
+      const [response, endStream] = await op_http2_client_get_response(
         this.#rid,
       );
       debugHttp2(">>> after get response", response);
@@ -831,7 +831,13 @@ export class ClientHttp2Stream extends Duplex {
         ...Object.fromEntries(response.headers),
       };
       debugHttp2(">>> emitting response", headers);
-      this.emit("response", headers, 0);
+      this.emit(
+        "response",
+        headers,
+        endStream
+          ? constants.NGHTTP2_FLAG_END_STREAM
+          : constants.NGHTTP2_FLAG_NONE,
+      );
       this[kDenoResponse] = response;
       this.emit("ready");
     })();