Json Stream Laravel Package

Getting Started

Install the package with Composer:

composer require clue/json-stream

The public entry point is Clue\JsonStream\StreamingJsonParser. A minimal smoke test looks like this:

use Clue\JsonStream\StreamingJsonParser;

$parser = new StreamingJsonParser();

// Concatenated JSON documents
foreach ($parser->parse('{"id":1}{"id":2}') as $record) {
    echo $record->id, PHP_EOL;
}

The same parser handles newline-delimited JSON:

foreach ($parser->parse("{\"id\":1}\n{\"id\":2}\n") as $record) {
    // ...
}

Where to look first:

1. The README for the canonical parse() examples.
2. src/StreamingJsonParser.php — the implementation is intentionally small, so read it rather than guessing at buffering semantics.
3. The test suite for chunk-boundary and malformed-input expectations.

Your first practical use case should be NDJSON ingestion: processing a large .ndjson export, log file, or chunked HTTP response without loading the entire payload into memory. Feed the parser fixed-size chunks and handle each yielded record as it becomes complete.

Implementation Patterns

Chunked file processing. Read in bounded chunks and fully drain the generator for each chunk:

$parser = new StreamingJsonParser();
$fh = fopen('events.ndjson', 'rb');

while (!feof($fh)) {
    $chunk = fread($fh, 8192);
    if ($chunk === false || $chunk === '') {
        break;
    }

    foreach ($parser->parse($chunk) as $event) {
        processEvent($event);
    }
}

fclose($fh);

HTTP/PSR-7 streams. The parser works well with any producer that emits chunks:

$parser = new StreamingJsonParser();
$body = $response->getBody();

while (!$body->eof()) {
    foreach ($parser->parse($body->read(8192)) as $record) {
        // handle one complete JSON document
    }
}

ReactPHP integration. Because this is from the ReactPHP ecosystem, the natural pattern is to feed incoming chunks from an event listener:

$parser = new StreamingJsonParser();

$response->on('data', function (string $chunk) use ($parser) {
    foreach ($parser->parse($chunk) as $event) {
        $this->handle($event);
    }
});

Keep handlers fast; parsing is synchronous and heavy per-record work will block the event loop. Queue or offload expensive processing.

Batching for databases/APIs. Since records are yielded incrementally, batch around the generator:

$batch = [];

foreach ($parser->parse($chunk) as $row) {
    $batch[] = $row;

    if (count($batch) >= 500) {
        insertBatch($batch);
        $batch = [];
    }
}

if ($batch !== []) {
    insertBatch($batch);
}

Error isolation. Wrap parsing in a try/catch. A malformed document typically invalidates the parser’s buffered state, so treat the parser instance as poisoned and create a new one:

try {
    foreach ($parser->parse($chunk) as $record) {
        // ...
    }
} catch (Throwable $e) {
    logError($e, $chunk);
    $parser = new StreamingJsonParser();
}

Be aware that recreating the parser discards any partial document it was holding; you cannot reliably resynchronize mid-stream unless your transport has its own record boundaries.

Normalize decoded values at the boundary. The parser delegates leaf decoding to json_decode(). Confirm whether your installed version produces stdClass objects or associative arrays, then normalize once rather than scattering casts through your domain code:

foreach ($parser->parse($chunk) as $value) {
    $row = json_decode(json_encode($value), true);
    // ...
}

Test chunk splitting explicitly. The most valuable test is a fuzz-style boundary test: split a known fixture at every byte offset and assert the same records come out.

$json = '{"a":1}{"b":2}';

for ($i = 1; $i < strlen($json); $i++) {
    $parser = new StreamingJsonParser();
    $out = [];

    foreach ([substr($json, 0, $i), substr($json, $i)] as $chunk) {
        foreach ($parser->parse($chunk) as $value) {
            $out[] = $value;
        }
    }

    assert(count($out) === 2);
}

Gotchas and Tips

• This is document-boundary streaming, not token-level streaming. It buffers until a complete top-level JSON value is available. If you need to stream through one gigantic JSON array or object without loading it, use a tokenizer such as halaxa/json-machine instead. This package helps with many documents, not one enormous document.

• Top-level scalars are ambiguous in concatenated JSON. 123456 could be one number or two. truefalse cannot be split reliably. Use newline-delimited JSON when documents may be scalars, and always include a delimiter between top-level values.

• Unclosed documents can buffer indefinitely. A truncated object, a missing quote, or a malicious stream with an open brace can grow the internal buffer without yielding. If you consume untrusted input, enforce a maximum bytes-since-last-yield limit in your stream wrapper and abort accordingly.

• Malformed JSON cannot usually be resynchronized. Once json_decode() fails inside the parser, you generally cannot know where the next valid document starts. Log the offending chunk plus a bounded prefix of the buffer, drop the connection/file, and restart from a known checkpoint.

• Trailing partial documents may be silent. If the stream ends mid-document and your version exposes no explicit finish/flush step, the parser simply holds the incomplete buffer and yields nothing. Track “bytes pending” or validate at end-of-stream so truncated input is not mistaken for clean EOF.

• json_decode() defaults still apply. Watch for depth limits on deeply nested documents, invalid UTF-8, and large integers turning into floats. If your version exposes no decode options, pre-normalize or post-process those fields. Strip a UTF-8 BOM from the first chunk before feeding the parser.

• Blank lines and CRLF. NDJSON with trailing \r, empty lines, or pretty-printed multi-line objects can behave differently than expected. Test your exact producer format; normalize line endings and decide explicitly whether blank lines should be skipped.

• Always fully drain the generator. If parse() returns a generator and you break early, buffered state may not advance the way you expect. Either consume every yielded value or discard the parser instance.

• One parser instance per stream. Do not share a parser between concurrent connections, files, or event-loop callbacks. It is stateful by design.

• Chunk size is a throughput/memory trade-off. 8–64 KB chunks are usually sensible. Tiny chunks increase per-call overhead; huge chunks increase peak memory without improving single-document memory usage.

• Package age and ecosystem. Last release was December 2020, so check composer why-not clue/json-stream against your PHP version. The codebase is tiny and MIT-licensed, so vendoring or forking is realistic if you need PHP 8.x compatibility fixes or decode-option support. Prefer wrapping or subclassing over patching vendor/.

• Extension points are thin by design. Add behavior with generator wrappers around parse() — logging, mapping, filtering, metrics — rather than trying to inject into the parser’s internals. This keeps the state machine untouched and makes chunk-boundary behavior easier to test.