[core] Support NodeJS Stream cancellation - v2 (#33385)

### Packages impacted by this PR
@azure-rest/core-client, and @azure/core-sse

### Issues associated with this PR
#33334

### Describe the problem that is addressed by this PR

The type of NodeJS streams used in our core doesn't support cancellation
even though the actual runtime object does support it. For context, this
type is being used by NodeJS to type ["old style streams" that existed
before NodeJS
v0.10](https://github.com/DefinitelyTyped/DefinitelyTyped/blob/3a17ff00600a50db103774cfb818685c5677e815/types/node/stream.d.ts#L449),
therefore, it doesn't have the `destroy` method.

### What are the possible designs available to address the problem? If
there are more than one possible design, why was the one in this PR
chosen?

A few options have been already discussed in the issue. .

### Are there test cases added in this PR? _(If not, why?)_

No, the changes are types only

### Provide a list of related PRs _(if any)_

N/A

### Command used to generate this PR:**_(Applicable only to SDK release
request PRs)_

### Checklists
- [x] Added impacted package name to the issue description
- [ ] Does this PR needs any fixes in the SDK Generator?** _(If so,
create an Issue in the
[Autorest/typescript](https://github.com/Azure/autorest.typescript)
repository and link it here)_
- [x] Added a changelog (if necessary)

Author: deyaaeldeen

sdk/core/core-client-rest/CHANGELOG.md

@@ -4,6 +4,8 @@
 
 ### Features Added
 
+- `asNodeStream` now returns a `NodeJSReadableStream` which can be canceled by calling the `destroy` method.
+
 ### Breaking Changes
 
 ### Bugs Fixed

sdk/core/core-client-rest/review/core-client.api.md

@@ -98,7 +98,7 @@ export type HttpBrowserStreamResponse = HttpResponse & {
 
 // @public
 export type HttpNodeStreamResponse = HttpResponse & {
-    body?: NodeJS.ReadableStream;
+    body?: NodeJSReadableStream;
 };
 
 // @public
@@ -115,6 +115,11 @@ export interface InnerError {
     innererror?: InnerError;
 }
 
+// @public
+export interface NodeJSReadableStream extends NodeJS.ReadableStream {
+    destroy(error?: Error): void;
+}
+
 // @public
 export interface OperationOptions {
     abortSignal?: AbortSignalLike;

sdk/core/core-client-rest/src/common.ts

@@ -217,14 +217,26 @@ export interface Client {
   pathUnchecked: PathUnchecked;
 }
 
+/**
+ * A Node.js Readable stream that also has a `destroy` method.
+ */
+export interface NodeJSReadableStream extends NodeJS.ReadableStream {
+  /**
+   * Destroy the stream. Optionally emit an 'error' event, and emit a
+   * 'close' event (unless emitClose is set to false). After this call,
+   * internal resources will be released.
+   */
+  destroy(error?: Error): void;
+}
+
 /**
  * Http Response which body is a NodeJS stream object
  */
 export type HttpNodeStreamResponse = HttpResponse & {
   /**
    * Streamable body
    */
-  body?: NodeJS.ReadableStream;
+  body?: NodeJSReadableStream;
 };
 
 /**

sdk/core/core-sse/CHANGELOG.md

@@ -4,6 +4,8 @@
 
 ### Features Added
 
+- `createSseStream` now supports NodeJS streams as input.
+
 ### Breaking Changes
 
 ### Bugs Fixed

sdk/core/core-sse/review/core-sse.api.md

@@ -12,6 +12,9 @@ export function createSseStream(chunkStream: ReadableStream<Uint8Array>): EventM
 // @public
 export function createSseStream(chunkStream: IncomingMessage): EventMessageStream;
 
+// @public
+export function createSseStream(chunkStream: NodeJSReadableStream): EventMessageStream;
+
 // @public
 export interface EventMessage {
     data: string;
@@ -24,6 +27,11 @@ export interface EventMessage {
 export interface EventMessageStream extends ReadableStream<EventMessage>, AsyncDisposable, AsyncIterable<EventMessage> {
 }
 
+// @public
+export interface NodeJSReadableStream extends NodeJS.ReadableStream {
+    destroy(error?: Error): void;
+}
+
 // (No @packageDocumentation comment for this package)
 
 ```

sdk/core/core-sse/src/index.ts

@@ -2,4 +2,4 @@
 // Licensed under the MIT License.
 
 export { createSseStream } from "./sse.js";
-export { EventMessage, EventMessageStream } from "./models.js";
+export { EventMessage, EventMessageStream, NodeJSReadableStream } from "./models.js";

sdk/core/core-sse/src/models.ts

@@ -25,3 +25,15 @@ export interface EventMessageStream
     AsyncIterable<EventMessage> {}
 
 export type PartialSome<T, K extends keyof T> = Omit<T, K> & Partial<Pick<T, K>>;
+
+/**
+ * A Node.js Readable stream that also has a `destroy` method.
+ */
+export interface NodeJSReadableStream extends NodeJS.ReadableStream {
+  /**
+   * Destroy the stream. Optionally emit an 'error' event, and emit a
+   * 'close' event (unless emitClose is set to false). After this call,
+   * internal resources will be released.
+   */
+  destroy(error?: Error): void;
+}

sdk/core/core-sse/src/sse.ts

@@ -2,7 +2,12 @@
 // Licensed under the MIT License.
 
 import type { IncomingMessage } from "node:http";
-import type { EventMessage, EventMessageStream, PartialSome } from "./models.js";
+import type {
+  EventMessage,
+  EventMessageStream,
+  NodeJSReadableStream,
+  PartialSome,
+} from "./models.js";
 import { createStream, ensureAsyncIterable } from "./utils.js";
 
 enum ControlChars {
@@ -15,17 +20,23 @@ enum ControlChars {
 /**
  * Processes a response stream into a stream of events.
  * @param chunkStream - A stream of Uint8Array chunks
- * @returns An async iterable of EventMessage objects
+ * @returns A stream of EventMessage objects
  */
 export function createSseStream(chunkStream: ReadableStream<Uint8Array>): EventMessageStream;
 /**
  * Processes a response stream into a stream of events.
- * @param chunkStream - An async iterable of Uint8Array chunks
- * @returns An async iterable of EventMessage objects
+ * @param chunkStream - A NodeJS HTTP response
+ * @returns A stream of EventMessage objects
  */
 export function createSseStream(chunkStream: IncomingMessage): EventMessageStream;
+/**
+ * Processes a response stream into a stream of events.
+ * @param chunkStream - A NodeJS Readable stream
+ * @returns A stream of EventMessage objects
+ */
+export function createSseStream(chunkStream: NodeJSReadableStream): EventMessageStream;
 export function createSseStream(
-  chunkStream: IncomingMessage | ReadableStream<Uint8Array>,
+  chunkStream: IncomingMessage | NodeJSReadableStream | ReadableStream<Uint8Array>,
 ): EventMessageStream {
   const { cancel, iterable } = ensureAsyncIterable(chunkStream);
   const asyncIter = toMessage(toLine(iterable));

sdk/core/core-sse/src/utils.ts

@@ -2,10 +2,7 @@
 // Licensed under the MIT License.
 
 import type { IncomingMessage } from "node:http";
-
-type Nullable<T, K extends keyof T> = {
-  [P in K]: T[P] | null;
-} & Omit<T, K>;
+import type { NodeJSReadableStream } from "./models.js";
 
 export function createStream<T>(
   asyncIter: AsyncIterableIterator<T>,
@@ -65,7 +62,7 @@ function iteratorToStream<T>(
 }
 
 export function ensureAsyncIterable(
-  stream: Nullable<IncomingMessage, "socket"> | ReadableStream<Uint8Array>,
+  stream: IncomingMessage | NodeJSReadableStream | ReadableStream<Uint8Array>,
 ): {
   cancel(): Promise<void>;
   iterable: AsyncIterable<Uint8Array>;
@@ -79,11 +76,12 @@ export function ensureAsyncIterable(
   } else {
     return {
       cancel: async () => {
-        /**
-         * socket is set to null when the stream has been consumed
-         * so we need to make sure to guard against nullability.
-         */
-        stream.socket?.end();
+        // socket could be null if the connection is already closed
+        if ("socket" in stream && stream.socket) {
+          stream.socket.end();
+        } else {
+          stream.destroy();
+        }
       },
       iterable: stream as AsyncIterable<Uint8Array>,
     };

sdk/core/core-sse/test/public/node/connection.spec.ts

@@ -1,11 +1,10 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT License.
 
-import { createSseStream } from "../../../src/index.js";
+import { createSseStream, NodeJSReadableStream } from "../../../src/index.js";
 import { type Client, getClient } from "@azure-rest/core-client";
 import { assert, beforeAll, beforeEach, afterEach, describe, it } from "vitest";
 import { port } from "../../server/config.mjs";
-import type { IncomingMessage } from "http";
 import { matrix } from "@azure-tools/test-utils-vitest";
 import { isRestError } from "@azure/core-rest-pipeline";
 
@@ -18,7 +17,7 @@ async function sendRequest(
   client: Client,
   path: string,
   abortSignal?: AbortSignal,
-): Promise<IncomingMessage> {
+): Promise<NodeJSReadableStream> {
   const res = await client
     .pathUnchecked(path)
     .get({ accept: contentType, abortSignal })
@@ -33,7 +32,7 @@ async function sendRequest(
   if (!receivedContentType.includes(contentType)) {
     throw new Error(`Expected a text/event-stream content but received"${receivedContentType}"`);
   }
-  return res.body as IncomingMessage;
+  return res.body;
 }
 
 describe("[Node] Connections", () => {
@@ -71,7 +70,7 @@ describe("[Node] Connections", () => {
         });
 
         it("loop until stream ends and then break", async function () {
-          let stream: IncomingMessage;
+          let stream: NodeJSReadableStream;
           try {
             stream = await sendRequest(client, path);
           } catch (e) {
@@ -88,7 +87,7 @@ describe("[Node] Connections", () => {
               ran = true;
               if (sse.data === "[DONE]") {
                 if (path.includes("no-fin")) {
-                  assert.isNull(stream.socket);
+                  assert.isNull((stream as any).socket);
                 }
                 break;
               }
@@ -100,7 +99,7 @@ describe("[Node] Connections", () => {
         });
 
         it("break early from loop", async function () {
-          let stream: IncomingMessage;
+          let stream: NodeJSReadableStream;
           try {
             // sometimes the server gets into a bad state and doesn't respond so we need to timeout
             stream = await sendRequest(client, path, AbortSignal.timeout(25000));

sdk/core/core-sse/test/public/node/util.ts

@@ -10,6 +10,5 @@ export function createStream(
   const stream = new PassThrough();
   cb((c) => stream.write(c));
   stream.end();
-  // This is a mock test that doesn't need a true IncomingMessage object
-  return createSseStream(stream as any);
+  return createSseStream(stream);
 }

sdk/core/ts-http-runtime/review/azure-core-comparison.diff

@@ -489,7 +489,7 @@ index 30dac33..66ea99c 100644
    if (!cachedHttpClient) {
      cachedHttpClient = createDefaultHttpClient();
 diff --git a/src/client/common.ts b/src/client/common.ts
-index 609878a..da6742a 100644
+index 58f6eba..da6742a 100644
 --- a/src/client/common.ts
 +++ b/src/client/common.ts
 @@ -3,19 +3,17 @@
@@ -572,7 +572,35 @@ index 609878a..da6742a 100644
     *
     * type MyClient = Client & {
     *   path: Routes;
-@@ -333,11 +314,9 @@ export type ClientOptions = PipelineOptions & {
+@@ -217,18 +198,6 @@ export interface Client {
+   pathUnchecked: PathUnchecked;
+ }
+ 
+-/**
+- * A Node.js Readable stream that also has a `destroy` method.
+- */
+-export interface NodeJSReadableStream extends NodeJS.ReadableStream {
+-  /**
+-   * Destroy the stream. Optionally emit an 'error' event, and emit a
+-   * 'close' event (unless emitClose is set to false). After this call,
+-   * internal resources will be released.
+-   */
+-  destroy(error?: Error): void;
+-}
+-
+ /**
+  * Http Response which body is a NodeJS stream object
+  */
+@@ -236,7 +205,7 @@ export type HttpNodeStreamResponse = HttpResponse & {
+   /**
+    * Streamable body
+    */
+-  body?: NodeJSReadableStream;
++  body?: NodeJS.ReadableStream;
+ };
+ 
+ /**
+@@ -345,11 +314,9 @@ export type ClientOptions = PipelineOptions & {
       */
      apiKeyHeaderName?: string;
    };