[core-rest-pipeline] Fix race condition in request

sdk/core/core-rest-pipeline/src/nodeHttpClient.ts

@@ -217,8 +217,9 @@ class NodeHttpClient implements HttpClient {
       });
 
       abortController.signal.addEventListener("abort", () => {
-        req.abort();
-        reject(new AbortError("The operation was aborted."));
+        const abortError = new AbortError("The operation was aborted.");
+        req.destroy(abortError);
+        reject(abortError);
       });
       if (body && isReadableStream(body)) {
         body.pipe(req);
@@ -229,7 +230,7 @@ class NodeHttpClient implements HttpClient {
           req.end(ArrayBuffer.isView(body) ? Buffer.from(body.buffer) : Buffer.from(body));
         } else {
           logger.error("Unrecognized body type", body);
-          throw new RestError("Unrecognized body type");
+          reject(new RestError("Unrecognized body type"));
         }
       } else {
         // streams don't like "undefined" being passed as data
@@ -313,11 +314,15 @@ function streamToText(stream: NodeJS.ReadableStream): Promise<string> {
       resolve(Buffer.concat(buffer).toString("utf8"));
     });
     stream.on("error", (e) => {
-      reject(
-        new RestError(`Error reading response as text: ${e.message}`, {
-          code: RestError.PARSE_ERROR
-        })
-      );
+      if (e && e?.name === "AbortError") {
+        reject(e);
+      } else {
+        reject(
+          new RestError(`Error reading response as text: ${e.message}`, {
+            code: RestError.PARSE_ERROR
+          })
+        );
+      }
     });
   });
 }

sdk/core/core-rest-pipeline/test/node/nodeHttpClient.spec.ts

@@ -15,12 +15,7 @@ class FakeResponse extends PassThrough {
   public headers?: IncomingHttpHeaders;
 }
 
-class FakeRequest extends PassThrough {
-  public finished?: boolean;
-  public abort(): void {
-    this.finished = true;
-  }
-}
+class FakeRequest extends PassThrough {}
 
 /**
  * Generic NodeJS streams accept typed arrays just fine,
@@ -53,7 +48,6 @@ function createResponse(statusCode: number, body = ""): IncomingMessage {
 
 function createRequest(): ClientRequest {
   const request = new FakeRequest();
-  request.finished = false;
   return (request as unknown) as ClientRequest;
 }
 
@@ -334,4 +328,40 @@ describe("NodeHttpClient", function() {
     const response = await promise;
     assert.strictEqual(response.status, 200);
   });
+
+  it("should return an AbortError when aborted while reading the HTTP response", async function() {
+    clock.restore();
+    const client = createDefaultHttpClient();
+    const controller = new AbortController();
+
+    const clientRequest = createRequest();
+    clientRequest.destroy = function(this: FakeRequest, e: Error) {
+      // give it some time to attach listeners and read from the stream
+      setTimeout(() => {
+        streamResponse.destroy(e);
+      }, 100);
+    };
+    stubbedHttpsRequest.returns(clientRequest);
+    const request = createPipelineRequest({
+      url: "https://example.com",
+      abortSignal: controller.signal
+    });
+    const promise = client.sendRequest(request);
+
+    const streamResponse = new FakeResponse();
+    streamResponse.headers = {};
+    streamResponse.statusCode = 200;
+    const buffer = Buffer.from("The start of an HTTP body");
+    streamResponse.write(buffer);
+    stubbedHttpsRequest.yield(streamResponse);
+    controller.abort();
+
+    try {
+      await promise;
+      assert.fail("Expected await to throw");
+    } catch (e) {
+      console.log(e);
+      assert.strictEqual(e.name, "AbortError");
+    }
+  });
 });