json-anystream

Takes any stream that provides JSON objects (array of objects or single object) or newline delimited JSON objects (NDJSON) and turns it into a JSON objects emitting stream.

This module is motivated by a need to read a set of JSON objects, which can be given in different forms. The module

• provide a stream that emits single JSON objects via the data event.
• supports JSON objects, JSON arrays of objects, NDJSON, as well as mutlipart/form-data with a file containing one of those.
• supports reading from local files (using fs). Content type is determined via file ending.
• supports reading from URLs (using http/https). Content type is determined via Content-Type header or file ending.
• provides an express middleware that wraps the above functionality and adds the stream as res.anystream.

Table of Contents

• Install
• Usage
  • Import
  • anystream.make
  • anystream.convert
  • anystream.StreamAnyObject
  • anystream.addStream
• Errors
• Maintainers
• Publish
• Contribute
• License

Install

1npm i json-anystream

json-anystream v2.0+ requires Node.js 18 and is ES module only. For older versions or CommonJS, use v1.1.1.

Usage

Import

1import * as anystream from "json-anystream"

anystream.make

async function make(input, type, adjust)

Takes an input (URL, file path, or existing stream) and makes a JSON object stream out of it. type can have one of the values "json", "ndjson", or "multipart", and is optional if the type can be inferred from the file ending or content type header. adjust is an optional adjustment method that is called for each object in the resulting stream before it is emitted.

Example:

1const stream = await anystream.make("https://example.com/test.json")
2stream.on("data", object => {
3  // object contains single JSON objects
4})
5stream.on("end", () => {
6  // stream is done
7})
8stream.on("error", error => {
9  // an error occurred, see below for error types
10})

You can also use for async:

1const stream = await anystream.make("https://example.com/api/test") // assuming content type header is set
2for await (let object of stream) {
3  // object contains single JSON objects
4}

For existing streams, anystream.make as simply a wrapper over anystream.convert. For URLs, http.get or https.get are used and the resulting stream is wrapped by anystream.convert. For files, fs.createReadStream is used and the resulting stream is wrapped by anystream.convert.

anystream.convert

async function convert(stream, type, adjust)

Takes an existing stream and returns a JSON object stream. The type parameter is required and is determined by the data of the stream:

• json: Any JSON stream (single object or array of object) that is compatible with stream-json.
• ndjson: Any stream of JSON objects compatible with ndjson.
• multipart: A request object containing multipart/form-data data that contains the field data with a compatible .json or .ndjson file (busboy is used for parsing the form data).

adjust is an optional adjustment method that is called for each object in the resulting stream before it is emitted.

anystream.StreamAnyObject

A custom streamer for stream-json that takes a stream of either a single JSON object or an array of JSON objects and transforms it into a stream that emits only JSON objects. I.e. if the input stream contains a single JSON object, that object is assembled and then emitted as a whole, if the input stream contains an array of JSON objects, the objects are emitted as direct JSON objects.

1import streamJson from "stream-json"
2const parser = streamJson.parser
3const pipeline = stream.pipe(parser()).pipe(new anystream.StreamAnyObject())
4pipeline.on("data", object => {
5  // ...
6})
7pipeline.on("end", () => {
8  // ...
9})

There is also a custom event in case the input was in fact a single JSON object:

1pipeline.on("isSingleObject", () => {
2  // if this event is called, the input stream contained a single object and NOT an array of objects
3})

The constructur takes an options objects that can optionally contain a adjust method that is called for each object in the resulting stream before it is emitted.

anystream.addStream

A middleware for express that provides the result of anystream.make as req.anystream. The input can be one of the following:

• A compatible data stream directly via POST data.
• A multipart/form-data request containing a .json or .ndjson file in the field data.
• A URL provided by the query parameter url.

An additional query parameter type, containing one of "multipart", "json", or "ndjson", may be provided if the type can't be inferred.

1import express from "express"
2const app = express()
3import * as anystream from "json-anystream"
4app.use((req, res, next) => {
5  if (req.method == "POST") {
6    // For POST requests, parse body with json-anystream middleware -> provides req.anystream property
7    anystream.addStream(req, res, next)
8  } else {
9    // For all other requests, parse as JSON -> provides req.body property
10    express.json()(req, res, next)
11  }
12})

anystream.addStream can also be called with an adjustment method first which is then forwarded into anystream.make (see above):

1anystream.addStream(object => {
2  // Adjust object
3  return object
4})(req, res, next)

Errors

json-anystream defines some custom errors:

• MissingTypeError - thrown when the type parameter is omitted and the type could not be inferred
• InvalidOrMissingDataFieldError - thrown for multipart/form-data when the required data field is missing or doesn't contain a .json or .ndjson file
• UrlRequestError - thrown when the input is a URL and there's an error during the request
• UnexpectedNonObjectValueError - thrown by the stream (i.e. for on("error")) when the stream did not contain a valid JSON object or array of objects; note that it is possible that this error is thrown after some values were already emitted in case that one of the later elements in an array is not an object

Maintainers

• @stefandesu

Publish

To publish a new version on npm after committing your changes, make sure you committed/merged all your changes to main successfully and then run:

1npm run release:patch
2# or for minor release:
3# npm run release:minor
4# or for major release:
5# npm run release:major

A new version will be published to npm automatically via GitHub Actions.

Contribute

PRs accepted. Please make sure the tests complete with your change. Add new tests if applicable.

Small note: If editing the README, please conform to the standard-readme specification.

License

MIT Copyright (c) 2022 Verbundzentrale des GBV (VZG)