Skip to content

Add streaming JSON module - #351

Open
Mendrejk wants to merge 13 commits into
softwaremill:mainfrom
Mendrejk:feature/json-streaming
Open

Add streaming JSON module#351
Mendrejk wants to merge 13 commits into
softwaremill:mainfrom
Mendrejk:feature/json-streaming

Conversation

@Mendrejk

@Mendrejk Mendrejk commented Aug 10, 2026

Copy link
Copy Markdown

Closes #344

Summary

  • Add a new json Maven module for lazy, backpressured NDJSON and top-level JSON array parsing/rendering over Jox Flow/ByteFlow.
  • Support Jackson Class, TypeReference, ObjectReader, and ObjectWriter overloads, plus JsonReadSettings for bounded NDJSON records.
  • Document the API and cover framing, UTF-8/BOM handling, null policy, cancellation, failure propagation, and I/O integrations.

Notes

  • Array parsing currently uses ByteFlow.runToInputStream. Bulk-read latency and read-ahead are inherited from flows and tracked separately.

Mendrejk and others added 13 commits August 6, 2026 15:30
Add the Java 25 json Maven module and its Jackson 3.1.5 LTS dependency so the integration participates in the reactor build.
Provide lazy NDJSON and top-level JSON-array parsing and rendering over Jox flows with Jackson readers and writers.
Exercise NDJSON and array parsing/rendering across chunk boundaries, errors, cancellation, generic types, and I/O integrations.
Document the json dependency, supported wire formats, Jackson configuration, and composition with Jox flows and I/O.
Keep null guards consistent across Class/TypeReference and reader/writer overloads.

Co-authored-by: Cursor <cursoragent@cursor.com>
Express JSON delimiters through flow composition instead of mutable first-element state, without treating byte chunk boundaries as API behavior.

Co-authored-by: Cursor <cursoragent@cursor.com>
Separate serialization, record validation, and byte framing into explicit flow transformations.

Co-authored-by: Cursor <cursoragent@cursor.com>
Fail clearly when Jackson produces null instead of allowing later asynchronous stages to crash, and strengthen edge-case coverage for streaming behavior and failures.

Co-authored-by: Cursor <cursoragent@cursor.com>
Separate test setup, execution, and assertions consistently so each JSON behavior is easier to scan and reason about.

Co-authored-by: Cursor <cursoragent@cursor.com>
Keep Java requirements and the structured-concurrency comparison accurate without relying on a particular publication date.

Co-authored-by: Cursor <cursoragent@cursor.com>
Comment thread docs/flows.md
Finite & infinite streaming using flows, with reactive streams compatibility, (blocking) I/O integration, and a
high-level, "functional" API.

Requires Java 25 (current LTS).

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it's still the current LTS :)

Comment thread README.md
* Programmer-friendly structured concurrency (Java 25 only)
* Finite & infinite streaming using flows, with reactive streams compatibility, (blocking) I/O integration and a
high-level, “functional” API (Java 25 only)
* Streaming NDJSON and top-level JSON array integration using flows (Java 25 only)

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd change this bullet to mention flow integrations - Kafka, NDJSON

Comment thread docs/json.md
<dependency>
<groupId>com.softwaremill.jox</groupId>
<artifactId>json</artifactId>
<version>0.1.0</version>

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we'd need to put the current flows version here, so that the automatic version-bump works properly; this should be versioned together with other flow modules

@adamw

adamw commented Aug 25, 2026

Copy link
Copy Markdown
Member

Looks good! I left some (primarily Claude-found) comments

@adamw adamw left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 Automated Claude code review. Findings are posted as inline comments. No logic bugs found — framing, limit accounting, BOM/CRLF chunk-boundary handling, early termination and scope shutdown in parseArray were verified against the flows sources and tests.

/**
* Creates flows which parse or render newline-delimited JSON (NDJSON) and top-level JSON arrays.
*
* <p>All transformations are lazy and preserve the backpressure and cancellation behavior of the

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 Automated Claude review

This claim doesn't hold for parseArray: runToInputStream forks a producer feeding a channel buffered at 16 chunks, plus Jackson's own read-ahead, so it consumes bytes beyond downstream demand. Worth qualifying here and in docs/json.md.

static <T> Flow<T> parseNdjson(ByteFlow bytes, ObjectReader reader, JsonReadSettings settings) {
var singleValueReader = reader.with(DeserializationFeature.FAIL_ON_TRAILING_TOKENS);
return NdjsonFraming.lines(bytes, settings.maxNdjsonRecordBytes())
.filter(line -> !line.isBlank())

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 Automated Claude review

isBlank() matches more than NDJSON whitespace, so a corrupt line made of e.g. vertical tab or U+2028 is silently dropped instead of failing the flow. Filtering only lines of space/tab/CR would surface such corruption.


private NdjsonFraming() {}

static Flow<String> lines(ByteFlow bytes, int maxRecordBytes) {

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 Automated Claude review

This duplicates the line splitting in ByteFlow.linesUtf8()/LinesImpl. The new parts (size bound, BOM strip, strict UTF-8) could live in a shared bounded lines(charset, maxLineBytes) variant in flows — that would also fix the unbounded memory use linesUtf8() still has on untrusted input, and keep one framer to maintain.

private void append(byte[] bytes, int offset, int length) {
if ((long) buffer.size() + length > maxRecordBytes) {
throw new IllegalArgumentException(
"NDJSON record exceeds the configured maximum of "

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 Automated Claude review

The limit is also enforced on blank lines, which the docs say are ignored — an oversized whitespace-only line fails the whole flow. Enforcing it is the right call for memory safety, but docs/json.md and the javadoc should say the limit applies to blank lines too.

JsonToken token;
while ((token = parser.nextToken())
!= JsonToken.END_ARRAY) {
if (token == null) {

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 Automated Claude review

This branch looks dead for stream input: an InputStream-backed Jackson parser throws JsonEOFException from nextToken() at premature EOF instead of returning null (the truncated-input test asserts a JacksonException cause, not this message). If it's meant as a guard for user-supplied ObjectReaders, a comment would help.

Comment thread README.md
* Programmer-friendly structured concurrency (Java 25 only)
* Finite & infinite streaming using flows, with reactive streams compatibility, (blocking) I/O integration and a
high-level, “functional” API (Java 25 only)
* Streaming NDJSON and top-level JSON array integration using flows (Java 25 only)

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 Automated Claude review

This list still omits the kafka module, while docs/index.md (updated in this PR) says "five main modules" and lists it. Good moment to add the kafka bullet.

}

private Record completeRecord() {
var record = new Record(buffer.toByteArray(), firstRecord);

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 Automated Claude review

Each record's bytes are copied twice (append into the buffer, then toByteArray()). When the buffer is empty and the newline is inside the current array, the record could be decoded in place via ByteBuffer.wrap(array, start, len), buffering only records that span chunk boundaries.

var offset = record.first() && startsWithBom(bytes) ? UTF_8_BOM.length : 0;
var decoder =
StandardCharsets.UTF_8
.newDecoder()

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 Automated Claude review

A new configured decoder is allocated per record. The usingEmit body is single-threaded, so one decoder hoisted before runToEmit, with reset() per record, would do.

import com.softwaremill.jox.flows.FlowEmit;
import com.softwaremill.jox.flows.Flows;

final class NdjsonFraming {

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 Automated Claude review

A short class comment stating the framing contract would help: LF-only splitting with CR retained, one-time BOM strip, final unterminated record emitted, strict UTF-8 → IllegalArgumentException, limit excludes the LF. JsonParsing relies on these, but they're only documented on the public JsonFlow surface.

Comment thread docs/json.md
Lazy, backpressured parsing and rendering of newline-delimited JSON (NDJSON) and top-level JSON arrays using Jox
`Flow` and `ByteFlow`.

Requires Java 25.

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 Automated Claude review

Missing the Javadocs: https://javadoc.io/doc/com.softwaremill.jox/... link that channels.md, flows.md and structured.md have right after this line.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

New jox integration: streaming JSON

2 participants