fix: Integrate wrapStreamError to prevent uncaught errors

src/init/Setup.ts

@@ -1,4 +1,3 @@
-import streamifyArray from 'streamify-array';
 import type { AclManager } from '../authorization/AclManager';
 import { RepresentationMetadata } from '../ldp/representation/RepresentationMetadata';
 import type { LoggerFactory } from '../logging/LoggerFactory';
@@ -6,6 +5,7 @@ import { getLoggerFor, setGlobalLoggerFactory } from '../logging/LogUtil';
 import type { ExpressHttpServerFactory } from '../server/ExpressHttpServerFactory';
 import type { ResourceStore } from '../storage/ResourceStore';
 import { TEXT_TURTLE } from '../util/ContentTypes';
+import { guardedStreamFrom } from '../util/StreamUtil';
 import { CONTENT_TYPE } from '../util/UriConstants';
 
 /**
@@ -65,7 +65,7 @@ export class Setup {
         baseAclId,
         {
           binary: true,
-          data: streamifyArray([ acl ]),
+          data: guardedStreamFrom([ acl ]),
           metadata,
         },
       );

src/ldp/http/response/OkResponseDescription.ts

@@ -1,4 +1,5 @@
 import type { Readable } from 'stream';
+import type { Guarded } from '../../../util/GuardedStream';
 import type { RepresentationMetadata } from '../../representation/RepresentationMetadata';
 import { ResponseDescription } from './ResponseDescription';
 
@@ -10,7 +11,7 @@ export class OkResponseDescription extends ResponseDescription {
    * @param metadata - Metadata concerning the response.
    * @param data - Potential data. @ignored
    */
-  public constructor(metadata: RepresentationMetadata, data?: Readable) {
+  public constructor(metadata: RepresentationMetadata, data?: Guarded<Readable>) {
     super(200, metadata, data);
   }
 }

src/ldp/http/response/ResponseDescription.ts

@@ -1,4 +1,5 @@
 import type { Readable } from 'stream';
+import type { Guarded } from '../../../util/GuardedStream';
 import type { RepresentationMetadata } from '../../representation/RepresentationMetadata';
 
 /**
@@ -7,14 +8,14 @@ import type { RepresentationMetadata } from '../../representation/Representation
 export class ResponseDescription {
   public readonly statusCode: number;
   public readonly metadata?: RepresentationMetadata;
-  public readonly data?: Readable;
+  public readonly data?: Guarded<Readable>;
 
   /**
    * @param statusCode - Status code to return.
    * @param metadata - Metadata corresponding to the response (and data potentially).
    * @param data - Data that needs to be returned. @ignored
    */
-  public constructor(statusCode: number, metadata?: RepresentationMetadata, data?: Readable) {
+  public constructor(statusCode: number, metadata?: RepresentationMetadata, data?: Guarded<Readable>) {
     this.statusCode = statusCode;
     this.metadata = metadata;
     this.data = data;

src/ldp/representation/Representation.ts

@@ -1,4 +1,5 @@
 import type { Readable } from 'stream';
+import type { Guarded } from '../../util/GuardedStream';
 import type { RepresentationMetadata } from './RepresentationMetadata';
 
 /**
@@ -12,7 +13,7 @@ export interface Representation {
   /**
    * The raw data stream for this representation.
    */
-  data: Readable;
+  data: Guarded<Readable>;
   /**
    * Whether the data stream consists of binary/string chunks
    * (as opposed to complex objects).

src/server/ExpressHttpServerFactory.ts

@@ -3,6 +3,7 @@ import cors from 'cors';
 import type { Express } from 'express';
 import express from 'express';
 import { getLoggerFor } from '../logging/LogUtil';
+import { guardStream } from '../util/GuardedStream';
 import type { HttpHandler } from './HttpHandler';
 import type { HttpServerFactory } from './HttpServerFactory';
 
@@ -40,7 +41,7 @@ export class ExpressHttpServerFactory implements HttpServerFactory {
     app.use(async(request, response, done): Promise<void> => {
       try {
         this.logger.info(`Received request for ${request.url}`);
-        await this.handler.handleSafe({ request, response });
+        await this.handler.handleSafe({ request: guardStream(request), response });
       } catch (error: unknown) {
         const errMsg = error instanceof Error ? `${error.name}: ${error.message}\n${error.stack}` : 'Unknown error.';
         this.logger.error(errMsg);

src/server/HttpRequest.ts

@@ -1,6 +1,7 @@
 import type { IncomingMessage } from 'http';
+import type { Guarded } from '../util/GuardedStream';
 
 /**
  * An incoming HTTP request;
  */
-export type HttpRequest = IncomingMessage;
+export type HttpRequest = Guarded<IncomingMessage>;

src/storage/DataAccessorBasedStore.ts

@@ -1,7 +1,6 @@
 import type { Readable } from 'stream';
 import { DataFactory } from 'n3';
 import type { Quad } from 'rdf-js';
-import streamifyArray from 'streamify-array';
 import { v4 as uuid } from 'uuid';
 import type { Representation } from '../ldp/representation/Representation';
 import { RepresentationMetadata } from '../ldp/representation/RepresentationMetadata';
@@ -12,6 +11,7 @@ import { MethodNotAllowedHttpError } from '../util/errors/MethodNotAllowedHttpEr
 import { NotFoundHttpError } from '../util/errors/NotFoundHttpError';
 import { NotImplementedError } from '../util/errors/NotImplementedError';
 import { UnsupportedHttpError } from '../util/errors/UnsupportedHttpError';
+import type { Guarded } from '../util/GuardedStream';
 import {
   ensureTrailingSlash,
   getParentContainer,
@@ -21,6 +21,7 @@ import {
 } from '../util/PathUtil';
 import { parseQuads } from '../util/QuadUtil';
 import { generateResourceQuads } from '../util/ResourceUtil';
+import { guardedStreamFrom } from '../util/StreamUtil';
 import { CONTENT_TYPE, HTTP, LDP, RDF } from '../util/UriConstants';
 import type { DataAccessor } from './accessors/DataAccessor';
 import type { ResourceStore } from './ResourceStore';
@@ -70,9 +71,9 @@ export class DataAccessorBasedStore implements ResourceStore {
       metadata.contentType = INTERNAL_QUADS;
       result = {
         binary: false,
-        get data(): Readable {
+        get data(): Guarded<Readable> {
           // This allows other modules to still add metadata before the output data is written
-          return streamifyArray(result.metadata.quads());
+          return guardedStreamFrom(result.metadata.quads());
         },
         metadata,
       };
@@ -365,7 +366,7 @@ export class DataAccessorBasedStore implements ResourceStore {
   protected getEmptyContainerRepresentation(container: ResourceIdentifier): Representation {
     return {
       binary: true,
-      data: streamifyArray([]),
+      data: guardedStreamFrom([]),
       metadata: new RepresentationMetadata(container.path),
     };
   }

src/storage/LockingResourceStore.ts

@@ -4,6 +4,7 @@ import type { Representation } from '../ldp/representation/Representation';
 import type { RepresentationPreferences } from '../ldp/representation/RepresentationPreferences';
 import type { ResourceIdentifier } from '../ldp/representation/ResourceIdentifier';
 import { getLoggerFor } from '../logging/LogUtil';
+import type { Guarded } from '../util/GuardedStream';
 import type { AtomicResourceStore } from './AtomicResourceStore';
 import type { Conditions } from './Conditions';
 import type { ExpiringLock } from './ExpiringLock';
@@ -118,7 +119,7 @@ export class LockingResourceStore implements AtomicResourceStore {
    * @param source - The readable to wrap
    * @param lock - The lock for the corresponding identifier.
    */
-  protected createExpiringReadable(source: Readable, lock: ExpiringLock): Readable {
+  protected createExpiringReadable(source: Guarded<Readable>, lock: ExpiringLock): Readable {
     // Destroy the source when a timeout occurs.
     lock.on('expired', (): void => {
       source.destroy(new Error(`Stream reading timout exceeded`));

src/storage/accessors/DataAccessor.ts

@@ -2,6 +2,7 @@ import type { Readable } from 'stream';
 import type { Representation } from '../../ldp/representation/Representation';
 import type { RepresentationMetadata } from '../../ldp/representation/RepresentationMetadata';
 import type { ResourceIdentifier } from '../../ldp/representation/ResourceIdentifier';
+import type { Guarded } from '../../util/GuardedStream';
 
 /**
  * A DataAccessor is the building block closest to the actual data storage.
@@ -27,7 +28,7 @@ export interface DataAccessor {
    * It can be assumed that the incoming identifier will always correspond to a document.
    * @param identifier - Identifier for which the data is requested.
    */
-  getData: (identifier: ResourceIdentifier) => Promise<Readable>;
+  getData: (identifier: ResourceIdentifier) => Promise<Guarded<Readable>>;
 
   /**
    * Returns the metadata corresponding to the identifier.
@@ -42,7 +43,8 @@ export interface DataAccessor {
    * @param data - Data to store.
    * @param metadata - Metadata to store.
    */
-  writeDocument: (identifier: ResourceIdentifier, data: Readable, metadata: RepresentationMetadata) => Promise<void>;
+  writeDocument: (identifier: ResourceIdentifier, data: Guarded<Readable>, metadata: RepresentationMetadata) =>
+  Promise<void>;
 
   /**
    * Writes metadata for a container.

src/storage/accessors/FileDataAccessor.ts

@@ -12,6 +12,8 @@ import { ConflictHttpError } from '../../util/errors/ConflictHttpError';
 import { NotFoundHttpError } from '../../util/errors/NotFoundHttpError';
 import { isSystemError } from '../../util/errors/SystemError';
 import { UnsupportedMediaTypeHttpError } from '../../util/errors/UnsupportedMediaTypeHttpError';
+import { guardStream } from '../../util/GuardedStream';
+import type { Guarded } from '../../util/GuardedStream';
 import { isContainerIdentifier } from '../../util/PathUtil';
 import { parseQuads, pushQuad, serializeQuads } from '../../util/QuadUtil';
 import { generateContainmentQuads, generateResourceQuads } from '../../util/ResourceUtil';
@@ -45,12 +47,12 @@ export class FileDataAccessor implements DataAccessor {
    * Will return data stream directly to the file corresponding to the resource.
    * Will throw NotFoundHttpError if the input is a container.
    */
-  public async getData(identifier: ResourceIdentifier): Promise<Readable> {
+  public async getData(identifier: ResourceIdentifier): Promise<Guarded<Readable>> {
     const link = await this.resourceMapper.mapUrlToFilePath(identifier);
     const stats = await this.getStats(link.filePath);
 
     if (stats.isFile()) {
-      return createReadStream(link.filePath);
+      return guardStream(createReadStream(link.filePath));
     }
 
     throw new NotFoundHttpError();
@@ -76,7 +78,7 @@ export class FileDataAccessor implements DataAccessor {
    * Writes the given data as a file (and potential metadata as additional file).
    * The metadata file will be written first and will be deleted if something goes wrong writing the actual data.
    */
-  public async writeDocument(identifier: ResourceIdentifier, data: Readable, metadata: RepresentationMetadata):
+  public async writeDocument(identifier: ResourceIdentifier, data: Guarded<Readable>, metadata: RepresentationMetadata):
   Promise<void> {
     if (this.isMetadataPath(identifier.path)) {
       throw new ConflictHttpError('Not allowed to create files with the metadata extension.');
@@ -264,7 +266,7 @@ export class FileDataAccessor implements DataAccessor {
       // Check if the metadata file exists first
       await fsPromises.lstat(metadataPath);
 
-      const readMetadataStream = createReadStream(metadataPath);
+      const readMetadataStream = guardStream(createReadStream(metadataPath));
       return await parseQuads(readMetadataStream);
     } catch (error: unknown) {
       // Metadata file doesn't exist so lets keep `rawMetaData` an empty array.

src/storage/accessors/InMemoryDataAccessor.ts

@@ -1,12 +1,14 @@
-import { Readable } from 'stream';
+import type { Readable } from 'stream';
 import arrayifyStream from 'arrayify-stream';
 import { DataFactory } from 'n3';
 import type { NamedNode } from 'rdf-js';
 import { RepresentationMetadata } from '../../ldp/representation/RepresentationMetadata';
 import type { ResourceIdentifier } from '../../ldp/representation/ResourceIdentifier';
 import { NotFoundHttpError } from '../../util/errors/NotFoundHttpError';
+import type { Guarded } from '../../util/GuardedStream';
 import { ensureTrailingSlash, isContainerIdentifier } from '../../util/PathUtil';
 import { generateContainmentQuads, generateResourceQuads } from '../../util/ResourceUtil';
+import { guardedStreamFrom } from '../../util/StreamUtil';
 import type { DataAccessor } from './DataAccessor';
 
 interface DataEntry {
@@ -19,27 +21,6 @@ interface ContainerEntry {
 }
 type CacheEntry = DataEntry | ContainerEntry;
 
-class ArrayReadable extends Readable {
-  private readonly data: any[];
-  private idx: number;
-
-  public constructor(data: any[]) {
-    super({ objectMode: true });
-    this.data = data;
-    this.idx = 0;
-  }
-
-  // eslint-disable-next-line @typescript-eslint/naming-convention
-  public _read(): void {
-    if (this.idx < this.data.length) {
-      this.push(this.data[this.idx]);
-      this.idx += 1;
-    } else {
-      this.push(null);
-    }
-  }
-}
-
 export class InMemoryDataAccessor implements DataAccessor {
   private readonly base: string;
   private readonly store: ContainerEntry;
@@ -56,12 +37,12 @@ export class InMemoryDataAccessor implements DataAccessor {
     // All data is supported since streams never get read, only copied
   }
 
-  public async getData(identifier: ResourceIdentifier): Promise<Readable> {
+  public async getData(identifier: ResourceIdentifier): Promise<Guarded<Readable>> {
     const entry = this.getEntry(identifier);
     if (!this.isDataEntry(entry)) {
       throw new NotFoundHttpError();
     }
-    return new ArrayReadable(entry.data);
+    return guardedStreamFrom(entry.data);
   }
 
   public async getMetadata(identifier: ResourceIdentifier): Promise<RepresentationMetadata> {
@@ -72,7 +53,7 @@ export class InMemoryDataAccessor implements DataAccessor {
     return this.generateMetadata(identifier, entry);
   }
 
-  public async writeDocument(identifier: ResourceIdentifier, data: Readable, metadata: RepresentationMetadata):
+  public async writeDocument(identifier: ResourceIdentifier, data: Guarded<Readable>, metadata: RepresentationMetadata):
   Promise<void> {
     const { parent, name } = this.getParentEntry(identifier);
     parent.entries[name] = {

src/storage/accessors/SparqlDataAccessor.ts

@@ -23,6 +23,8 @@ import { ConflictHttpError } from '../../util/errors/ConflictHttpError';
 import { NotFoundHttpError } from '../../util/errors/NotFoundHttpError';
 import { UnsupportedHttpError } from '../../util/errors/UnsupportedHttpError';
 import { UnsupportedMediaTypeHttpError } from '../../util/errors/UnsupportedMediaTypeHttpError';
+import { guardStream } from '../../util/GuardedStream';
+import type { Guarded } from '../../util/GuardedStream';
 import { ensureTrailingSlash, getParentContainer, isContainerIdentifier } from '../../util/PathUtil';
 import { generateResourceQuads } from '../../util/ResourceUtil';
 import { CONTENT_TYPE, LDP } from '../../util/UriConstants';
@@ -70,9 +72,9 @@ export class SparqlDataAccessor implements DataAccessor {
    * Returns all triples stored for the corresponding identifier.
    * Note that this will not throw a 404 if no results were found.
    */
-  public async getData(identifier: ResourceIdentifier): Promise<Readable> {
+  public async getData(identifier: ResourceIdentifier): Promise<Guarded<Readable>> {
     const name = namedNode(identifier.path);
-    return this.sendSparqlConstruct(this.sparqlConstruct(name));
+    return await this.sendSparqlConstruct(this.sparqlConstruct(name));
   }
 
   /**
@@ -114,7 +116,7 @@ export class SparqlDataAccessor implements DataAccessor {
   /**
    * Reads the given data stream and stores it together with the metadata.
    */
-  public async writeDocument(identifier: ResourceIdentifier, data: Readable, metadata: RepresentationMetadata):
+  public async writeDocument(identifier: ResourceIdentifier, data: Guarded<Readable>, metadata: RepresentationMetadata):
   Promise<void> {
     if (this.isMetadataIdentifier(identifier)) {
       throw new ConflictHttpError('Not allowed to create NamedNodes with the metadata extension.');
@@ -292,11 +294,11 @@ export class SparqlDataAccessor implements DataAccessor {
    * Sends a SPARQL CONSTRUCT query to the endpoint and returns a stream of quads.
    * @param sparqlQuery - Query to execute.
    */
-  private async sendSparqlConstruct(sparqlQuery: ConstructQuery): Promise<Readable> {
+  private async sendSparqlConstruct(sparqlQuery: ConstructQuery): Promise<Guarded<Readable>> {
     const query = this.generator.stringify(sparqlQuery);
     this.logger.info(`Sending SPARQL CONSTRUCT query to ${this.endpoint}: ${query}`);
     try {
-      return await this.fetcher.fetchTriples(this.endpoint, query);
+      return guardStream(await this.fetcher.fetchTriples(this.endpoint, query));
     } catch (error: unknown) {
       if (error instanceof Error) {
         this.logger.error(`SPARQL endpoint ${this.endpoint} error: ${error.message}`);

src/storage/conversion/QuadToRdfConverter.ts

@@ -4,6 +4,7 @@ import type { Representation } from '../../ldp/representation/Representation';
 import { RepresentationMetadata } from '../../ldp/representation/RepresentationMetadata';
 import type { RepresentationPreferences } from '../../ldp/representation/RepresentationPreferences';
 import { INTERNAL_QUADS } from '../../util/ContentTypes';
+import { guardStream } from '../../util/GuardedStream';
 import { CONTENT_TYPE } from '../../util/UriConstants';
 import { validateRequestArgs, matchingTypes } from './ConversionUtil';
 import type { RepresentationConverterArgs } from './RepresentationConverter';
@@ -34,7 +35,7 @@ export class QuadToRdfConverter extends TypedRepresentationConverter {
     const metadata = new RepresentationMetadata(quads.metadata, { [CONTENT_TYPE]: contentType });
     return {
       binary: true,
-      data: rdfSerializer.serialize(quads.data, { contentType }) as Readable,
+      data: guardStream(rdfSerializer.serialize(quads.data, { contentType }) as Readable),
       metadata,
     };
   }

src/storage/patch/SparqlUpdatePatchHandler.ts

@@ -11,6 +11,7 @@ import type { ResourceIdentifier } from '../../ldp/representation/ResourceIdenti
 import { getLoggerFor } from '../../logging/LogUtil';
 import { INTERNAL_QUADS } from '../../util/ContentTypes';
 import { UnsupportedHttpError } from '../../util/errors/UnsupportedHttpError';
+import { guardStream } from '../../util/GuardedStream';
 import { CONTENT_TYPE } from '../../util/UriConstants';
 import type { ResourceLocker } from '../ResourceLocker';
 import type { ResourceStore } from '../ResourceStore';
@@ -77,7 +78,7 @@ export class SparqlUpdatePatchHandler extends PatchHandler {
     const metadata = new RepresentationMetadata(input.identifier.path, { [CONTENT_TYPE]: INTERNAL_QUADS });
     const representation: Representation = {
       binary: false,
-      data: store.match() as Readable,
+      data: guardStream(store.match() as Readable),
       metadata,
     };
     await this.source.setRepresentation(input.identifier, representation);

src/util/QuadUtil.ts

@@ -4,6 +4,7 @@ import { DataFactory, StreamParser, StreamWriter } from 'n3';
 import type { Literal, NamedNode, Quad } from 'rdf-js';
 import streamifyArray from 'streamify-array';
 import { TEXT_TURTLE } from './ContentTypes';
+import type { Guarded } from './GuardedStream';
 import { pipeSafely } from './StreamUtil';
 
 /**
@@ -19,7 +20,7 @@ export const pushQuad =
  *
  * @returns The Readable object.
  */
-export const serializeQuads = (quads: Quad[]): Readable =>
+export const serializeQuads = (quads: Quad[]): Guarded<Readable> =>
   pipeSafely(streamifyArray(quads), new StreamWriter({ format: TEXT_TURTLE }));
 
 /**
@@ -28,5 +29,5 @@ export const serializeQuads = (quads: Quad[]): Readable =>
  *
  * @returns A promise containing the array of quads.
  */
-export const parseQuads = async(readable: Readable): Promise<Quad[]> =>
+export const parseQuads = async(readable: Guarded<Readable>): Promise<Quad[]> =>
   arrayifyStream(pipeSafely(readable, new StreamParser({ format: TEXT_TURTLE })));

test/integration/FullConfig.acl.test.ts

@@ -5,6 +5,7 @@ import { RepresentationMetadata } from '../../src/ldp/representation/Representat
 import { FileDataAccessor } from '../../src/storage/accessors/FileDataAccessor';
 import { InMemoryDataAccessor } from '../../src/storage/accessors/InMemoryDataAccessor';
 import { ExtensionBasedMapper } from '../../src/storage/mapping/ExtensionBasedMapper';
+import { guardStream } from '../../src/util/GuardedStream';
 import { ensureTrailingSlash } from '../../src/util/PathUtil';
 import { CONTENT_TYPE, LDP } from '../../src/util/UriConstants';
 import { AuthenticatedDataAccessorBasedConfig } from '../configs/AuthenticatedDataAccessorBasedConfig';
@@ -42,7 +43,7 @@ describe.each([ dataAccessorStore, inMemoryDataAccessorStore ])('A server using
       // Use store instead of file access so tests also work for non-file backends
       await config.store.setRepresentation({ path: `${BASE}/permanent.txt` }, {
         binary: true,
-        data: createReadStream(join(__dirname, '../assets/permanent.txt')),
+        data: guardStream(createReadStream(join(__dirname, '../assets/permanent.txt'))),
         metadata: new RepresentationMetadata({ [CONTENT_TYPE]: 'text/plain' }),
       });
     });

test/integration/RepresentationConverter.test.ts

@@ -1,10 +1,9 @@
-import streamifyArray from 'streamify-array';
 import type { Representation } from '../../src/ldp/representation/Representation';
 import { RepresentationMetadata } from '../../src/ldp/representation/RepresentationMetadata';
 import { ChainedConverter } from '../../src/storage/conversion/ChainedConverter';
 import { QuadToRdfConverter } from '../../src/storage/conversion/QuadToRdfConverter';
 import { RdfToQuadConverter } from '../../src/storage/conversion/RdfToQuadConverter';
-import { readableToString } from '../../src/util/StreamUtil';
+import { guardedStreamFrom, readableToString } from '../../src/util/StreamUtil';
 import { CONTENT_TYPE } from '../../src/util/UriConstants';
 
 describe('A ChainedConverter', (): void => {
@@ -18,7 +17,9 @@ describe('A ChainedConverter', (): void => {
     const metadata = new RepresentationMetadata({ [CONTENT_TYPE]: 'application/ld+json' });
     const representation: Representation = {
       binary: true,
-      data: streamifyArray([ '{"@id": "http://test.com/s", "http://test.com/p": { "@id": "http://test.com/o" }}' ]),
+      data: guardedStreamFrom(
+        [ '{"@id": "http://test.com/s", "http://test.com/p": { "@id": "http://test.com/o" }}' ],
+      ),
       metadata,
     };
 
@@ -36,7 +37,7 @@ describe('A ChainedConverter', (): void => {
     const metadata = new RepresentationMetadata({ [CONTENT_TYPE]: 'text/turtle' });
     const representation: Representation = {
       binary: true,
-      data: streamifyArray([ '<http://test.com/s> <http://test.com/p> <http://test.com/o>.' ]),
+      data: guardedStreamFrom([ '<http://test.com/s> <http://test.com/p> <http://test.com/o>.' ]),
       metadata,
     };
 

test/unit/ldp/http/BasicResponseWriter.test.ts

@@ -2,13 +2,13 @@ import { EventEmitter } from 'events';
 import { PassThrough } from 'stream';
 import type { MockResponse } from 'node-mocks-http';
 import { createResponse } from 'node-mocks-http';
-import streamifyArray from 'streamify-array';
 import { BasicResponseWriter } from '../../../../src/ldp/http/BasicResponseWriter';
 import type { MetadataWriter } from '../../../../src/ldp/http/metadata/MetadataWriter';
 import type { ResponseDescription } from '../../../../src/ldp/http/response/ResponseDescription';
 import { RepresentationMetadata } from '../../../../src/ldp/representation/RepresentationMetadata';
 import { INTERNAL_QUADS } from '../../../../src/util/ContentTypes';
 import { UnsupportedHttpError } from '../../../../src/util/errors/UnsupportedHttpError';
+import { guardedStreamFrom } from '../../../../src/util/StreamUtil';
 import { CONTENT_TYPE } from '../../../../src/util/UriConstants';
 import { StaticAsyncHandler } from '../../../util/StaticAsyncHandler';
 
@@ -42,7 +42,7 @@ describe('A BasicResponseWriter', (): void => {
   });
 
   it('responds with a body if the description has a body.', async(): Promise<void> => {
-    const data = streamifyArray([ '<http://test.com/s> <http://test.com/p> <http://test.com/o>.' ]);
+    const data = guardedStreamFrom([ '<http://test.com/s> <http://test.com/p> <http://test.com/o>.' ]);
     result = { statusCode: 201, data };
 
     const end = new Promise((resolve): void => {
@@ -69,7 +69,7 @@ describe('A BasicResponseWriter', (): void => {
   });
 
   it('can handle the data stream erroring.', async(): Promise<void> => {
-    const data = new PassThrough();
+    const data = guardedStreamFrom([]);
     data.read = (): any => {
       data.emit('error', new Error('bad data!'));
       return null;

test/unit/storage/DataAccessorBasedStore.test.ts

@@ -1,7 +1,6 @@
 import type { Readable } from 'stream';
 import arrayifyStream from 'arrayify-stream';
 import { DataFactory } from 'n3';
-import streamifyArray from 'streamify-array';
 import type { Representation } from '../../../src/ldp/representation/Representation';
 import { RepresentationMetadata } from '../../../src/ldp/representation/RepresentationMetadata';
 import type { ResourceIdentifier } from '../../../src/ldp/representation/ResourceIdentifier';
@@ -13,7 +12,9 @@ import { MethodNotAllowedHttpError } from '../../../src/util/errors/MethodNotAll
 import { NotFoundHttpError } from '../../../src/util/errors/NotFoundHttpError';
 import { NotImplementedError } from '../../../src/util/errors/NotImplementedError';
 import { UnsupportedHttpError } from '../../../src/util/errors/UnsupportedHttpError';
+import type { Guarded } from '../../../src/util/GuardedStream';
 import * as quadUtil from '../../../src/util/QuadUtil';
+import { guardedStreamFrom } from '../../../src/util/StreamUtil';
 import { CONTENT_TYPE, HTTP, LDP, RDF } from '../../../src/util/UriConstants';
 import { toNamedNode } from '../../../src/util/UriUtil';
 
@@ -39,7 +40,7 @@ class SimpleDataAccessor implements DataAccessor {
     return undefined;
   }
 
-  public async getData(identifier: ResourceIdentifier): Promise<Readable> {
+  public async getData(identifier: ResourceIdentifier): Promise<Guarded<Readable>> {
     this.checkExists(identifier);
     return this.data[identifier.path].data;
   }
@@ -83,11 +84,11 @@ describe('A DataAccessorBasedStore', (): void => {
 
     representation = {
       binary: true,
-      data: streamifyArray([ resourceData ]),
+      data: guardedStreamFrom([ resourceData ]),
       metadata: new RepresentationMetadata(
         { [CONTENT_TYPE]: 'text/plain', [RDF.type]: DataFactory.namedNode(LDP.Resource) },
       ),
-    } as Representation;
+    };
   });
 
   describe('getting a Representation', (): void => {
@@ -176,7 +177,7 @@ describe('A DataAccessorBasedStore', (): void => {
       const resourceID = { path: root };
       representation.metadata.add(RDF.type, toNamedNode(LDP.Container));
       representation.metadata.contentType = 'text/turtle';
-      representation.data = streamifyArray([ `<${`${root}resource/`}> a <coolContainer>.` ]);
+      representation.data = guardedStreamFrom([ `<${`${root}resource/`}> a <coolContainer>.` ]);
       const result = await store.addResource(resourceID, representation);
       expect(result).toEqual({
         path: expect.stringMatching(new RegExp(`^${root}[^/]+/$`, 'u')),
@@ -287,7 +288,7 @@ describe('A DataAccessorBasedStore', (): void => {
       // Generate based on URI
       representation.metadata.removeAll(RDF.type);
       representation.metadata.contentType = 'text/turtle';
-      representation.data = streamifyArray([ `<${`${root}resource/`}> a <coolContainer>.` ]);
+      representation.data = guardedStreamFrom([ `<${`${root}resource/`}> a <coolContainer>.` ]);
       await expect(store.setRepresentation(resourceID, representation)).resolves.toBeUndefined();
       expect(accessor.data[resourceID.path]).toBeTruthy();
       expect(accessor.data[resourceID.path].metadata.contentType).toBeUndefined();
@@ -298,7 +299,9 @@ describe('A DataAccessorBasedStore', (): void => {
       representation.metadata.add(RDF.type, toNamedNode(LDP.Container));
       representation.metadata.contentType = 'text/turtle';
       representation.metadata.identifier = DataFactory.namedNode(`${root}resource/`);
-      representation.data = streamifyArray([ `<${`${root}resource/`}> <http://www.w3.org/ns/ldp#contains> <uri>.` ]);
+      representation.data = guardedStreamFrom(
+        [ `<${`${root}resource/`}> <http://www.w3.org/ns/ldp#contains> <uri>.` ],
+      );
       await expect(store.setRepresentation(resourceID, representation))
         .rejects.toThrow(new ConflictHttpError('Container bodies are not allowed to have containment triples.'));
     });

test/unit/storage/accessors/FileDataAccessor.test.ts

@@ -1,5 +1,5 @@
+import type { Readable } from 'stream';
 import { DataFactory } from 'n3';
-import streamifyArray from 'streamify-array';
 import type { Representation } from '../../../../src/ldp/representation/Representation';
 import { RepresentationMetadata } from '../../../../src/ldp/representation/RepresentationMetadata';
 import { FileDataAccessor } from '../../../../src/storage/accessors/FileDataAccessor';
@@ -9,7 +9,8 @@ import { ConflictHttpError } from '../../../../src/util/errors/ConflictHttpError
 import { NotFoundHttpError } from '../../../../src/util/errors/NotFoundHttpError';
 import type { SystemError } from '../../../../src/util/errors/SystemError';
 import { UnsupportedMediaTypeHttpError } from '../../../../src/util/errors/UnsupportedMediaTypeHttpError';
-import { readableToString } from '../../../../src/util/StreamUtil';
+import type { Guarded } from '../../../../src/util/GuardedStream';
+import { guardedStreamFrom, readableToString } from '../../../../src/util/StreamUtil';
 import { CONTENT_TYPE, DCTERMS, LDP, POSIX, RDF, XSD } from '../../../../src/util/UriConstants';
 import { toNamedNode, toTypedLiteral } from '../../../../src/util/UriUtil';
 import { mockFs } from '../../../util/Util';
@@ -24,12 +25,15 @@ describe('A FileDataAccessor', (): void => {
   let accessor: FileDataAccessor;
   let cache: { data: any };
   let metadata: RepresentationMetadata;
+  let data: Guarded<Readable>;
 
   beforeEach(async(): Promise<void> => {
     cache = mockFs(rootFilePath, now);
     accessor = new FileDataAccessor(new ExtensionBasedMapper(base, rootFilePath));
 
     metadata = new RepresentationMetadata({ [CONTENT_TYPE]: APPLICATION_OCTET_STREAM });
+
+    data = guardedStreamFrom([ 'data' ]);
   });
 
   it('can only handle binary data.', async(): Promise<void> => {
@@ -140,23 +144,21 @@ describe('A FileDataAccessor', (): void => {
 
   describe('writing a document', (): void => {
     it('throws a 404 if the identifier does not start with the base.', async(): Promise<void> => {
-      await expect(accessor.writeDocument({ path: 'badpath' }, streamifyArray([]), metadata))
+      await expect(accessor.writeDocument({ path: 'badpath' }, data, metadata))
         .rejects.toThrow(NotFoundHttpError);
     });
 
     it('throws an error when writing to a metadata path.', async(): Promise<void> => {
-      await expect(accessor.writeDocument({ path: `${base}resource.meta` }, streamifyArray([]), metadata))
+      await expect(accessor.writeDocument({ path: `${base}resource.meta` }, data, metadata))
         .rejects.toThrow(new ConflictHttpError('Not allowed to create files with the metadata extension.'));
     });
 
     it('writes the data to the corresponding file.', async(): Promise<void> => {
-      const data = streamifyArray([ 'data' ]);
       await expect(accessor.writeDocument({ path: `${base}resource` }, data, metadata)).resolves.toBeUndefined();
       expect(cache.data.resource).toBe('data');
     });
 
     it('writes metadata to the corresponding metadata file.', async(): Promise<void> => {
-      const data = streamifyArray([ 'data' ]);
       metadata = new RepresentationMetadata(`${base}res.ttl`, { [CONTENT_TYPE]: 'text/turtle', likes: 'apples' });
       await expect(accessor.writeDocument({ path: `${base}res.ttl` }, data, metadata)).resolves.toBeUndefined();
       expect(cache.data['res.ttl']).toBe('data');
@@ -164,7 +166,6 @@ describe('A FileDataAccessor', (): void => {
     });
 
     it('does not write metadata that is stored by the file system.', async(): Promise<void> => {
-      const data = streamifyArray([ 'data' ]);
       metadata.add(RDF.type, toNamedNode(LDP.Resource));
       await expect(accessor.writeDocument({ path: `${base}resource` }, data, metadata)).resolves.toBeUndefined();
       expect(cache.data.resource).toBe('data');
@@ -173,7 +174,6 @@ describe('A FileDataAccessor', (): void => {
 
     it('deletes existing metadata if nothing new needs to be stored.', async(): Promise<void> => {
       cache.data = { resource: 'data', 'resource.meta': 'metadata!' };
-      const data = streamifyArray([ 'data' ]);
       await expect(accessor.writeDocument({ path: `${base}resource` }, data, metadata)).resolves.toBeUndefined();
       expect(cache.data.resource).toBe('data');
       expect(cache.data['resource.meta']).toBeUndefined();
@@ -184,13 +184,11 @@ describe('A FileDataAccessor', (): void => {
       jest.requireMock('fs').promises.unlink = (): any => {
         throw new Error('error');
       };
-      const data = streamifyArray([ 'data' ]);
       await expect(accessor.writeDocument({ path: `${base}resource` }, data, metadata))
         .rejects.toThrow(new Error('error'));
     });
 
     it('throws if something went wrong writing a file.', async(): Promise<void> => {
-      const data = streamifyArray([ 'data' ]);
       data.read = (): any => {
         data.emit('error', new Error('error'));
         return null;
@@ -200,7 +198,6 @@ describe('A FileDataAccessor', (): void => {
     });
 
     it('deletes the metadata file if something went wrong writing the file.', async(): Promise<void> => {
-      const data = streamifyArray([ 'data' ]);
       data.read = (): any => {
         data.emit('error', new Error('error'));
         return null;
@@ -216,10 +213,10 @@ describe('A FileDataAccessor', (): void => {
       metadata.identifier = DataFactory.namedNode(`${base}resource`);
       metadata.contentType = 'text/plain';
       metadata.add('new', 'metadata');
-      await expect(accessor.writeDocument({ path: `${base}resource` }, streamifyArray([ 'text' ]), metadata))
+      await expect(accessor.writeDocument({ path: `${base}resource` }, data, metadata))
         .resolves.toBeUndefined();
       expect(cache.data).toEqual({
-        'resource$.txt': 'text',
+        'resource$.txt': 'data',
         'resource.meta': expect.stringMatching(`<${base}resource> <new> "metadata".`),
       });
     });
@@ -235,11 +232,11 @@ describe('A FileDataAccessor', (): void => {
 
       // `unlink` throwing ENOENT should not be an issue if the content-type does not change
       metadata.contentType = 'text/turtle';
-      await expect(accessor.writeDocument({ path: `${base}resource` }, streamifyArray([ 'text' ]), metadata))
+      await expect(accessor.writeDocument({ path: `${base}resource` }, data, metadata))
         .resolves.toBeUndefined();
 
       metadata.contentType = 'text/plain';
-      await expect(accessor.writeDocument({ path: `${base}resource` }, streamifyArray([ 'text' ]), metadata))
+      await expect(accessor.writeDocument({ path: `${base}resource` }, data, metadata))
         .rejects.toThrow(new Error('error'));
     });
   });

test/unit/storage/accessors/InMemoryDataAccessor.test.ts

@@ -1,9 +1,10 @@
-import streamifyArray from 'streamify-array';
+import type { Readable } from 'stream';
 import { RepresentationMetadata } from '../../../../src/ldp/representation/RepresentationMetadata';
 import { InMemoryDataAccessor } from '../../../../src/storage/accessors/InMemoryDataAccessor';
 import { APPLICATION_OCTET_STREAM } from '../../../../src/util/ContentTypes';
 import { NotFoundHttpError } from '../../../../src/util/errors/NotFoundHttpError';
-import { readableToString } from '../../../../src/util/StreamUtil';
+import type { Guarded } from '../../../../src/util/GuardedStream';
+import { guardedStreamFrom, readableToString } from '../../../../src/util/StreamUtil';
 import { CONTENT_TYPE, LDP, RDF } from '../../../../src/util/UriConstants';
 import { toNamedNode } from '../../../../src/util/UriUtil';
 
@@ -11,11 +12,14 @@ describe('An InMemoryDataAccessor', (): void => {
   const base = 'http://test.com/';
   let accessor: InMemoryDataAccessor;
   let metadata: RepresentationMetadata;
+  let data: Guarded<Readable>;
 
   beforeEach(async(): Promise<void> => {
     accessor = new InMemoryDataAccessor(base);
 
     metadata = new RepresentationMetadata({ [CONTENT_TYPE]: APPLICATION_OCTET_STREAM });
+
+    data = guardedStreamFrom([ 'data' ]);
   });
 
   it('can only handle all data.', async(): Promise<void> => {
@@ -33,12 +37,11 @@ describe('An InMemoryDataAccessor', (): void => {
     });
 
     it('throws an error if part of the path matches a document.', async(): Promise<void> => {
-      await accessor.writeDocument({ path: `${base}resource` }, streamifyArray([ 'data' ]), metadata);
+      await accessor.writeDocument({ path: `${base}resource` }, data, metadata);
       await expect(accessor.getData({ path: `${base}resource/resource2` })).rejects.toThrow(new Error('Invalid path.'));
     });
 
     it('returns the corresponding data every time.', async(): Promise<void> => {
-      const data = streamifyArray([ 'data' ]);
       await accessor.writeDocument({ path: `${base}resource` }, data, metadata);
 
       // Run twice to make sure the data is stored correctly
@@ -53,12 +56,12 @@ describe('An InMemoryDataAccessor', (): void => {
     });
 
     it('errors when trying to access the parent of root.', async(): Promise<void> => {
-      await expect(accessor.writeDocument({ path: `${base}` }, streamifyArray([ 'data' ]), metadata))
+      await expect(accessor.writeDocument({ path: `${base}` }, data, metadata))
         .rejects.toThrow(new Error('Root container has no parent.'));
     });
 
     it('throws a 404 if the trailing slash does not match its type.', async(): Promise<void> => {
-      await accessor.writeDocument({ path: `${base}resource` }, streamifyArray([ 'data' ]), metadata);
+      await accessor.writeDocument({ path: `${base}resource` }, data, metadata);
       await expect(accessor.getMetadata({ path: `${base}resource/` })).rejects.toThrow(NotFoundHttpError);
       await accessor.writeContainer({ path: `${base}container/` }, metadata);
       await expect(accessor.getMetadata({ path: `${base}container` })).rejects.toThrow(NotFoundHttpError);
@@ -66,14 +69,14 @@ describe('An InMemoryDataAccessor', (): void => {
 
     it('returns empty metadata if there was none stored.', async(): Promise<void> => {
       metadata = new RepresentationMetadata();
-      await accessor.writeDocument({ path: `${base}resource` }, streamifyArray([ 'data' ]), metadata);
+      await accessor.writeDocument({ path: `${base}resource` }, data, metadata);
       metadata = await accessor.getMetadata({ path: `${base}resource` });
       expect(metadata.quads()).toHaveLength(0);
     });
 
     it('generates the containment metadata for a container.', async(): Promise<void> => {
       await accessor.writeContainer({ path: `${base}container/` }, metadata);
-      await accessor.writeDocument({ path: `${base}container/resource` }, streamifyArray([ 'data' ]), metadata);
+      await accessor.writeDocument({ path: `${base}container/resource` }, data, metadata);
       await accessor.writeContainer({ path: `${base}container/container2` }, metadata);
       metadata = await accessor.getMetadata({ path: `${base}container/` });
       expect(metadata.getAll(LDP.contains)).toEqualRdfTermArray(
@@ -83,7 +86,7 @@ describe('An InMemoryDataAccessor', (): void => {
 
     it('adds stored metadata when requesting document metadata.', async(): Promise<void> => {
       const inputMetadata = new RepresentationMetadata(`${base}resource`, { [RDF.type]: toNamedNode(LDP.Resource) });
-      await accessor.writeDocument({ path: `${base}resource` }, streamifyArray([ 'data' ]), inputMetadata);
+      await accessor.writeDocument({ path: `${base}resource` }, data, inputMetadata);
       metadata = await accessor.getMetadata({ path: `${base}resource` });
       expect(metadata.identifier.value).toBe(`${base}resource`);
       const quads = metadata.quads();
@@ -107,7 +110,7 @@ describe('An InMemoryDataAccessor', (): void => {
       await accessor.writeContainer({ path: `${base}container/` }, inputMetadata);
       const resourceMetadata = new RepresentationMetadata();
       await accessor.writeDocument(
-        { path: `${base}container/resource` }, streamifyArray([ 'data' ]), resourceMetadata,
+        { path: `${base}container/resource` }, data, resourceMetadata,
       );
 
       const newMetadata = new RepresentationMetadata(inputMetadata);
@@ -128,7 +131,7 @@ describe('An InMemoryDataAccessor', (): void => {
     });
 
     it('errors when writing to an invalid container path..', async(): Promise<void> => {
-      await accessor.writeDocument({ path: `${base}resource` }, streamifyArray([ 'data' ]), metadata);
+      await accessor.writeDocument({ path: `${base}resource` }, data, metadata);
 
       await expect(accessor.writeContainer({ path: `${base}resource/container` }, metadata))
         .rejects.toThrow(new Error('Invalid path.'));
@@ -141,7 +144,7 @@ describe('An InMemoryDataAccessor', (): void => {
     });
 
     it('removes the corresponding resource.', async(): Promise<void> => {
-      await accessor.writeDocument({ path: `${base}resource` }, streamifyArray([ 'data' ]), metadata);
+      await accessor.writeDocument({ path: `${base}resource` }, data, metadata);
       await accessor.writeContainer({ path: `${base}container/` }, metadata);
       await expect(accessor.deleteResource({ path: `${base}resource` })).resolves.toBeUndefined();
       await expect(accessor.deleteResource({ path: `${base}container/` })).resolves.toBeUndefined();

test/unit/storage/accessors/SparqlDataAccessor.test.ts

@@ -1,9 +1,8 @@
-import type { Readable } from 'stream';
+import { Readable } from 'stream';
 import arrayifyStream from 'arrayify-stream';
 import { SparqlEndpointFetcher } from 'fetch-sparql-endpoint';
 import { DataFactory } from 'n3';
 import type { Quad } from 'rdf-js';
-import streamifyArray from 'streamify-array';
 import { RepresentationMetadata } from '../../../../src/ldp/representation/RepresentationMetadata';
 import { SparqlDataAccessor } from '../../../../src/storage/accessors/SparqlDataAccessor';
 import { INTERNAL_QUADS } from '../../../../src/util/ContentTypes';
@@ -11,6 +10,8 @@ import { ConflictHttpError } from '../../../../src/util/errors/ConflictHttpError
 import { NotFoundHttpError } from '../../../../src/util/errors/NotFoundHttpError';
 import { UnsupportedHttpError } from '../../../../src/util/errors/UnsupportedHttpError';
 import { UnsupportedMediaTypeHttpError } from '../../../../src/util/errors/UnsupportedMediaTypeHttpError';
+import type { Guarded } from '../../../../src/util/GuardedStream';
+import { guardedStreamFrom } from '../../../../src/util/StreamUtil';
 import { CONTENT_TYPE, LDP, RDF } from '../../../../src/util/UriConstants';
 import { toNamedNode } from '../../../../src/util/UriUtil';
 
@@ -30,6 +31,7 @@ describe('A SparqlDataAccessor', (): void => {
   const base = 'http://test.com/';
   let accessor: SparqlDataAccessor;
   let metadata: RepresentationMetadata;
+  let data: Guarded<Readable>;
   let fetchTriples: jest.Mock<Promise<Readable>>;
   let fetchUpdate: jest.Mock<Promise<void>>;
   let triples: Quad[];
@@ -38,6 +40,9 @@ describe('A SparqlDataAccessor', (): void => {
 
   beforeEach(async(): Promise<void> => {
     metadata = new RepresentationMetadata();
+    data = guardedStreamFrom(
+      [ quad(namedNode('http://name'), namedNode('http://pred'), literal('value')) ],
+    );
     triples = [ quad(namedNode('this'), namedNode('a'), namedNode('triple')) ];
 
     // Makes it so the `SparqlEndpointFetcher` will always return the contents of the `triples` array
@@ -45,7 +50,7 @@ describe('A SparqlDataAccessor', (): void => {
       if (fetchError) {
         throw fetchError;
       }
-      return streamifyArray(triples);
+      return Readable.from(triples);
     });
     fetchUpdate = jest.fn(async(): Promise<void> => {
       if (updateError) {
@@ -62,7 +67,6 @@ describe('A SparqlDataAccessor', (): void => {
   });
 
   it('can only handle quad data.', async(): Promise<void> => {
-    const data = streamifyArray([]);
     await expect(accessor.canHandle({ binary: true, data, metadata })).rejects.toThrow(UnsupportedMediaTypeHttpError);
     metadata.contentType = 'newInternalType';
     await expect(accessor.canHandle({ binary: false, data, metadata })).rejects.toThrow(UnsupportedMediaTypeHttpError);
@@ -71,8 +75,8 @@ describe('A SparqlDataAccessor', (): void => {
   });
 
   it('returns the corresponding quads when data is requested.', async(): Promise<void> => {
-    const data = await accessor.getData({ path: 'http://identifier' });
+    const result = await accessor.getData({ path: 'http://identifier' });
-    await expect(arrayifyStream(data)).resolves.toBeRdfIsomorphic([
+    await expect(arrayifyStream(result)).resolves.toBeRdfIsomorphic([
       quad(namedNode('this'), namedNode('a'), namedNode('triple')),
     ]);
 
@@ -168,7 +172,6 @@ describe('A SparqlDataAccessor', (): void => {
   });
 
   it('overwrites the data and metadata when writing a resource and updates parent.', async(): Promise<void> => {
-    const data = streamifyArray([ quad(namedNode('http://name'), namedNode('http://pred'), literal('value')) ]);
     metadata = new RepresentationMetadata('http://test.com/container/resource',
       { [RDF.type]: [ toNamedNode(LDP.Resource) ]});
     await expect(accessor.writeDocument({ path: 'http://test.com/container/resource' }, data, metadata))
@@ -202,13 +205,12 @@ describe('A SparqlDataAccessor', (): void => {
   });
 
   it('errors when trying to write to a metadata document.', async(): Promise<void> => {
-    const data = streamifyArray([ quad(namedNode('http://name'), namedNode('http://pred'), literal('value')) ]);
     await expect(accessor.writeDocument({ path: 'meta:http://test.com/container/resource' }, data, metadata))
       .rejects.toThrow(new ConflictHttpError('Not allowed to create NamedNodes with the metadata extension.'));
   });
 
   it('errors when writing triples in a non-default graph.', async(): Promise<void> => {
-    const data = streamifyArray(
+    data = guardedStreamFrom(
       [ quad(namedNode('http://name'), namedNode('http://pred'), literal('value'), namedNode('badGraph!')) ],
     );
     await expect(accessor.writeDocument({ path: 'http://test.com/container/resource' }, data, metadata))

test/util/TestHelpers.ts

@@ -2,12 +2,12 @@ import { EventEmitter } from 'events';
 import { promises as fs } from 'fs';
 import type { IncomingHttpHeaders } from 'http';
 import { join } from 'path';
+import { Readable } from 'stream';
 import * as url from 'url';
 import type { MockResponse } from 'node-mocks-http';
 import { createResponse } from 'node-mocks-http';
-import streamifyArray from 'streamify-array';
 import type { ResourceStore } from '../../index';
-import { RepresentationMetadata } from '../../index';
+import { guardedStreamFrom, RepresentationMetadata } from '../../index';
 import type { PermissionSet } from '../../src/ldp/permissions/PermissionSet';
 import type { HttpHandler } from '../../src/server/HttpHandler';
 import type { HttpRequest } from '../../src/server/HttpRequest';
@@ -52,7 +52,7 @@ export class AclTestHelper {
 
     const representation = {
       binary: true,
-      data: streamifyArray(acl),
+      data: guardedStreamFrom(acl),
       metadata: new RepresentationMetadata({ [CONTENT_TYPE]: 'text/turtle' }),
     };
 
@@ -86,7 +86,7 @@ export class FileTestHelper {
     headers: IncomingHttpHeaders,
     data: Buffer,
   ): Promise<MockResponse<any>> {
-    const request = streamifyArray([ data ]) as HttpRequest;
+    const request = Readable.from([ data ]) as HttpRequest;
 
     request.url = requestUrl.pathname;
     request.method = method;