Stream Support? about fast-json-stringify

[mcollina]

You can completely avoid the through stream, because it's jut adding overhead. You are not listening to any of the backpressure event. BTW, you should use through2, and read https://nodejs.org/en/docs/guides/backpressuring-in-streams/.

In your example, you should reuse the fastJson instance, otherwise it will be very very slow.

[mcollina]

@AutoSponge that would be handy, but it will change the logic heavily. Supporting streams means that we need to support backpressure, otherwise it is somewhat useless. In order to support backpressure, we need to generate the object in chunks, and to maintain a state of the object while we are doing that.

I think it's extremely complicated, and probably very expensive.

If you would like to lift some code from here in another OSS project, I'll be very happy to review that, but I feel it's too much for fast-json-stringify.

[AutoSponge]

This is what I came up with, not sure if it's naive:

const fastify = require('fastify')()
const through = require('through')
const fastJson = require('fast-json-stringify')

const schema = {
  out: {
    type: 'object',
    properties: {
      hello: { type: 'string' }
    }
  }
}

const everyoneInTheWorld = [
  'Olivia', 'Ethan', 'Isabella', 'Liam', 'Ava', 'Mason', 'Lily', 'Noah'
]

const jsonStream = (schema) => {
  let stream
  let first = true
  let anyData = false
  stream = through((data) => {
      let json
      anyData = true
      try {
        json = fastJson(schema)(data)
      } catch (err) {
        return stream.emit('error', err)
      }
      if (first) {
        first = false;
        stream.queue(`[\n${json}`)
      } else {
        stream.queue(`\n,\n${json}`)
      }
    },
    (data) => {
      if (!anyData)
      stream.queue('[\n')
      stream.queue('\n]\n')
      stream.queue(null)
    })
  return stream
}

// skip the output schema since it's a collection endpoint
fastify.get('/', function (req, reply) {
  reply.header('Content-Type', 'application/stream+json')
  const stream = jsonStream(schema.out)
  stream.on('data', (data) => reply.res.write(data))
  stream.on('end', () => reply.res.end())
  let i = 0
  // use a pump just to prove it's a stream in the console: curl http://localhost:4000/ -v
  let pump = setInterval(() => {
    const hello = everyoneInTheWorld[i++]
    if (i === everyoneInTheWorld.length) {
      clearInterval(pump)
      stream.end({ hello })
    } else {
      stream.write({ hello })
    }
  }, 100)
})

fastify.listen(4000)

[mcollina]

That won't work as you expect. It will generate all the chunks and add them for sending. You will be allocating the same amount of memory, and adding the stream overhead.

[AutoSponge]

I appreciate the quick reply, but I'm not sure I understand the issue. If it's the backpressure/flushing issues with .write, then adding the callback to .write should take that away (see below). But if it's something else, I'm not sure how to observe it.

fastify.get('/', function (req, reply) {
  let i = 0
  const pump = () => {
    const hello = everyoneInTheWorld[i++]
    if (i === everyoneInTheWorld.length) {
      stream.end({ hello })
    } else {
      stream.write({ hello })
    }
  }

  reply.header('Content-Type', 'application/stream+json')
  const stream = jsonStream(schema.out)
  stream.on('data', (data) => reply.res.write(data, pump))
  stream.on('end', () => reply.res.end())
  pump()
})

EDIT: I'm keeping the current implementation in a gist if anyone wants to follow along or provide feedback.