How to use the @comunica/bus-http.ActorHttp.toNodeReadable function in @comunica/bus-http

public async test(action: IActionRdfSourceIdentifier): Promise {
    const sourceUrl = this.getSourceUrl(action);
    const headers: Headers = new Headers();
    headers.append('Accept', this.acceptHeader);

    const httpAction: IActionHttp = { context: action.context, input: sourceUrl, init: { headers } };
    const httpResponse: IActorHttpOutput = await this.mediatorHttp.mediate(httpAction);
    if (httpResponse.ok) {
      const stream = ActorHttp.toNodeReadable(httpResponse.body);
      const body = (await require('stream-to-string')(stream));

      // Check if body contains all required things
      let valid = true;
      for (const line of this.toContain) {
        if (body.indexOf(line) < 0) {
          valid = false;
          break;
        }
      }
      if (valid) {
        return { priority: this.priority };
      }
    }

    // Avoid memory leaks

http.init.method = this.method;
    }
    if (this.headers) {
      const headers: Headers = new Headers();
      for (const value of this.headers) {
        const i: number = value.indexOf(':');
        headers.append(value.substr(0, i).toLowerCase(), value.substr(i + 2));
      }
      http.init.headers = headers;
    }

    const httpResponse: IActorHttpOutput = await this.mediatorHttp.mediate(http);
    const output: IActorOutputInit = {};
    // Wrap WhatWG readable stream into a Node.js readable stream
    // If the body already is a Node.js stream (in the case of node-fetch), don't do explicit conversion.
    const responseStream: NodeJS.ReadableStream = ActorHttp.toNodeReadable(httpResponse.body);
    if (httpResponse.status === 200) {
      output.stdout = responseStream.pipe(new PassThrough());
    } else {
      output.stderr = responseStream.pipe(new PassThrough());
    }
    return output;
  }