| Server IP : 167.235.224.122 / Your IP : 216.73.216.110 Web Server : Apache/2.4.58 (Ubuntu) System : Linux newplayground 6.8.0-136-generic #136-Ubuntu SMP PREEMPT_DYNAMIC Wed Jul 1 21:33:11 UTC 2026 aarch64 User : deploy ( 1000) PHP Version : 8.4.23 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : OFF Directory : /var/www/html/jungly/node_modules/bfj/src/ |
Upload File : |
'use strict'
const check = require('check-types')
const eventify = require('./eventify')
const events = require('./events')
const JsonStream = require('./jsonstream')
const Hoopy = require('hoopy')
const promise = require('./promise')
const tryer = require('tryer')
const DEFAULT_BUFFER_LENGTH = 1024
module.exports = streamify
/**
* Public function `streamify`.
*
* Asynchronously serialises a data structure to a stream of JSON
* data. Sanely handles promises, buffers, maps and other iterables.
*
* @param data: The data to transform.
*
* @option space: Indentation string, or the number of spaces
* to indent each nested level by.
*
* @option promises: 'resolve' or 'ignore', default is 'resolve'.
*
* @option buffers: 'toString' or 'ignore', default is 'toString'.
*
* @option maps: 'object' or 'ignore', default is 'object'.
*
* @option iterables: 'array' or 'ignore', default is 'array'.
*
* @option circular: 'error' or 'ignore', default is 'error'.
*
* @option yieldRate: The number of data items to process per timeslice,
* default is 16384.
*
* @option bufferLength: The length of the buffer, default is 1024.
*
* @option highWaterMark: If set, will be passed to the readable stream constructor
* as the value for the highWaterMark option.
*
* @option Promise: The promise constructor to use, defaults to bluebird.
**/
function streamify (data, options = {}) {
const emitter = eventify(data, options)
const json = new Hoopy(options.bufferLength || DEFAULT_BUFFER_LENGTH)
const Promise = promise(options)
const space = normaliseSpace(options)
let streamOptions
const { highWaterMark } = options
if (highWaterMark) {
streamOptions = { highWaterMark }
}
const stream = new JsonStream(read, streamOptions)
let awaitPush = true
let index = 0
let indentation = ''
let isEnded
let isPaused = false
let isProperty
let length = 0
let mutex = Promise.resolve()
let needsComma
emitter.on(events.array, noRacing(array))
emitter.on(events.object, noRacing(object))
emitter.on(events.property, noRacing(property))
emitter.on(events.string, noRacing(string))
emitter.on(events.number, noRacing(value))
emitter.on(events.literal, noRacing(value))
emitter.on(events.endArray, noRacing(endArray))
emitter.on(events.endObject, noRacing(endObject))
emitter.on(events.end, noRacing(end))
emitter.on(events.error, noRacing(error))
emitter.on(events.dataError, noRacing(dataError))
return stream
function read () {
if (awaitPush) {
awaitPush = false
if (isEnded) {
if (length > 0) {
after()
}
return endStream()
}
}
if (isPaused) {
after()
}
}
function after () {
if (awaitPush) {
return
}
let i
for (i = 0; i < length && ! awaitPush; ++i) {
if (! stream.push(json[i + index], 'utf8')) {
awaitPush = true
}
}
if (i === length) {
index = length = 0
} else {
length -= i
index += i
}
}
function endStream () {
if (! awaitPush) {
stream.push(null)
}
}
function noRacing (handler) {
return eventData => mutex = mutex.then(() => handler(eventData))
}
function array () {
return beforeScope()
.then(() => addJson('['))
.then(() => afterScope())
}
function beforeScope () {
return before(true)
}
function before (isScope) {
if (isProperty) {
isProperty = false
if (space) {
return addJson(' ')
}
return Promise.resolve()
}
return Promise.resolve()
.then(() => {
if (needsComma) {
if (isScope) {
needsComma = false
}
return addJson(',')
}
if (! isScope) {
needsComma = true
}
})
.then(() => {
if (space && indentation) {
return indent()
}
})
}
function addJson (chunk) {
if (length + 1 <= json.length) {
json[index + length++] = chunk
after()
return Promise.resolve()
}
isPaused = true
return new Promise(resolve => {
const unpause = emitter.pause()
tryer({
interval: -10,
until () {
return length + 1 <= json.length
},
pass () {
isPaused = false
json[index + length++] = chunk
resolve()
setImmediate(unpause)
}
})
})
}
function indent () {
return addJson(`\n${indentation}`)
}
function afterScope () {
needsComma = false
if (space) {
indentation += space
}
}
function object () {
return beforeScope()
.then(() => addJson('{'))
.then(() => afterScope())
}
function property (name) {
return before()
.then(() => addJson(`"${name}":`))
.then(() => {
isProperty = true
})
}
function string (s) {
return value(`"${s}"`)
}
function value (v) {
return before()
.then(() => addJson(`${v}`))
}
function endArray () {
return beforeScopeEnd()
.then(() => addJson(']'))
.then(() => afterScopeEnd())
}
function beforeScopeEnd () {
if (space) {
indentation = indentation.substr(space.length)
return indent()
}
return Promise.resolve()
}
function afterScopeEnd () {
needsComma = true
}
function endObject () {
return beforeScopeEnd()
.then(() => addJson('}'))
.then(() => afterScopeEnd())
}
function end () {
after()
isEnded = true
endStream()
}
function error (err) {
stream.emit('error', err)
}
function dataError (err) {
stream.emit('dataError', err)
}
}
function normaliseSpace (options) {
if (check.positive(options.space)) {
return new Array(options.space + 1).join(' ')
}
if (check.nonEmptyString(options.space)) {
return options.space
}
}