Add streaming JSON module by Mendrejk · Pull Request #351 · softwaremill/jox

Mendrejk · 2026-08-10T17:33:59Z

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.

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>

adamw · 2026-08-25T14:07:52Z

 Finite & infinite streaming using flows, with reactive streams compatibility, (blocking) I/O integration, and a
 high-level, "functional" API.

-Requires Java 25 (current LTS).


it's still the current LTS :)

adamw · 2026-08-25T14:10:11Z

 * 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)


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

adamw · 2026-08-25T14:11:54Z

+<dependency>
+    <groupId>com.softwaremill.jox</groupId>
+    <artifactId>json</artifactId>
+    <version>0.1.0</version>


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 · 2026-08-25T14:19:55Z

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

adamw

🤖 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.

adamw · 2026-08-25T14:21:18Z

+/**
+ * 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


🤖 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.

adamw · 2026-08-25T14:21:18Z

+    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())


🤖 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.

adamw · 2026-08-25T14:21:18Z

+
+    private NdjsonFraming() {}
+
+    static Flow<String> lines(ByteFlow bytes, int maxRecordBytes) {


🤖 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.

adamw · 2026-08-25T14:21:18Z

+        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 "


🤖 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.

adamw · 2026-08-25T14:21:18Z

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


🤖 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.

adamw · 2026-08-25T14:21:18Z

 * 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)


🤖 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.

adamw · 2026-08-25T14:21:18Z

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


🤖 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.

adamw · 2026-08-25T14:21:18Z

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


🤖 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.

adamw · 2026-08-25T14:21:18Z

+import com.softwaremill.jox.flows.FlowEmit;
+import com.softwaremill.jox.flows.Flows;
+
+final class NdjsonFraming {


🤖 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.

adamw · 2026-08-25T14:21:19Z

+Lazy, backpressured parsing and rendering of newline-delimited JSON (NDJSON) and top-level JSON arrays using Jox
+`Flow` and `ByteFlow`.
+
+Requires Java 25.


🤖 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.

Mendrejk and others added 13 commits August 6, 2026 15:30

build(json): register Jackson streaming module

7bb7d70

Add the Java 25 json Maven module and its Jackson 3.1.5 LTS dependency so the integration participates in the reactor build.

feat(json): add streaming JSON flow operations

0ea220c

Provide lazy NDJSON and top-level JSON-array parsing and rendering over Jox flows with Jackson readers and writers.

test(json): cover streaming JSON flows

6cc84f3

Exercise NDJSON and array parsing/rendering across chunk boundaries, errors, cancellation, generic types, and I/O integrations.

docs(json): document streaming JSON integration

1264baf

Document the json dependency, supported wire formats, Jackson configuration, and composition with Jox flows and I/O.

style(json): inline requireNonNull checks in JsonFlow

6641006

Keep null guards consistent across Class/TypeReference and reader/writer overloads. Co-authored-by: Cursor <cursoragent@cursor.com>

refactor(json): render arrays using intersperse

e3cee22

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>

refactor(json): compose NDJSON rendering stages

b2e3f0a

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

fix(json): reject null flow elements

b87111b

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>

test(json): structure scenarios as BDD

318cff0

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>

docs: remove time-sensitive version wording

81f9e4d

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

fix(json): harden streaming parsing and rendering

c4de2cb

docs(json): trim repetitive API documentation

783b7dc

refactor(json): simplify NDJSON framing

f34fb9a

adamw reviewed Aug 25, 2026

View reviewed changes


		private NdjsonFraming() {}

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

Uh oh!

Conversation

Mendrejk commented Aug 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Notes

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

adamw commented Aug 25, 2026

Uh oh!

adamw left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Mendrejk commented Aug 10, 2026 •

edited

Loading