Until and while may cause an error in the unpiped stream.

[MichalCz]

Consider this more a use case than a problem (as it is outside of scramjet).

Let's consider code:

StringStream.from(request(someUrl))
    .lines()
    .parse(JSON.parse)
    .until(x => x.date < Date.now())

What we do is unpipe the response, but that is not closed by request so how do we handle this?

Mention: openaq-fetch#547

[MichalCz]

The current solution has been deployed to OpenAQ adapters.

  const requestObject = request.get(stationUrl);
  return DataStream	  return DataStream
    .from(() => request.get(stationUrl).pipe(JSONStream('*')))
    .pipeline(
      requestObject,
      JSONStream('*')
    )
    .catch(e => {
      requestObject.abort();
      e.stream.end();
      throw e;
    })

Should it and can it be any cleaner - the abort operation is specific to request module...

[MichalCz]

After giving this some thought it seems that the current solution is good enough.