Revert "backend-common: work around premature close errors"

This reverts commit b29587c3187fdd5c78341dce5c6f406b631e1f21.

Signed-off-by: Patrik Oldsberg <poldsberg@gmail.com>

.changeset/swift-suits-reply.md

@@ -1,5 +0,0 @@
----
-'@backstage/backend-common': patch
----
-
-Refactored internal usage of the build-in `pipeline` from `'stream'` to work around `tar` bug in Node 18.

packages/backend-common/src/reading/GerritUrlReader.ts

@@ -32,7 +32,8 @@ import fetch, { Response } from 'node-fetch';
 import os from 'os';
 import { join as joinPath } from 'path';
 import tar from 'tar';
-import { Readable } from 'stream';
+import { pipeline as pipelineCb, Readable } from 'stream';
+import { promisify } from 'util';
 import {
   ReaderFactory,
   ReadTreeOptions,
@@ -44,7 +45,8 @@ import {
   UrlReader,
 } from './types';
 import { ScmIntegrations } from '@backstage/integration';
-import { pipeStream } from './tree/util';
+
+const pipeline = promisify(pipelineCb);
 
 const createTemporaryDirectory = async (workDir: string): Promise<string> =>
   await fs.mkdtemp(joinPath(workDir, '/gerrit-clone-'));
@@ -195,7 +197,7 @@ export class GerritUrlReader implements UrlReader {
       });
 
       const data = await new Promise<Buffer>(async resolve => {
-        await pipeStream(
+        await pipeline(
           tar.create({ cwd: tempDir }, ['']),
           concatStream(resolve),
         );

packages/backend-common/src/reading/tree/ReadableArrayResponse.ts

@@ -19,15 +19,17 @@ import platformPath, { basename } from 'path';
 
 import getRawBody from 'raw-body';
 import fs from 'fs-extra';
+import { promisify } from 'util';
 import tar from 'tar';
-import { Readable } from 'stream';
+import { pipeline as pipelineCb, Readable } from 'stream';
 import {
   ReadTreeResponse,
   ReadTreeResponseFile,
   ReadTreeResponseDirOptions,
   FromReadableArrayOptions,
 } from '../types';
-import { pipeStream } from './util';
+
+const pipeline = promisify(pipelineCb);
 
 /**
  * Wraps a array of Readable objects into a tree response reader.
@@ -73,7 +75,7 @@ export class ReadableArrayResponse implements ReadTreeResponse {
 
     try {
       const data = await new Promise<Buffer>(async resolve => {
-        await pipeStream(
+        await pipeline(
           tar.create({ cwd: tmpDir }, ['']),
           concatStream(resolve),
         );
@@ -93,7 +95,7 @@ export class ReadableArrayResponse implements ReadTreeResponse {
 
     for (let i = 0; i < this.stream.length; i++) {
       if (!this.stream[i].path.endsWith('/')) {
-        await pipeStream(
+        await pipeline(
           this.stream[i].data,
           fs.createWriteStream(
             platformPath.join(dir, basename(this.stream[i].path)),

packages/backend-common/src/reading/tree/TarArchiveResponse.ts

@@ -17,18 +17,21 @@
 import concatStream from 'concat-stream';
 import fs from 'fs-extra';
 import platformPath from 'path';
-import { Readable } from 'stream';
+import { pipeline as pipelineCb, Readable } from 'stream';
 import tar, { Parse, ParseStream, ReadEntry } from 'tar';
+import { promisify } from 'util';
 import {
   ReadTreeResponse,
   ReadTreeResponseDirOptions,
   ReadTreeResponseFile,
 } from '../types';
-import { pipeStream, stripFirstDirectoryFromPath } from './util';
+import { stripFirstDirectoryFromPath } from './util';
 
 // Tar types for `Parse` is not a proper constructor, but it should be
 const TarParseStream = Parse as unknown as { new (): ParseStream };
 
+const pipeline = promisify(pipelineCb);
+
 /**
  * Wraps a tar archive stream into a tree response reader.
  */
@@ -96,7 +99,7 @@ export class TarArchiveResponse implements ReadTreeResponse {
       }
 
       const content = new Promise<Buffer>(async resolve => {
-        await pipeStream(entry, concatStream(resolve));
+        await pipeline(entry, concatStream(resolve));
       });
 
       files.push({
@@ -107,7 +110,7 @@ export class TarArchiveResponse implements ReadTreeResponse {
       entry.resume();
     });
 
-    await pipeStream(this.stream, parser);
+    await pipeline(this.stream, parser);
 
     return files;
   }
@@ -125,7 +128,7 @@ export class TarArchiveResponse implements ReadTreeResponse {
 
     try {
       const data = await new Promise<Buffer>(async resolve => {
-        await pipeStream(
+        await pipeline(
           tar.create({ cwd: tmpDir }, ['']),
           concatStream(resolve),
         );
@@ -149,7 +152,7 @@ export class TarArchiveResponse implements ReadTreeResponse {
 
     let filterError: Error | undefined = undefined;
 
-    await pipeStream(
+    await pipeline(
       this.stream,
       tar.extract({
         strip,

packages/backend-common/src/reading/tree/util.test.ts

@@ -1,63 +0,0 @@
-/*
- * Copyright 2021 The Backstage Authors
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-import { Readable, Writable } from 'stream';
-import { pipeStream, streamToBuffer } from './util';
-
-describe('pipeStream', () => {
-  it('should pipe a stream', async () => {
-    const from = Readable.from(['hello']);
-
-    let written = '';
-    const to = new Writable({
-      write(chunk, encoding, callback) {
-        written = `${encoding}:${chunk}`;
-        callback();
-      },
-    });
-
-    await pipeStream(from, to);
-    expect(written).toBe('buffer:hello');
-  });
-
-  it('should forward errors', async () => {
-    const from = new Readable({
-      read() {
-        throw new Error('oh no');
-      },
-    });
-    const to = new Writable();
-
-    await expect(pipeStream(from, to)).rejects.toThrow('oh no');
-  });
-});
-
-describe('streamToBuffer', () => {
-  it('should read a stream', async () => {
-    await expect(streamToBuffer(Readable.from(['hello']))).resolves.toBe(
-      'hello',
-    );
-  });
-
-  it('should fail on errors', async () => {
-    const stream = new Readable({
-      read() {
-        throw new Error('oh no');
-      },
-    });
-    await expect(streamToBuffer(stream)).rejects.toThrow('oh no');
-  });
-});

packages/backend-common/src/reading/tree/util.ts

@@ -14,9 +14,12 @@
  * limitations under the License.
  */
 
-import { Readable, finished } from 'stream';
+import { Readable, pipeline as pipelineCb } from 'stream';
+import { promisify } from 'util';
 import concatStream from 'concat-stream';
 
+const pipeline = promisify(pipelineCb);
+
 // Matches a directory name + one `/` at the start of any string,
 // containing any character except `/` one or more times, and ending with a `/`
 // e.g. Will match `dirA/` in `dirA/dirB/file.ext`
@@ -26,37 +29,13 @@ export function stripFirstDirectoryFromPath(path: string): string {
   return path.replace(directoryNameRegex, '');
 }
 
-// Custom pipeline implementation, since pipeline doesn't work well with tar on node 18
-// See https://github.com/npm/node-tar/issues/321
-export function pipeStream(
-  from: NodeJS.ReadableStream,
-  to: NodeJS.WritableStream,
-): Promise<void> {
-  return new Promise((resolve, reject) => {
-    from.pipe(to);
-    finished(from, fromErr => {
-      if (fromErr) {
-        reject(fromErr);
-      } else {
-        finished(to, toErr => {
-          if (toErr) {
-            reject(toErr);
-          } else {
-            resolve();
-          }
-        });
-      }
-    });
-  });
-}
-
 // Collect the stream into a buffer and return
-export function streamToBuffer(stream: Readable): Promise<Buffer> {
+export const streamToBuffer = (stream: Readable): Promise<Buffer> => {
   return new Promise(async (resolve, reject) => {
     try {
-      await pipeStream(stream, concatStream(resolve));
+      await pipeline(stream, concatStream(resolve));
     } catch (ex) {
       reject(ex);
     }
   });
-}
+};