From 7bb7d70ce05c04764cfe77bd2e164a2273ec30f6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sebastian=20=C5=81akomy?= Date: Thu, 6 Aug 2026 15:30:35 +0200 Subject: [PATCH 01/13] build(json): register Jackson streaming module Add the Java 25 json Maven module and its Jackson 3.1.5 LTS dependency so the integration participates in the reactor build. --- json/pom.xml | 82 ++++++++++++++++++++++++++++++++++++++++++++++++++++ pom.xml | 1 + 2 files changed, 83 insertions(+) create mode 100644 json/pom.xml diff --git a/json/pom.xml b/json/pom.xml new file mode 100644 index 0000000..3d538cf --- /dev/null +++ b/json/pom.xml @@ -0,0 +1,82 @@ + + + 4.0.0 + + + com.softwaremill.jox + parent + 1.1.2 + + + json + 0.1.0 + jar + ${project.groupId}:${project.artifactId} + + + 25 + 3.1.5 + + + + + com.softwaremill.jox + structured + 0.5.3 + + + com.softwaremill.jox + flows + 0.5.3 + + + tools.jackson.core + jackson-databind + ${jackson.version} + + + + + org.junit.jupiter + junit-jupiter + 6.1.2 + test + + + + + + + com.diffplug.spotless + spotless-maven-plugin + + + + + + org.apache.maven.plugins + maven-compiler-plugin + + true + + + + org.apache.maven.plugins + maven-surefire-plugin + + --enable-preview + + + + org.apache.maven.plugins + maven-javadoc-plugin + + 25 + --enable-preview + + + + + + diff --git a/pom.xml b/pom.xml index 8414103..49742ce 100644 --- a/pom.xml +++ b/pom.xml @@ -22,6 +22,7 @@ flows channels-fray-tests kafka + json From 0ea220c6fd48dc1cf9f13aca06990e1794eabcf7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sebastian=20=C5=81akomy?= Date: Thu, 6 Aug 2026 15:30:47 +0200 Subject: [PATCH 02/13] feat(json): add streaming JSON flow operations Provide lazy NDJSON and top-level JSON-array parsing and rendering over Jox flows with Jackson readers and writers. --- .../com/softwaremill/jox/json/JsonFlow.java | 199 ++++++++++++++++++ .../softwaremill/jox/json/JsonParsing.java | 65 ++++++ .../softwaremill/jox/json/JsonRendering.java | 58 +++++ 3 files changed, 322 insertions(+) create mode 100644 json/src/main/java/com/softwaremill/jox/json/JsonFlow.java create mode 100644 json/src/main/java/com/softwaremill/jox/json/JsonParsing.java create mode 100644 json/src/main/java/com/softwaremill/jox/json/JsonRendering.java diff --git a/json/src/main/java/com/softwaremill/jox/json/JsonFlow.java b/json/src/main/java/com/softwaremill/jox/json/JsonFlow.java new file mode 100644 index 0000000..ac5c1cb --- /dev/null +++ b/json/src/main/java/com/softwaremill/jox/json/JsonFlow.java @@ -0,0 +1,199 @@ +package com.softwaremill.jox.json; + +import java.util.Objects; + +import com.softwaremill.jox.flows.Flow; +import com.softwaremill.jox.flows.Flow.ByteFlow; + +import tools.jackson.core.type.TypeReference; +import tools.jackson.databind.ObjectMapper; +import tools.jackson.databind.ObjectReader; +import tools.jackson.databind.ObjectWriter; + +/** + * Creates flows which parse or render newline-delimited JSON (NDJSON) and top-level JSON arrays. + * + *

All transformations are lazy and preserve the backpressure and cancellation behavior of the + * supplied flow. Values are parsed or rendered one at a time. + */ +public final class JsonFlow { + + private static final ObjectMapper DEFAULT_MAPPER = new ObjectMapper(); + + private JsonFlow() {} + + /** + * Parses newline-delimited JSON using a default {@link ObjectMapper}. Empty and whitespace-only + * lines are ignored. Both LF and CRLF line endings are accepted, as is a final record without a + * line ending. + * + * @param bytes the UTF-8 encoded NDJSON + * @param valueType the type of each parsed value + * @param the type of parsed values + * @return a flow emitting one value for each non-blank input line + */ + public static Flow parseNdjson(ByteFlow bytes, Class valueType) { + Objects.requireNonNull(valueType, "valueType"); + return parseNdjson(bytes, DEFAULT_MAPPER.readerFor(valueType)); + } + + /** + * Parses newline-delimited JSON using a default {@link ObjectMapper}. Empty and whitespace-only + * lines are ignored. Both LF and CRLF line endings are accepted, as is a final record without a + * line ending. + * + * @param bytes the UTF-8 encoded NDJSON + * @param valueType the generic type of each parsed value + * @param the type of parsed values + * @return a flow emitting one value for each non-blank input line + */ + public static Flow parseNdjson(ByteFlow bytes, TypeReference valueType) { + Objects.requireNonNull(valueType, "valueType"); + return parseNdjson(bytes, DEFAULT_MAPPER.readerFor(valueType)); + } + + /** + * Parses newline-delimited JSON using the supplied Jackson reader. Empty and whitespace-only + * lines are ignored. Both LF and CRLF line endings are accepted, as is a final record without a + * line ending. Each non-blank line must contain exactly one JSON value. + * + * @param bytes the UTF-8 encoded NDJSON + * @param reader the reader used to deserialize each value + * @param the type of parsed values + * @return a flow emitting one value for each non-blank input line + */ + public static Flow parseNdjson(ByteFlow bytes, ObjectReader reader) { + return JsonParsing.parseNdjson( + Objects.requireNonNull(bytes, "bytes"), Objects.requireNonNull(reader, "reader")); + } + + /** + * Parses a top-level JSON array using a default {@link ObjectMapper}. The returned flow emits + * each array element as soon as it is available. Input that is not one complete top-level + * array, including input containing trailing JSON, fails the flow. + * + * @param bytes the UTF-8 encoded JSON array + * @param valueType the type of each array element + * @param the type of parsed values + * @return a flow emitting the array elements + */ + public static Flow parseArray(ByteFlow bytes, Class valueType) { + Objects.requireNonNull(valueType, "valueType"); + return parseArray(bytes, DEFAULT_MAPPER.readerFor(valueType)); + } + + /** + * Parses a top-level JSON array using a default {@link ObjectMapper}. The returned flow emits + * each array element as soon as it is available. Input that is not one complete top-level + * array, including input containing trailing JSON, fails the flow. + * + * @param bytes the UTF-8 encoded JSON array + * @param valueType the generic type of each array element + * @param the type of parsed values + * @return a flow emitting the array elements + */ + public static Flow parseArray(ByteFlow bytes, TypeReference valueType) { + Objects.requireNonNull(valueType, "valueType"); + return parseArray(bytes, DEFAULT_MAPPER.readerFor(valueType)); + } + + /** + * Parses a top-level JSON array using the supplied Jackson reader. The returned flow emits each + * array element as soon as it is available. Input that is not one complete top-level array, + * including input containing trailing JSON, fails the flow. + * + * @param bytes the UTF-8 encoded JSON array + * @param reader the reader used to deserialize each array element + * @param the type of parsed values + * @return a flow emitting the array elements + */ + public static Flow parseArray(ByteFlow bytes, ObjectReader reader) { + return JsonParsing.parseArray( + Objects.requireNonNull(bytes, "bytes"), Objects.requireNonNull(reader, "reader")); + } + + /** + * Renders values as newline-delimited JSON using a default {@link ObjectMapper}. Every value is + * followed by an LF byte, including the final value. + * + * @param values the values to render + * @param valueType the type of each value + * @param the type of rendered values + * @return a flow emitting UTF-8 encoded NDJSON + */ + public static ByteFlow renderNdjson(Flow values, Class valueType) { + Objects.requireNonNull(valueType, "valueType"); + return renderNdjson(values, DEFAULT_MAPPER.writerFor(valueType)); + } + + /** + * Renders values as newline-delimited JSON using a default {@link ObjectMapper}. Every value is + * followed by an LF byte, including the final value. + * + * @param values the values to render + * @param valueType the generic type of each value + * @param the type of rendered values + * @return a flow emitting UTF-8 encoded NDJSON + */ + public static ByteFlow renderNdjson(Flow values, TypeReference valueType) { + Objects.requireNonNull(valueType, "valueType"); + return renderNdjson(values, DEFAULT_MAPPER.writerFor(valueType)); + } + + /** + * Renders values as newline-delimited JSON using the supplied Jackson writer. Every value is + * followed by an LF byte, including the final value. Writer output containing a raw CR or LF + * byte is rejected, as it would produce invalid NDJSON records. + * + * @param values the values to render + * @param writer the writer used to serialize each value + * @param the type of rendered values + * @return a flow emitting UTF-8 encoded NDJSON + */ + public static ByteFlow renderNdjson(Flow values, ObjectWriter writer) { + return JsonRendering.renderNdjson( + Objects.requireNonNull(values, "values"), Objects.requireNonNull(writer, "writer")); + } + + /** + * Renders values as one JSON array using a default {@link ObjectMapper}. Elements are + * serialized one at a time. An empty input flow produces {@code []}. + * + * @param values the values to render + * @param valueType the type of each array element + * @param the type of rendered values + * @return a flow emitting one UTF-8 encoded JSON array + */ + public static ByteFlow renderArray(Flow values, Class valueType) { + Objects.requireNonNull(valueType, "valueType"); + return renderArray(values, DEFAULT_MAPPER.writerFor(valueType)); + } + + /** + * Renders values as one JSON array using a default {@link ObjectMapper}. Elements are + * serialized one at a time. An empty input flow produces {@code []}. + * + * @param values the values to render + * @param valueType the generic type of each array element + * @param the type of rendered values + * @return a flow emitting one UTF-8 encoded JSON array + */ + public static ByteFlow renderArray(Flow values, TypeReference valueType) { + Objects.requireNonNull(valueType, "valueType"); + return renderArray(values, DEFAULT_MAPPER.writerFor(valueType)); + } + + /** + * Renders values as one JSON array using the supplied Jackson writer. Elements are serialized + * one at a time. An empty input flow produces {@code []}. + * + * @param values the values to render + * @param writer the writer used to serialize each array element + * @param the type of rendered values + * @return a flow emitting one UTF-8 encoded JSON array + */ + public static ByteFlow renderArray(Flow values, ObjectWriter writer) { + return JsonRendering.renderArray( + Objects.requireNonNull(values, "values"), Objects.requireNonNull(writer, "writer")); + } +} diff --git a/json/src/main/java/com/softwaremill/jox/json/JsonParsing.java b/json/src/main/java/com/softwaremill/jox/json/JsonParsing.java new file mode 100644 index 0000000..466d9a9 --- /dev/null +++ b/json/src/main/java/com/softwaremill/jox/json/JsonParsing.java @@ -0,0 +1,65 @@ +package com.softwaremill.jox.json; + +import static com.softwaremill.jox.structured.Scopes.supervised; + +import com.softwaremill.jox.flows.Flow; +import com.softwaremill.jox.flows.Flow.ByteFlow; +import com.softwaremill.jox.flows.Flows; + +import tools.jackson.core.JsonParser; +import tools.jackson.core.JsonToken; +import tools.jackson.databind.DeserializationFeature; +import tools.jackson.databind.ObjectReader; + +final class JsonParsing { + + private JsonParsing() {} + + static Flow parseNdjson(ByteFlow bytes, ObjectReader reader) { + var singleValueReader = reader.with(DeserializationFeature.FAIL_ON_TRAILING_TOKENS); + return bytes.linesUtf8() + .filter(line -> !line.isBlank()) + .map(line -> singleValueReader.readValue(line)); + } + + static Flow parseArray(ByteFlow bytes, ObjectReader reader) { + var elementReader = reader.without(DeserializationFeature.FAIL_ON_TRAILING_TOKENS); + return Flows.usingEmit( + emit -> + supervised( + scope -> { + try (var inputStream = bytes.runToInputStream(scope); + JsonParser parser = + elementReader.createParser(inputStream)) { + requireToken( + parser.nextToken(), + JsonToken.START_ARRAY, + "Expected one top-level JSON array"); + + JsonToken token; + while ((token = parser.nextToken()) + != JsonToken.END_ARRAY) { + if (token == null) { + throw new IllegalArgumentException( + "Unexpected end of input while parsing the" + + " top-level JSON array"); + } + emit.apply(elementReader.readValue(parser)); + } + + if (parser.nextToken() != null) { + throw new IllegalArgumentException( + "Unexpected content after the top-level JSON" + + " array"); + } + } + return null; + })); + } + + private static void requireToken(JsonToken actual, JsonToken expected, String message) { + if (actual != expected) { + throw new IllegalArgumentException(message); + } + } +} diff --git a/json/src/main/java/com/softwaremill/jox/json/JsonRendering.java b/json/src/main/java/com/softwaremill/jox/json/JsonRendering.java new file mode 100644 index 0000000..c80f3a3 --- /dev/null +++ b/json/src/main/java/com/softwaremill/jox/json/JsonRendering.java @@ -0,0 +1,58 @@ +package com.softwaremill.jox.json; + +import com.softwaremill.jox.flows.ByteChunk; +import com.softwaremill.jox.flows.Flow; +import com.softwaremill.jox.flows.Flow.ByteFlow; +import com.softwaremill.jox.flows.Flows; + +import tools.jackson.databind.ObjectWriter; + +final class JsonRendering { + + private static final ByteChunk ARRAY_START = ByteChunk.fromArray(new byte[] {'['}); + private static final ByteChunk ARRAY_END = ByteChunk.fromArray(new byte[] {']'}); + private static final ByteChunk COMMA = ByteChunk.fromArray(new byte[] {','}); + private static final ByteChunk NEW_LINE = ByteChunk.fromArray(new byte[] {'\n'}); + + private JsonRendering() {} + + static ByteFlow renderNdjson(Flow values, ObjectWriter writer) { + return Flows.usingEmit( + emit -> + values.runForeach( + value -> { + byte[] json = writer.writeValueAsBytes(value); + rejectLineBreaks(json); + emit.apply(ByteChunk.fromArray(json).concat(NEW_LINE)); + })) + .toByteFlow(); + } + + static ByteFlow renderArray(Flow values, ObjectWriter writer) { + return Flows.usingEmit( + emit -> { + emit.apply(ARRAY_START); + boolean[] first = {true}; + values.runForeach( + value -> { + ByteChunk json = + ByteChunk.fromArray( + writer.writeValueAsBytes(value)); + emit.apply(first[0] ? json : COMMA.concat(json)); + first[0] = false; + }); + emit.apply(ARRAY_END); + }) + .toByteFlow(); + } + + private static void rejectLineBreaks(byte[] json) { + for (byte value : json) { + if (value == '\r' || value == '\n') { + throw new IllegalArgumentException( + "ObjectWriter output contains a raw line break and cannot be rendered as" + + " NDJSON"); + } + } + } +} From 6cc84f30b5e0c6e9dd053da3e6828c1467fc2483 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sebastian=20=C5=81akomy?= Date: Thu, 6 Aug 2026 15:30:57 +0200 Subject: [PATCH 03/13] test(json): cover streaming JSON flows Exercise NDJSON and array parsing/rendering across chunk boundaries, errors, cancellation, generic types, and I/O integrations. --- .../softwaremill/jox/json/JsonFlowTest.java | 536 ++++++++++++++++++ 1 file changed, 536 insertions(+) create mode 100644 json/src/test/java/com/softwaremill/jox/json/JsonFlowTest.java diff --git a/json/src/test/java/com/softwaremill/jox/json/JsonFlowTest.java b/json/src/test/java/com/softwaremill/jox/json/JsonFlowTest.java new file mode 100644 index 0000000..a48e6f2 --- /dev/null +++ b/json/src/test/java/com/softwaremill/jox/json/JsonFlowTest.java @@ -0,0 +1,536 @@ +package com.softwaremill.jox.json; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import com.softwaremill.jox.flows.ByteChunk; +import com.softwaremill.jox.flows.Flow; +import com.softwaremill.jox.flows.Flows; + +import tools.jackson.core.type.TypeReference; +import tools.jackson.databind.JsonNode; +import tools.jackson.databind.ObjectMapper; + +class JsonFlowTest { + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + @TempDir Path tempDir; + + @Test + void shouldParseNdjsonLineEndingsBlankLinesAndFinalUnterminatedRecord() throws Exception { + // given + var input = + byteFlow( + """ + + {"name":"Ada","age":36}\r + \t + {"name":"Łukasz","age":41}\ + """); + + // when + var result = JsonFlow.parseNdjson(input, Person.class).runToList(); + + // then + assertEquals(List.of(new Person("Ada", 36), new Person("Łukasz", 41)), result); + } + + @Test + void shouldParseEmptyNdjsonAndEmptyArray() throws Exception { + assertEquals(List.of(), JsonFlow.parseNdjson(byteFlow(""), Person.class).runToList()); + assertEquals(List.of(), JsonFlow.parseArray(byteFlow("[]"), Person.class).runToList()); + } + + @Test + void shouldParseNdjsonAcrossEveryByteBoundaryIncludingUtf8() throws Exception { + // given + var input = oneByteChunks("{\"name\":\"Zażółć 🦊\",\"age\":7}\n"); + + // when & then + assertEquals( + List.of(new Person("Zażółć 🦊", 7)), + JsonFlow.parseNdjson(input, Person.class).runToList()); + } + + @Test + void shouldParseArrayAcrossEveryByteBoundaryIncludingUtf8() throws Exception { + // given + var input = + oneByteChunks( + """ + [{"name":"東京","age":10},{"name":"Málaga 🌊","age":20}] + """); + + // when & then + assertEquals( + List.of(new Person("東京", 10), new Person("Málaga 🌊", 20)), + JsonFlow.parseArray(input, Person.class).runToList()); + } + + @Test + void shouldParseGenericTypesUsingTypeReferenceOverloads() throws Exception { + // given + TypeReference> type = new TypeReference<>() {}; + + // when + var ndjson = + JsonFlow.parseNdjson( + byteFlow( + """ + [{"name":"Ada","age":36}] + [{"name":"Grace","age":37},{"name":"Linus","age":28}] + """), + type) + .runToList(); + var array = + JsonFlow.parseArray( + byteFlow( + """ + [[{"name":"Ada","age":36}],[{"name":"Grace","age":37}]] + """), + type) + .runToList(); + + // then + assertEquals( + List.of( + List.of(new Person("Ada", 36)), + List.of(new Person("Grace", 37), new Person("Linus", 28))), + ndjson); + assertEquals( + List.of(List.of(new Person("Ada", 36)), List.of(new Person("Grace", 37))), array); + } + + @Test + void shouldParseJsonNodesUsingConfiguredReaderOverloads() throws Exception { + // given + var reader = MAPPER.readerFor(JsonNode.class); + + // when + List ndjson = + JsonFlow.parseNdjson(byteFlow("{\"n\":1}\n[true,null]\n"), reader) + .runToList(); + List array = + JsonFlow.parseArray(byteFlow("[{\"n\":1},[true,null]]"), reader) + .runToList(); + + // then + assertEquals(List.of(MAPPER.readTree("{\"n\":1}"), MAPPER.readTree("[true,null]")), ndjson); + assertEquals(ndjson, array); + } + + @Test + void shouldRejectMalformedNdjsonAndMultipleValuesOnOneLine() { + assertFails( + () -> JsonFlow.parseNdjson(byteFlow("{\"name\":}\n"), Person.class).runToList()); + assertFails( + () -> + JsonFlow.parseNdjson( + byteFlow("{\"name\":\"Ada\",\"age\":36} true\n"), + Person.class) + .runToList()); + } + + @Test + void shouldRejectMalformedArrayWrongTopLevelShapeAndTrailingContent() { + assertFails( + () -> + JsonFlow.parseArray(byteFlow("[{\"name\":\"Ada\"}"), Person.class) + .runToList()); + + var wrongShape = + assertThrows( + Exception.class, + () -> + JsonFlow.parseArray(byteFlow("{\"name\":\"Ada\"}"), Person.class) + .runToList()); + assertCauseMessage(wrongShape, "Expected one top-level JSON array"); + + var trailing = + assertThrows( + Exception.class, + () -> JsonFlow.parseArray(byteFlow("[] true"), Person.class).runToList()); + assertCauseMessage(trailing, "Unexpected content after the top-level JSON array"); + } + + @Test + void shouldPropagateParsingUpstreamErrors() { + // given + var ndjsonFailure = new IllegalStateException("ndjson upstream failed"); + var arrayFailure = new IllegalStateException("array upstream failed"); + var ndjson = + Flows.concat( + Flows.fromByteArrays( + "{\"name\":\"Ada\",\"age\":36}\n" + .getBytes(StandardCharsets.UTF_8)), + Flows.failed(ndjsonFailure)) + .toByteFlow(); + var array = + Flows.concat( + Flows.fromByteArrays( + "[{\"name\":\"Ada\",\"age\":36}" + .getBytes(StandardCharsets.UTF_8)), + Flows.failed(arrayFailure)) + .toByteFlow(); + + // when + var ndjsonException = + assertThrows( + Exception.class, + () -> JsonFlow.parseNdjson(ndjson, Person.class).runToList()); + var arrayException = + assertThrows( + Exception.class, + () -> JsonFlow.parseArray(array, Person.class).runToList()); + + // then + assertHasCause(ndjsonException, ndjsonFailure); + assertHasCause(arrayException, arrayFailure); + } + + @Test + void shouldBeLazyAndStopArrayUpstreamAfterDownstreamTakesInitialElements() throws Exception { + // given + var input = new StringBuilder("["); + for (int i = 0; i < 20_000; i++) { + if (i > 0) { + input.append(','); + } + input.append(i); + } + input.append(']'); + var bytes = input.toString().getBytes(StandardCharsets.UTF_8); + var emittedBytes = new AtomicInteger(); + var source = + Flows.usingEmit( + emit -> { + for (byte value : bytes) { + emittedBytes.incrementAndGet(); + emit.apply(ByteChunk.fromArray(new byte[] {value})); + } + }) + .toByteFlow(); + + // when + var parsed = JsonFlow.parseArray(source, Integer.class); + + // then + assertEquals(0, emittedBytes.get()); + assertEquals(List.of(0, 1), parsed.take(2).runToList()); + assertTrue(emittedBytes.get() < bytes.length); + } + + @Test + void shouldStopNdjsonUpstreamAndPropagateDownstreamFailure() throws Exception { + // given + var emittedRecords = new AtomicInteger(); + var source = + Flows.usingEmit( + emit -> { + for (int i = 0; i < 100; i++) { + emittedRecords.incrementAndGet(); + emit.apply( + ByteChunk.fromArray( + ("%d\n".formatted(i)) + .getBytes(StandardCharsets.UTF_8))); + } + }) + .toByteFlow(); + var downstreamFailure = new IllegalStateException("downstream failed"); + + // when + var exception = + assertThrows( + Exception.class, + () -> + JsonFlow.parseNdjson(source, Integer.class) + .map( + value -> { + if (value == 1) { + throw downstreamFailure; + } + return value; + }) + .runToList()); + + // then + assertHasCause(exception, downstreamFailure); + assertEquals(2, emittedRecords.get()); + } + + @Test + void shouldRunParsingFlowsRepeatedly() throws Exception { + // given + var ndjson = JsonFlow.parseNdjson(byteFlow("1\n2\n"), Integer.class); + var array = JsonFlow.parseArray(byteFlow("[1,2]"), Integer.class); + + // when & then + assertEquals(List.of(1, 2), ndjson.runToList()); + assertEquals(List.of(1, 2), ndjson.runToList()); + assertEquals(List.of(1, 2), array.runToList()); + assertEquals(List.of(1, 2), array.runToList()); + } + + @Test + void shouldRenderNdjsonWithMandatoryNewlinesUsingClassOverload() throws Exception { + // given + var values = Flows.fromValues(new Person("Ada", 36), new Person("Łukasz", 41)); + + // when + var result = render(JsonFlow.renderNdjson(values, Person.class)); + + // then + assertEquals( + """ + {"name":"Ada","age":36} + {"name":"Łukasz","age":41} + """, + result); + } + + @Test + void shouldRenderEmptyFlows() throws Exception { + assertEquals("", render(JsonFlow.renderNdjson(Flows.empty(), Person.class))); + assertEquals("[]", render(JsonFlow.renderArray(Flows.empty(), Person.class))); + } + + @Test + void shouldRenderGenericTypesUsingTypeReferenceOverloads() throws Exception { + // given + TypeReference> type = new TypeReference<>() {}; + Flow> values = Flows.fromValues(List.of(1, 2), List.of(3)); + + // when & then + assertEquals("[1,2]\n[3]\n", render(JsonFlow.renderNdjson(values, type))); + assertEquals( + "[[1,2],[3]]", + render(JsonFlow.renderArray(Flows.fromValues(List.of(1, 2), List.of(3)), type))); + } + + @Test + void shouldRenderJsonNodesUsingConfiguredWriterOverloads() throws Exception { + // given + var writer = MAPPER.writerFor(JsonNode.class); + var values = Flows.fromValues(MAPPER.readTree("{\"n\":1}"), MAPPER.readTree("[true,null]")); + + // when & then + assertEquals("{\"n\":1}\n[true,null]\n", render(JsonFlow.renderNdjson(values, writer))); + assertEquals( + "[{\"n\":1},[true,null]]", + render( + JsonFlow.renderArray( + Flows.fromValues( + MAPPER.readTree("{\"n\":1}"), + MAPPER.readTree("[true,null]")), + writer))); + } + + @Test + void shouldRejectRawLineBreaksProducedByNdjsonWriter() { + // given + var prettyWriter = MAPPER.writerFor(Person.class).withDefaultPrettyPrinter(); + + // when + var exception = + assertThrows( + Exception.class, + () -> + JsonFlow.renderNdjson( + Flows.fromValues(new Person("Ada", 36)), + prettyWriter) + .runToList()); + + // then + assertCauseMessage(exception, "cannot be rendered as NDJSON"); + } + + @Test + void shouldPropagateRenderingUpstreamErrors() { + // given + var ndjsonFailure = new IllegalStateException("ndjson values failed"); + var arrayFailure = new IllegalStateException("array values failed"); + var ndjsonValues = + Flows.concat( + Flows.fromValues(new Person("Ada", 36)), + Flows.failed(ndjsonFailure)); + var arrayValues = + Flows.concat( + Flows.fromValues(new Person("Ada", 36)), + Flows.failed(arrayFailure)); + + // when + var ndjsonException = + assertThrows( + Exception.class, + () -> JsonFlow.renderNdjson(ndjsonValues, Person.class).runToList()); + var arrayException = + assertThrows( + Exception.class, + () -> JsonFlow.renderArray(arrayValues, Person.class).runToList()); + + // then + assertHasCause(ndjsonException, ndjsonFailure); + assertHasCause(arrayException, arrayFailure); + } + + @Test + void shouldRenderLazilyAndStopAfterDownstreamTakesInitialChunks() throws Exception { + // given + var renderedValues = new AtomicInteger(); + var values = + Flows.usingEmit( + emit -> { + for (int i = 0; i < 100; i++) { + renderedValues.incrementAndGet(); + emit.apply(i); + } + }); + + // when + var rendered = JsonFlow.renderArray(values, Integer.class); + + // then + assertEquals(0, renderedValues.get()); + assertEquals("[0,1", chunksToString(rendered.take(3).runToList())); + assertEquals(2, renderedValues.get()); + } + + @Test + void shouldRunRenderingFlowsRepeatedly() throws Exception { + // given + var ndjson = JsonFlow.renderNdjson(Flows.fromValues(1, 2), Integer.class); + var array = JsonFlow.renderArray(Flows.fromValues(1, 2), Integer.class); + + // when & then + assertEquals("1\n2\n", render(ndjson)); + assertEquals("1\n2\n", render(ndjson)); + assertEquals("[1,2]", render(array)); + assertEquals("[1,2]", render(array)); + } + + @Test + void shouldRoundTripNdjsonAndArrays() throws Exception { + // given + var people = + List.of( + new Person("Zażółć 🦊", 7), + new Person("東京", 10), + new Person("Málaga 🌊", 20)); + + // when + var ndjson = + JsonFlow.parseNdjson( + JsonFlow.renderNdjson(Flows.fromIterable(people), Person.class), + Person.class) + .runToList(); + var array = + JsonFlow.parseArray( + JsonFlow.renderArray(Flows.fromIterable(people), Person.class), + Person.class) + .runToList(); + + // then + assertEquals(people, ndjson); + assertEquals(people, array); + } + + @Test + void shouldUseInputStreamFileAndRenderedOutputIntegrations() throws Exception { + // given + var stream = + new ByteArrayInputStream( + "[{\"name\":\"Ada\",\"age\":36}]".getBytes(StandardCharsets.UTF_8)); + var path = tempDir.resolve("people.ndjson"); + Files.writeString( + path, + "{\"name\":\"Grace\",\"age\":37}\n{\"name\":\"Linus\",\"age\":28}", + StandardCharsets.UTF_8); + + // when + var fromInputStream = + JsonFlow.parseArray(Flows.fromInputStream(stream, 1), Person.class).runToList(); + var fromFile = JsonFlow.parseNdjson(Flows.fromFile(path, 3), Person.class).runToList(); + var output = + render(JsonFlow.renderArray(Flows.fromValues(new Person("Ada", 36)), Person.class)); + + // then + assertEquals(List.of(new Person("Ada", 36)), fromInputStream); + assertEquals(List.of(new Person("Grace", 37), new Person("Linus", 28)), fromFile); + assertEquals("[{\"name\":\"Ada\",\"age\":36}]", output); + } + + private static Flow.ByteFlow byteFlow(String value) { + return Flows.fromByteArrays(value.getBytes(StandardCharsets.UTF_8)); + } + + private static Flow.ByteFlow oneByteChunks(String value) { + var bytes = value.getBytes(StandardCharsets.UTF_8); + var chunks = new ByteChunk[bytes.length]; + for (int i = 0; i < bytes.length; i++) { + chunks[i] = ByteChunk.fromArray(new byte[] {bytes[i]}); + } + return Flows.fromByteChunks(chunks); + } + + private static String render(Flow.ByteFlow flow) throws Exception { + var output = new ByteArrayOutputStream(); + flow.runToOutputStream(output); + return output.toString(StandardCharsets.UTF_8); + } + + private static String chunksToString(List chunks) { + var output = new ByteArrayOutputStream(); + for (var chunk : chunks) { + for (var array : chunk.getArrays()) { + output.writeBytes(array); + } + } + return output.toString(StandardCharsets.UTF_8); + } + + private static void assertFails(ThrowingRunnable action) { + assertThrows(Exception.class, action::run); + } + + private static void assertCauseMessage(Throwable exception, String expectedFragment) { + for (Throwable current = exception; current != null; current = current.getCause()) { + if (current.getMessage() != null && current.getMessage().contains(expectedFragment)) { + return; + } + } + throw new AssertionError( + "No exception in the cause chain contained: " + expectedFragment, exception); + } + + private static void assertHasCause(Throwable exception, Throwable expected) { + for (Throwable current = exception; current != null; current = current.getCause()) { + if (current == expected) { + assertSame(expected, current); + return; + } + } + throw new AssertionError( + "Expected exception was not present in the cause chain", exception); + } + + @FunctionalInterface + private interface ThrowingRunnable { + void run() throws Exception; + } + + private record Person(String name, int age) {} +} From 1264baf9e574bdbdef9992d1969dccfdfc7af509 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sebastian=20=C5=81akomy?= Date: Thu, 6 Aug 2026 15:31:09 +0200 Subject: [PATCH 04/13] docs(json): document streaming JSON integration Document the json dependency, supported wire formats, Jackson configuration, and composition with Jox flows and I/O. --- docs/index.md | 4 +- docs/json.md | 231 ++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 234 insertions(+), 1 deletion(-) create mode 100644 docs/json.md diff --git a/docs/index.md b/docs/index.md index 9b53f10..c69bba4 100644 --- a/docs/index.md +++ b/docs/index.md @@ -3,13 +3,14 @@ [Virtual-thread](https://docs.oracle.com/en/java/javase/21/core/virtual-threads.html) based safe concurrency & streaming for Java. Open-source, Apache2 licensed. -Jox contains three main modules: +Jox contains five main modules: * Fast & scalable, completable [channels](channels.md), with Go-like `select`s (Java 21+) * Programmer-friendly [structured concurrency](structured.md) (Java 25 only) * Finite & infinite streaming using [flows](flows.md), with reactive streams compatibility, (blocking) I/O integration and a high-level, "functional" API (Java 25 only) * [Kafka](kafka.md) integration for reading from and writing to Kafka topics using flows (Java 25 only) +* [JSON](json.md) integration for streaming NDJSON and top-level JSON arrays using flows (Java 25 only) Source code is [available on GitHub](https://github.com/softwaremill/jox). @@ -108,4 +109,5 @@ For a Scala version, see the [Ox project](https://github.com/softwaremill/ox). flows structured kafka + json contributing diff --git a/docs/json.md b/docs/json.md new file mode 100644 index 0000000..361bcc0 --- /dev/null +++ b/docs/json.md @@ -0,0 +1,231 @@ +# JSON flows + +Lazy, backpressured parsing and rendering of newline-delimited JSON (NDJSON) and top-level JSON arrays using Jox +`Flow` and `ByteFlow`. + +Requires Java 25 (current LTS). + +## Dependency + +Maven: + +```xml + + com.softwaremill.jox + json + 0.1.0 + +``` + +Gradle: + +```groovy +implementation 'com.softwaremill.jox:json:0.1.0' +``` + +Gradle (Kotlin DSL): + +```kotlin +implementation("com.softwaremill.jox:json:0.1.0") +``` + +## API + +`JsonFlow` provides four transformations: + +* `parseNdjson(ByteFlow, ...)` parses UTF-8 NDJSON into a `Flow`. +* `parseArray(ByteFlow, ...)` parses one top-level JSON array into a `Flow` of its elements. +* `renderNdjson(Flow, ...)` renders values as a UTF-8 NDJSON `ByteFlow`. +* `renderArray(Flow, ...)` renders values as one UTF-8 JSON array `ByteFlow`. + +Each method is lazy: parsing, rendering and I/O start only when the returned flow is run. Values are processed one at +a time, preserving the backpressure, failure propagation and cancellation behavior of the underlying Jox flow. + +Each operation has overloads accepting a `Class`, a Jackson `TypeReference`, or a configured Jackson +`ObjectReader`/`ObjectWriter`. The `Class` and `TypeReference` overloads use the module's default `ObjectMapper`. + +## NDJSON + +NDJSON parsing accepts LF and CRLF line endings, ignores empty and whitespace-only lines, and accepts a final record +without a line ending. Every non-blank line must contain exactly one JSON value; malformed JSON or trailing content on +a record fails the flow when it is run. + +```java +import java.nio.charset.StandardCharsets; + +import com.softwaremill.jox.flows.Flow; +import com.softwaremill.jox.flows.Flows; +import com.softwaremill.jox.json.JsonFlow; + +record Event(long id, String message) {} + +void main() throws Exception { + var input = """ + {"id":1,"message":"created"} + {"id":2,"message":"updated"} + """; + + Flow events = JsonFlow.parseNdjson( + Flows.fromByteArrays(input.getBytes(StandardCharsets.UTF_8)), + Event.class); + + events.filter(event -> event.id() > 1) + .runForeach(System.out::println); +} +``` + +NDJSON rendering writes one JSON value followed by an LF byte. The final value also has a terminating LF. A writer +whose output contains raw CR or LF bytes is rejected, as such output would break NDJSON record boundaries. In +particular, do not use a pretty-printing writer for NDJSON. + +```java +import java.io.ByteArrayOutputStream; +import java.nio.charset.StandardCharsets; + +import com.softwaremill.jox.flows.Flows; +import com.softwaremill.jox.json.JsonFlow; + +record Event(long id, String message) {} + +void main() throws Exception { + var events = Flows.fromValues( + new Event(1, "created"), + new Event(2, "updated")); + + var output = new ByteArrayOutputStream(); + JsonFlow.renderNdjson(events, Event.class).runToOutputStream(output); + + System.out.print(output.toString(StandardCharsets.UTF_8)); +} +``` + +## JSON arrays + +Array parsing requires exactly one complete top-level array and emits every element as soon as it is decoded. A +different top-level JSON value, an incomplete array, malformed input, or JSON content after the array fails the flow. +An empty array produces an empty flow. + +```java +import java.nio.charset.StandardCharsets; + +import com.softwaremill.jox.flows.Flow; +import com.softwaremill.jox.flows.Flows; +import com.softwaremill.jox.json.JsonFlow; + +record Event(long id, String message) {} + +void main() throws Exception { + var input = """ + [ + {"id":1,"message":"created"}, + {"id":2,"message":"updated"} + ] + """; + + Flow events = JsonFlow.parseArray( + Flows.fromByteArrays(input.getBytes(StandardCharsets.UTF_8)), + Event.class); + + events.runForeach(System.out::println); +} +``` + +Array rendering writes `[` and `]` around comma-separated values. Elements are serialized one at a time, and an empty +input flow produces `[]`. + +```java +import java.nio.file.Path; + +import com.softwaremill.jox.flows.Flows; +import com.softwaremill.jox.json.JsonFlow; + +record Event(long id, String message) {} + +void main() throws Exception { + var events = Flows.fromValues( + new Event(1, "created"), + new Event(2, "updated")); + + JsonFlow.renderArray(events, Event.class) + .runToFile(Path.of("events.json")); +} +``` + +For generic element types, use a Jackson `TypeReference`: + +```java +import java.util.List; + +import com.softwaremill.jox.flows.Flow; +import com.softwaremill.jox.flows.Flow.ByteFlow; +import com.softwaremill.jox.json.JsonFlow; + +import tools.jackson.core.type.TypeReference; + +record Event(long id, String message) {} + +Flow> parseBatches(ByteFlow input) { + return JsonFlow.parseArray(input, new TypeReference>() {}); +} +``` + +## Jackson configuration + +For custom Jackson modules, naming strategies, date handling, polymorphism, tree-model values, or other mapper +features, configure an `ObjectMapper` and derive an `ObjectReader` or `ObjectWriter`. The reader or writer determines +the type and Jackson behavior for each NDJSON record or array element. + +```java +import com.softwaremill.jox.flows.Flow; +import com.softwaremill.jox.flows.Flow.ByteFlow; +import com.softwaremill.jox.json.JsonFlow; + +import tools.jackson.databind.DeserializationFeature; +import tools.jackson.databind.ObjectMapper; + +record Event(long id, String message) {} + +Flow parseEvents(ByteFlow input) { + var mapper = new ObjectMapper(); + var reader = mapper.readerFor(Event.class) + .without(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES); + + return JsonFlow.parseNdjson(input, reader); +} +``` + +The same configured mapper can create a writer using `mapper.writerFor(Event.class)`, which can be passed to +`renderNdjson` or `renderArray`. + +## Composing with Jox flows and I/O + +The results are ordinary Jox `Flow` and `ByteFlow` values. Parsed values can use transformations such as `map`, +`filter`, `mapPar`, `buffer` and error recovery. Rendered bytes can be written using existing `ByteFlow` operations. +Likewise, JSON input can come from any `ByteFlow`, including files, `InputStream`s and in-memory byte chunks. + +For example, this pipeline reads NDJSON from a file, applies regular flow transformations, and streams one JSON array +to another file: + +```java +import java.nio.file.Path; + +import com.softwaremill.jox.flows.Flow; +import com.softwaremill.jox.flows.Flows; +import com.softwaremill.jox.json.JsonFlow; + +record Event(long id, boolean accepted) {} + +void main() throws Exception { + Flow accepted = JsonFlow.parseNdjson( + Flows.fromFile(Path.of("events.ndjson")), + Event.class) + .filter(Event::accepted) + .map(event -> new Event(event.id(), true)); + + JsonFlow.renderArray(accepted, Event.class) + .runToFile(Path.of("accepted.json")); +} +``` + +Use `Flows.fromInputStream(...)` for stream input, and `runToOutputStream(...)` for stream output. These operations +retain their normal Jox resource ownership: the input or output stream is closed when the flow finishes or fails. From 66410062edc7d1d165685a23f808116318266e19 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sebastian=20=C5=81akomy?= Date: Fri, 7 Aug 2026 14:52:00 +0200 Subject: [PATCH 05/13] style(json): inline requireNonNull checks in JsonFlow Keep null guards consistent across Class/TypeReference and reader/writer overloads. Co-authored-by: Cursor --- .../com/softwaremill/jox/json/JsonFlow.java | 24 +++++++------------ 1 file changed, 8 insertions(+), 16 deletions(-) diff --git a/json/src/main/java/com/softwaremill/jox/json/JsonFlow.java b/json/src/main/java/com/softwaremill/jox/json/JsonFlow.java index ac5c1cb..94fcb2c 100644 --- a/json/src/main/java/com/softwaremill/jox/json/JsonFlow.java +++ b/json/src/main/java/com/softwaremill/jox/json/JsonFlow.java @@ -33,8 +33,7 @@ private JsonFlow() {} * @return a flow emitting one value for each non-blank input line */ public static Flow parseNdjson(ByteFlow bytes, Class valueType) { - Objects.requireNonNull(valueType, "valueType"); - return parseNdjson(bytes, DEFAULT_MAPPER.readerFor(valueType)); + return parseNdjson(bytes, DEFAULT_MAPPER.readerFor(Objects.requireNonNull(valueType, "valueType"))); } /** @@ -48,8 +47,7 @@ public static Flow parseNdjson(ByteFlow bytes, Class valueType) { * @return a flow emitting one value for each non-blank input line */ public static Flow parseNdjson(ByteFlow bytes, TypeReference valueType) { - Objects.requireNonNull(valueType, "valueType"); - return parseNdjson(bytes, DEFAULT_MAPPER.readerFor(valueType)); + return parseNdjson(bytes, DEFAULT_MAPPER.readerFor(Objects.requireNonNull(valueType, "valueType"))); } /** @@ -78,8 +76,7 @@ public static Flow parseNdjson(ByteFlow bytes, ObjectReader reader) { * @return a flow emitting the array elements */ public static Flow parseArray(ByteFlow bytes, Class valueType) { - Objects.requireNonNull(valueType, "valueType"); - return parseArray(bytes, DEFAULT_MAPPER.readerFor(valueType)); + return parseArray(bytes, DEFAULT_MAPPER.readerFor(Objects.requireNonNull(valueType, "valueType"))); } /** @@ -93,8 +90,7 @@ public static Flow parseArray(ByteFlow bytes, Class valueType) { * @return a flow emitting the array elements */ public static Flow parseArray(ByteFlow bytes, TypeReference valueType) { - Objects.requireNonNull(valueType, "valueType"); - return parseArray(bytes, DEFAULT_MAPPER.readerFor(valueType)); + return parseArray(bytes, DEFAULT_MAPPER.readerFor(Objects.requireNonNull(valueType, "valueType"))); } /** @@ -122,8 +118,7 @@ public static Flow parseArray(ByteFlow bytes, ObjectReader reader) { * @return a flow emitting UTF-8 encoded NDJSON */ public static ByteFlow renderNdjson(Flow values, Class valueType) { - Objects.requireNonNull(valueType, "valueType"); - return renderNdjson(values, DEFAULT_MAPPER.writerFor(valueType)); + return renderNdjson(values, DEFAULT_MAPPER.writerFor(Objects.requireNonNull(valueType, "valueType"))); } /** @@ -136,8 +131,7 @@ public static ByteFlow renderNdjson(Flow values, Class valueType) { * @return a flow emitting UTF-8 encoded NDJSON */ public static ByteFlow renderNdjson(Flow values, TypeReference valueType) { - Objects.requireNonNull(valueType, "valueType"); - return renderNdjson(values, DEFAULT_MAPPER.writerFor(valueType)); + return renderNdjson(values, DEFAULT_MAPPER.writerFor(Objects.requireNonNull(valueType, "valueType"))); } /** @@ -165,8 +159,7 @@ public static ByteFlow renderNdjson(Flow values, ObjectWriter writer) { * @return a flow emitting one UTF-8 encoded JSON array */ public static ByteFlow renderArray(Flow values, Class valueType) { - Objects.requireNonNull(valueType, "valueType"); - return renderArray(values, DEFAULT_MAPPER.writerFor(valueType)); + return renderArray(values, DEFAULT_MAPPER.writerFor(Objects.requireNonNull(valueType, "valueType"))); } /** @@ -179,8 +172,7 @@ public static ByteFlow renderArray(Flow values, Class valueType) { * @return a flow emitting one UTF-8 encoded JSON array */ public static ByteFlow renderArray(Flow values, TypeReference valueType) { - Objects.requireNonNull(valueType, "valueType"); - return renderArray(values, DEFAULT_MAPPER.writerFor(valueType)); + return renderArray(values, DEFAULT_MAPPER.writerFor(Objects.requireNonNull(valueType, "valueType"))); } /** From e3cee2259171869f2daaf30896ce58de950bc4cf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sebastian=20=C5=81akomy?= Date: Fri, 7 Aug 2026 16:00:26 +0200 Subject: [PATCH 06/13] refactor(json): render arrays using intersperse Express JSON delimiters through flow composition instead of mutable first-element state, without treating byte chunk boundaries as API behavior. Co-authored-by: Cursor --- .../com/softwaremill/jox/json/JsonRendering.java | 16 ++-------------- .../com/softwaremill/jox/json/JsonFlowTest.java | 4 ++-- 2 files changed, 4 insertions(+), 16 deletions(-) diff --git a/json/src/main/java/com/softwaremill/jox/json/JsonRendering.java b/json/src/main/java/com/softwaremill/jox/json/JsonRendering.java index c80f3a3..3968bdb 100644 --- a/json/src/main/java/com/softwaremill/jox/json/JsonRendering.java +++ b/json/src/main/java/com/softwaremill/jox/json/JsonRendering.java @@ -29,20 +29,8 @@ static ByteFlow renderNdjson(Flow values, ObjectWriter writer) { } static ByteFlow renderArray(Flow values, ObjectWriter writer) { - return Flows.usingEmit( - emit -> { - emit.apply(ARRAY_START); - boolean[] first = {true}; - values.runForeach( - value -> { - ByteChunk json = - ByteChunk.fromArray( - writer.writeValueAsBytes(value)); - emit.apply(first[0] ? json : COMMA.concat(json)); - first[0] = false; - }); - emit.apply(ARRAY_END); - }) + return values.map(value -> ByteChunk.fromArray(writer.writeValueAsBytes(value))) + .intersperse(ARRAY_START, COMMA, ARRAY_END) .toByteFlow(); } diff --git a/json/src/test/java/com/softwaremill/jox/json/JsonFlowTest.java b/json/src/test/java/com/softwaremill/jox/json/JsonFlowTest.java index a48e6f2..f6ce66e 100644 --- a/json/src/test/java/com/softwaremill/jox/json/JsonFlowTest.java +++ b/json/src/test/java/com/softwaremill/jox/json/JsonFlowTest.java @@ -405,8 +405,8 @@ void shouldRenderLazilyAndStopAfterDownstreamTakesInitialChunks() throws Excepti // then assertEquals(0, renderedValues.get()); - assertEquals("[0,1", chunksToString(rendered.take(3).runToList())); - assertEquals(2, renderedValues.get()); + assertEquals("[0", chunksToString(rendered.take(2).runToList())); + assertEquals(1, renderedValues.get()); } @Test From b2e3f0ad40f57674100880170ca15f1e995e42be Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sebastian=20=C5=81akomy?= Date: Mon, 10 Aug 2026 09:26:21 +0200 Subject: [PATCH 07/13] refactor(json): compose NDJSON rendering stages Separate serialization, record validation, and byte framing into explicit flow transformations. Co-authored-by: Cursor --- .../com/softwaremill/jox/json/JsonRendering.java | 14 ++++---------- 1 file changed, 4 insertions(+), 10 deletions(-) diff --git a/json/src/main/java/com/softwaremill/jox/json/JsonRendering.java b/json/src/main/java/com/softwaremill/jox/json/JsonRendering.java index 3968bdb..672d9f9 100644 --- a/json/src/main/java/com/softwaremill/jox/json/JsonRendering.java +++ b/json/src/main/java/com/softwaremill/jox/json/JsonRendering.java @@ -3,7 +3,6 @@ import com.softwaremill.jox.flows.ByteChunk; import com.softwaremill.jox.flows.Flow; import com.softwaremill.jox.flows.Flow.ByteFlow; -import com.softwaremill.jox.flows.Flows; import tools.jackson.databind.ObjectWriter; @@ -17,14 +16,9 @@ final class JsonRendering { private JsonRendering() {} static ByteFlow renderNdjson(Flow values, ObjectWriter writer) { - return Flows.usingEmit( - emit -> - values.runForeach( - value -> { - byte[] json = writer.writeValueAsBytes(value); - rejectLineBreaks(json); - emit.apply(ByteChunk.fromArray(json).concat(NEW_LINE)); - })) + return values.map(writer::writeValueAsBytes) + .tap(JsonRendering::requireNoLineBreaks) + .map(json -> ByteChunk.fromArray(json).concat(NEW_LINE)) .toByteFlow(); } @@ -34,7 +28,7 @@ static ByteFlow renderArray(Flow values, ObjectWriter writer) { .toByteFlow(); } - private static void rejectLineBreaks(byte[] json) { + private static void requireNoLineBreaks(byte[] json) { for (byte value : json) { if (value == '\r' || value == '\n') { throw new IllegalArgumentException( From b87111b6b332d4eab4070b05c081104c71440914 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sebastian=20=C5=81akomy?= Date: Mon, 10 Aug 2026 10:05:40 +0200 Subject: [PATCH 08/13] fix(json): reject null flow elements 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 --- docs/json.md | 3 + .../com/softwaremill/jox/json/JsonFlow.java | 28 ++- .../softwaremill/jox/json/JsonParsing.java | 14 +- .../softwaremill/jox/json/JsonFlowTest.java | 191 ++++++++++++++++++ 4 files changed, 225 insertions(+), 11 deletions(-) diff --git a/docs/json.md b/docs/json.md index 361bcc0..d511412 100644 --- a/docs/json.md +++ b/docs/json.md @@ -41,6 +41,9 @@ implementation("com.softwaremill.jox:json:0.1.0") Each method is lazy: parsing, rendering and I/O start only when the returned flow is run. Values are processed one at a time, preserving the backpressure, failure propagation and cancellation behavior of the underlying Jox flow. +Parsing rejects top-level values that Jackson deserializes as Java `null`, as Jox flows do not support `null` elements. +To retain JSON `null` values, deserialize into Jackson's tree model, where they are represented by non-null null nodes. + Each operation has overloads accepting a `Class`, a Jackson `TypeReference`, or a configured Jackson `ObjectReader`/`ObjectWriter`. The `Class` and `TypeReference` overloads use the module's default `ObjectMapper`. diff --git a/json/src/main/java/com/softwaremill/jox/json/JsonFlow.java b/json/src/main/java/com/softwaremill/jox/json/JsonFlow.java index 94fcb2c..f227489 100644 --- a/json/src/main/java/com/softwaremill/jox/json/JsonFlow.java +++ b/json/src/main/java/com/softwaremill/jox/json/JsonFlow.java @@ -14,7 +14,9 @@ * Creates flows which parse or render newline-delimited JSON (NDJSON) and top-level JSON arrays. * *

All transformations are lazy and preserve the backpressure and cancellation behavior of the - * supplied flow. Values are parsed or rendered one at a time. + * supplied flow. Values are parsed or rendered one at a time. Parsing fails when Jackson + * deserializes a top-level value as {@code null}, which Jox flows do not support. Use Jackson's + * tree model to represent a JSON {@code null} as a non-null node. */ public final class JsonFlow { @@ -33,7 +35,8 @@ private JsonFlow() {} * @return a flow emitting one value for each non-blank input line */ public static Flow parseNdjson(ByteFlow bytes, Class valueType) { - return parseNdjson(bytes, DEFAULT_MAPPER.readerFor(Objects.requireNonNull(valueType, "valueType"))); + return parseNdjson( + bytes, DEFAULT_MAPPER.readerFor(Objects.requireNonNull(valueType, "valueType"))); } /** @@ -47,7 +50,8 @@ public static Flow parseNdjson(ByteFlow bytes, Class valueType) { * @return a flow emitting one value for each non-blank input line */ public static Flow parseNdjson(ByteFlow bytes, TypeReference valueType) { - return parseNdjson(bytes, DEFAULT_MAPPER.readerFor(Objects.requireNonNull(valueType, "valueType"))); + return parseNdjson( + bytes, DEFAULT_MAPPER.readerFor(Objects.requireNonNull(valueType, "valueType"))); } /** @@ -76,7 +80,8 @@ public static Flow parseNdjson(ByteFlow bytes, ObjectReader reader) { * @return a flow emitting the array elements */ public static Flow parseArray(ByteFlow bytes, Class valueType) { - return parseArray(bytes, DEFAULT_MAPPER.readerFor(Objects.requireNonNull(valueType, "valueType"))); + return parseArray( + bytes, DEFAULT_MAPPER.readerFor(Objects.requireNonNull(valueType, "valueType"))); } /** @@ -90,7 +95,8 @@ public static Flow parseArray(ByteFlow bytes, Class valueType) { * @return a flow emitting the array elements */ public static Flow parseArray(ByteFlow bytes, TypeReference valueType) { - return parseArray(bytes, DEFAULT_MAPPER.readerFor(Objects.requireNonNull(valueType, "valueType"))); + return parseArray( + bytes, DEFAULT_MAPPER.readerFor(Objects.requireNonNull(valueType, "valueType"))); } /** @@ -118,7 +124,8 @@ public static Flow parseArray(ByteFlow bytes, ObjectReader reader) { * @return a flow emitting UTF-8 encoded NDJSON */ public static ByteFlow renderNdjson(Flow values, Class valueType) { - return renderNdjson(values, DEFAULT_MAPPER.writerFor(Objects.requireNonNull(valueType, "valueType"))); + return renderNdjson( + values, DEFAULT_MAPPER.writerFor(Objects.requireNonNull(valueType, "valueType"))); } /** @@ -131,7 +138,8 @@ public static ByteFlow renderNdjson(Flow values, Class valueType) { * @return a flow emitting UTF-8 encoded NDJSON */ public static ByteFlow renderNdjson(Flow values, TypeReference valueType) { - return renderNdjson(values, DEFAULT_MAPPER.writerFor(Objects.requireNonNull(valueType, "valueType"))); + return renderNdjson( + values, DEFAULT_MAPPER.writerFor(Objects.requireNonNull(valueType, "valueType"))); } /** @@ -159,7 +167,8 @@ public static ByteFlow renderNdjson(Flow values, ObjectWriter writer) { * @return a flow emitting one UTF-8 encoded JSON array */ public static ByteFlow renderArray(Flow values, Class valueType) { - return renderArray(values, DEFAULT_MAPPER.writerFor(Objects.requireNonNull(valueType, "valueType"))); + return renderArray( + values, DEFAULT_MAPPER.writerFor(Objects.requireNonNull(valueType, "valueType"))); } /** @@ -172,7 +181,8 @@ public static ByteFlow renderArray(Flow values, Class valueType) { * @return a flow emitting one UTF-8 encoded JSON array */ public static ByteFlow renderArray(Flow values, TypeReference valueType) { - return renderArray(values, DEFAULT_MAPPER.writerFor(Objects.requireNonNull(valueType, "valueType"))); + return renderArray( + values, DEFAULT_MAPPER.writerFor(Objects.requireNonNull(valueType, "valueType"))); } /** diff --git a/json/src/main/java/com/softwaremill/jox/json/JsonParsing.java b/json/src/main/java/com/softwaremill/jox/json/JsonParsing.java index 466d9a9..393f875 100644 --- a/json/src/main/java/com/softwaremill/jox/json/JsonParsing.java +++ b/json/src/main/java/com/softwaremill/jox/json/JsonParsing.java @@ -19,7 +19,7 @@ static Flow parseNdjson(ByteFlow bytes, ObjectReader reader) { var singleValueReader = reader.with(DeserializationFeature.FAIL_ON_TRAILING_TOKENS); return bytes.linesUtf8() .filter(line -> !line.isBlank()) - .map(line -> singleValueReader.readValue(line)); + .map(line -> requireNonNullValue(singleValueReader.readValue(line))); } static Flow parseArray(ByteFlow bytes, ObjectReader reader) { @@ -44,7 +44,9 @@ static Flow parseArray(ByteFlow bytes, ObjectReader reader) { "Unexpected end of input while parsing the" + " top-level JSON array"); } - emit.apply(elementReader.readValue(parser)); + emit.apply( + requireNonNullValue( + elementReader.readValue(parser))); } if (parser.nextToken() != null) { @@ -57,6 +59,14 @@ static Flow parseArray(ByteFlow bytes, ObjectReader reader) { })); } + private static T requireNonNullValue(T value) { + if (value == null) { + throw new IllegalArgumentException( + "JSON null cannot be emitted because Jox flows do not support null values"); + } + return value; + } + private static void requireToken(JsonToken actual, JsonToken expected, String message) { if (actual != expected) { throw new IllegalArgumentException(message); diff --git a/json/src/test/java/com/softwaremill/jox/json/JsonFlowTest.java b/json/src/test/java/com/softwaremill/jox/json/JsonFlowTest.java index f6ce66e..a2509d3 100644 --- a/json/src/test/java/com/softwaremill/jox/json/JsonFlowTest.java +++ b/json/src/test/java/com/softwaremill/jox/json/JsonFlowTest.java @@ -11,6 +11,7 @@ import java.nio.file.Files; import java.nio.file.Path; import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Test; @@ -27,6 +28,10 @@ class JsonFlowTest { private static final ObjectMapper MAPPER = new ObjectMapper(); + private static final IllegalStateException DESERIALIZATION_FAILURE = + new IllegalStateException("deserialization failed"); + private static final IllegalStateException SERIALIZATION_FAILURE = + new IllegalStateException("serialization failed"); @TempDir Path tempDir; @@ -133,6 +138,33 @@ void shouldParseJsonNodesUsingConfiguredReaderOverloads() throws Exception { assertEquals(ndjson, array); } + @Test + void shouldRejectDeserializedNullValuesButAllowJsonNullNodes() throws Exception { + // when + var ndjsonException = + assertThrows( + Exception.class, + () -> JsonFlow.parseNdjson(byteFlow("null\n"), String.class).runToList()); + var arrayException = + assertThrows( + Exception.class, + () -> + JsonFlow.parseArray(byteFlow("[null]"), String.class) + .buffer() + .runToList()); + var nullNode = MAPPER.readTree("null"); + + // then + assertCauseMessage(ndjsonException, "Jox flows do not support null values"); + assertCauseMessage(arrayException, "Jox flows do not support null values"); + assertEquals( + List.of(nullNode), + JsonFlow.parseNdjson(byteFlow("null\n"), JsonNode.class).runToList()); + assertEquals( + List.of(nullNode), + JsonFlow.parseArray(byteFlow("[null]"), JsonNode.class).runToList()); + } + @Test void shouldRejectMalformedNdjsonAndMultipleValuesOnOneLine() { assertFails( @@ -145,6 +177,26 @@ void shouldRejectMalformedNdjsonAndMultipleValuesOnOneLine() { .runToList()); } + @Test + void shouldHandleArrayWhitespaceAndRejectMissingInputAndTrailingCommas() throws Exception { + assertEquals( + List.of(1), + JsonFlow.parseArray(byteFlow(" \n\t[ 1 ]\r\n "), Integer.class).runToList()); + + var empty = + assertThrows( + Exception.class, + () -> JsonFlow.parseArray(byteFlow(""), Integer.class).runToList()); + var whitespaceOnly = + assertThrows( + Exception.class, + () -> JsonFlow.parseArray(byteFlow(" \r\n\t"), Integer.class).runToList()); + + assertCauseMessage(empty, "Expected one top-level JSON array"); + assertCauseMessage(whitespaceOnly, "Expected one top-level JSON array"); + assertFails(() -> JsonFlow.parseArray(byteFlow("[1,]"), Integer.class).runToList()); + } + @Test void shouldRejectMalformedArrayWrongTopLevelShapeAndTrailingContent() { assertFails( @@ -234,6 +286,61 @@ void shouldBeLazyAndStopArrayUpstreamAfterDownstreamTakesInitialElements() throw assertTrue(emittedBytes.get() < bytes.length); } + @Test + void shouldCancelAndCloseArrayInputAfterDownstreamFailure() { + // given + var input = new StringBuilder("["); + for (int i = 0; i < 20_000; i++) { + if (i > 0) { + input.append(','); + } + input.append(i); + } + input.append(']'); + var bytes = input.toString().getBytes(StandardCharsets.UTF_8); + var readBytes = new AtomicInteger(); + var closed = new AtomicBoolean(); + var inputStream = + new ByteArrayInputStream(bytes) { + @Override + public synchronized int read(byte[] target, int offset, int length) { + int read = super.read(target, offset, length); + if (read > 0) { + readBytes.addAndGet(read); + } + return read; + } + + @Override + public void close() { + closed.set(true); + } + }; + var downstreamFailure = new IllegalStateException("downstream failed"); + + // when + var exception = + assertThrows( + Exception.class, + () -> + JsonFlow.parseArray( + Flows.fromInputStream(inputStream, 1), + Integer.class) + .map( + value -> { + if (value == 1) { + throw downstreamFailure; + } + return value; + }) + .runToList()); + + // then + assertHasCause(exception, downstreamFailure); + assertTrue(closed.get()); + assertTrue(readBytes.get() < bytes.length); + } + @Test void shouldStopNdjsonUpstreamAndPropagateDownstreamFailure() throws Exception { // given @@ -302,6 +409,20 @@ void shouldRenderNdjsonWithMandatoryNewlinesUsingClassOverload() throws Exceptio result); } + @Test + void shouldAllowEscapedLineBreaksInNdjsonValues() throws Exception { + // given + var value = "first line\nsecond line\rthird line"; + + // when + var rendered = render(JsonFlow.renderNdjson(Flows.fromValues(value), String.class)); + + // then + assertEquals("\"first line\\nsecond line\\rthird line\"\n", rendered); + assertEquals( + List.of(value), JsonFlow.parseNdjson(byteFlow(rendered), String.class).runToList()); + } + @Test void shouldRenderEmptyFlows() throws Exception { assertEquals("", render(JsonFlow.renderNdjson(Flows.empty(), Person.class))); @@ -358,6 +479,21 @@ void shouldRejectRawLineBreaksProducedByNdjsonWriter() { assertCauseMessage(exception, "cannot be rendered as NDJSON"); } + @Test + void shouldAllowPrettyPrintedArrayElements() throws Exception { + // given + var person = new Person("Ada", 36); + var prettyWriter = MAPPER.writerFor(Person.class).withDefaultPrettyPrinter(); + + // when + var rendered = render(JsonFlow.renderArray(Flows.fromValues(person), prettyWriter)); + + // then + assertTrue(rendered.contains("\n")); + assertEquals( + List.of(person), JsonFlow.parseArray(byteFlow(rendered), Person.class).runToList()); + } + @Test void shouldPropagateRenderingUpstreamErrors() { // given @@ -387,6 +523,49 @@ void shouldPropagateRenderingUpstreamErrors() { assertHasCause(arrayException, arrayFailure); } + @Test + void shouldPropagateJacksonReaderAndWriterFailures() { + // when + var ndjsonReaderException = + assertThrows( + Exception.class, + () -> + JsonFlow.parseNdjson( + byteFlow("{\"value\":\"x\"}\n"), + FailingDeserialization.class) + .runToList()); + var arrayReaderException = + assertThrows( + Exception.class, + () -> + JsonFlow.parseArray( + byteFlow("[{\"value\":\"x\"}]"), + FailingDeserialization.class) + .runToList()); + var ndjsonWriterException = + assertThrows( + Exception.class, + () -> + JsonFlow.renderNdjson( + Flows.fromValues(new FailingSerialization()), + FailingSerialization.class) + .runToList()); + var arrayWriterException = + assertThrows( + Exception.class, + () -> + JsonFlow.renderArray( + Flows.fromValues(new FailingSerialization()), + FailingSerialization.class) + .runToList()); + + // then + assertHasCause(ndjsonReaderException, DESERIALIZATION_FAILURE); + assertHasCause(arrayReaderException, DESERIALIZATION_FAILURE); + assertHasCause(ndjsonWriterException, SERIALIZATION_FAILURE); + assertHasCause(arrayWriterException, SERIALIZATION_FAILURE); + } + @Test void shouldRenderLazilyAndStopAfterDownstreamTakesInitialChunks() throws Exception { // given @@ -532,5 +711,17 @@ private interface ThrowingRunnable { void run() throws Exception; } + private record FailingDeserialization(String value) { + private FailingDeserialization { + throw DESERIALIZATION_FAILURE; + } + } + + private static final class FailingSerialization { + public String getValue() { + throw SERIALIZATION_FAILURE; + } + } + private record Person(String name, int age) {} } From 318cff0c7f29de28f26a491744a5b3f2d58487f6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sebastian=20=C5=81akomy?= Date: Mon, 10 Aug 2026 10:25:02 +0200 Subject: [PATCH 09/13] test(json): structure scenarios as BDD Separate test setup, execution, and assertions consistently so each JSON behavior is easier to scan and reason about. Co-authored-by: Cursor --- .../softwaremill/jox/json/JsonFlowTest.java | 111 +++++++++++------- 1 file changed, 68 insertions(+), 43 deletions(-) diff --git a/json/src/test/java/com/softwaremill/jox/json/JsonFlowTest.java b/json/src/test/java/com/softwaremill/jox/json/JsonFlowTest.java index a2509d3..a53ed88 100644 --- a/json/src/test/java/com/softwaremill/jox/json/JsonFlowTest.java +++ b/json/src/test/java/com/softwaremill/jox/json/JsonFlowTest.java @@ -56,8 +56,17 @@ void shouldParseNdjsonLineEndingsBlankLinesAndFinalUnterminatedRecord() throws E @Test void shouldParseEmptyNdjsonAndEmptyArray() throws Exception { - assertEquals(List.of(), JsonFlow.parseNdjson(byteFlow(""), Person.class).runToList()); - assertEquals(List.of(), JsonFlow.parseArray(byteFlow("[]"), Person.class).runToList()); + // given + var ndjsonInput = byteFlow(""); + var arrayInput = byteFlow("[]"); + + // when + var ndjson = JsonFlow.parseNdjson(ndjsonInput, Person.class).runToList(); + var array = JsonFlow.parseArray(arrayInput, Person.class).runToList(); + + // then + assertEquals(List.of(), ndjson); + assertEquals(List.of(), array); } @Test @@ -140,6 +149,9 @@ void shouldParseJsonNodesUsingConfiguredReaderOverloads() throws Exception { @Test void shouldRejectDeserializedNullValuesButAllowJsonNullNodes() throws Exception { + // given + var nullNode = MAPPER.readTree("null"); + // when var ndjsonException = assertThrows( @@ -152,70 +164,74 @@ void shouldRejectDeserializedNullValuesButAllowJsonNullNodes() throws Exception JsonFlow.parseArray(byteFlow("[null]"), String.class) .buffer() .runToList()); - var nullNode = MAPPER.readTree("null"); + var ndjsonNodes = JsonFlow.parseNdjson(byteFlow("null\n"), JsonNode.class).runToList(); + var arrayNodes = JsonFlow.parseArray(byteFlow("[null]"), JsonNode.class).runToList(); // then assertCauseMessage(ndjsonException, "Jox flows do not support null values"); assertCauseMessage(arrayException, "Jox flows do not support null values"); - assertEquals( - List.of(nullNode), - JsonFlow.parseNdjson(byteFlow("null\n"), JsonNode.class).runToList()); - assertEquals( - List.of(nullNode), - JsonFlow.parseArray(byteFlow("[null]"), JsonNode.class).runToList()); + assertEquals(List.of(nullNode), ndjsonNodes); + assertEquals(List.of(nullNode), arrayNodes); } @Test void shouldRejectMalformedNdjsonAndMultipleValuesOnOneLine() { - assertFails( - () -> JsonFlow.parseNdjson(byteFlow("{\"name\":}\n"), Person.class).runToList()); - assertFails( - () -> - JsonFlow.parseNdjson( - byteFlow("{\"name\":\"Ada\",\"age\":36} true\n"), - Person.class) - .runToList()); + // given + var malformed = byteFlow("{\"name\":}\n"); + var multipleValues = byteFlow("{\"name\":\"Ada\",\"age\":36} true\n"); + + // when & then + assertFails(() -> JsonFlow.parseNdjson(malformed, Person.class).runToList()); + assertFails(() -> JsonFlow.parseNdjson(multipleValues, Person.class).runToList()); } @Test void shouldHandleArrayWhitespaceAndRejectMissingInputAndTrailingCommas() throws Exception { - assertEquals( - List.of(1), - JsonFlow.parseArray(byteFlow(" \n\t[ 1 ]\r\n "), Integer.class).runToList()); + // given + var inputWithWhitespace = byteFlow(" \n\t[ 1 ]\r\n "); + var emptyInput = byteFlow(""); + var whitespaceOnlyInput = byteFlow(" \r\n\t"); + var trailingCommaInput = byteFlow("[1,]"); + // when + var result = JsonFlow.parseArray(inputWithWhitespace, Integer.class).runToList(); var empty = assertThrows( Exception.class, - () -> JsonFlow.parseArray(byteFlow(""), Integer.class).runToList()); + () -> JsonFlow.parseArray(emptyInput, Integer.class).runToList()); var whitespaceOnly = assertThrows( Exception.class, - () -> JsonFlow.parseArray(byteFlow(" \r\n\t"), Integer.class).runToList()); + () -> JsonFlow.parseArray(whitespaceOnlyInput, Integer.class).runToList()); + // then + assertEquals(List.of(1), result); assertCauseMessage(empty, "Expected one top-level JSON array"); assertCauseMessage(whitespaceOnly, "Expected one top-level JSON array"); - assertFails(() -> JsonFlow.parseArray(byteFlow("[1,]"), Integer.class).runToList()); + assertFails(() -> JsonFlow.parseArray(trailingCommaInput, Integer.class).runToList()); } @Test void shouldRejectMalformedArrayWrongTopLevelShapeAndTrailingContent() { - assertFails( - () -> - JsonFlow.parseArray(byteFlow("[{\"name\":\"Ada\"}"), Person.class) - .runToList()); + // given + var incomplete = byteFlow("[{\"name\":\"Ada\"}"); + var wrongShapeInput = byteFlow("{\"name\":\"Ada\"}"); + var trailingInput = byteFlow("[] true"); + + // when + assertFails(() -> JsonFlow.parseArray(incomplete, Person.class).runToList()); var wrongShape = assertThrows( Exception.class, - () -> - JsonFlow.parseArray(byteFlow("{\"name\":\"Ada\"}"), Person.class) - .runToList()); - assertCauseMessage(wrongShape, "Expected one top-level JSON array"); - + () -> JsonFlow.parseArray(wrongShapeInput, Person.class).runToList()); var trailing = assertThrows( Exception.class, - () -> JsonFlow.parseArray(byteFlow("[] true"), Person.class).runToList()); + () -> JsonFlow.parseArray(trailingInput, Person.class).runToList()); + + // then + assertCauseMessage(wrongShape, "Expected one top-level JSON array"); assertCauseMessage(trailing, "Unexpected content after the top-level JSON array"); } @@ -425,8 +441,16 @@ void shouldAllowEscapedLineBreaksInNdjsonValues() throws Exception { @Test void shouldRenderEmptyFlows() throws Exception { - assertEquals("", render(JsonFlow.renderNdjson(Flows.empty(), Person.class))); - assertEquals("[]", render(JsonFlow.renderArray(Flows.empty(), Person.class))); + // given + Flow empty = Flows.empty(); + + // when + var ndjson = render(JsonFlow.renderNdjson(empty, Person.class)); + var array = render(JsonFlow.renderArray(empty, Person.class)); + + // then + assertEquals("", ndjson); + assertEquals("[]", array); } @Test @@ -525,29 +549,30 @@ void shouldPropagateRenderingUpstreamErrors() { @Test void shouldPropagateJacksonReaderAndWriterFailures() { + // given + var ndjsonInput = byteFlow("{\"value\":\"x\"}\n"); + var arrayInput = byteFlow("[{\"value\":\"x\"}]"); + var failingValue = new FailingSerialization(); + // when var ndjsonReaderException = assertThrows( Exception.class, () -> - JsonFlow.parseNdjson( - byteFlow("{\"value\":\"x\"}\n"), - FailingDeserialization.class) + JsonFlow.parseNdjson(ndjsonInput, FailingDeserialization.class) .runToList()); var arrayReaderException = assertThrows( Exception.class, () -> - JsonFlow.parseArray( - byteFlow("[{\"value\":\"x\"}]"), - FailingDeserialization.class) + JsonFlow.parseArray(arrayInput, FailingDeserialization.class) .runToList()); var ndjsonWriterException = assertThrows( Exception.class, () -> JsonFlow.renderNdjson( - Flows.fromValues(new FailingSerialization()), + Flows.fromValues(failingValue), FailingSerialization.class) .runToList()); var arrayWriterException = @@ -555,7 +580,7 @@ void shouldPropagateJacksonReaderAndWriterFailures() { Exception.class, () -> JsonFlow.renderArray( - Flows.fromValues(new FailingSerialization()), + Flows.fromValues(failingValue), FailingSerialization.class) .runToList()); From 81f9e4dc07dd6a49d4bed66830dbd1ca807e3f55 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sebastian=20=C5=81akomy?= Date: Mon, 10 Aug 2026 10:25:06 +0200 Subject: [PATCH 10/13] docs: remove time-sensitive version wording Keep Java requirements and the structured-concurrency comparison accurate without relying on a particular publication date. Co-authored-by: Cursor --- docs/flows.md | 2 +- docs/json.md | 2 +- docs/structured.md | 6 +++--- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/docs/flows.md b/docs/flows.md index 039c5f0..272c7dd 100644 --- a/docs/flows.md +++ b/docs/flows.md @@ -3,7 +3,7 @@ Finite & infinite streaming using flows, with reactive streams compatibility, (blocking) I/O integration, and a high-level, "functional" API. -Requires Java 25 (current LTS). +Requires Java 25. Javadocs: [https://javadoc.io](https://javadoc.io/doc/com.softwaremill.jox/flows). diff --git a/docs/json.md b/docs/json.md index d511412..115ca75 100644 --- a/docs/json.md +++ b/docs/json.md @@ -3,7 +3,7 @@ Lazy, backpressured parsing and rendering of newline-delimited JSON (NDJSON) and top-level JSON arrays using Jox `Flow` and `ByteFlow`. -Requires Java 25 (current LTS). +Requires Java 25. ## Dependency diff --git a/docs/structured.md b/docs/structured.md index a363f77..1a8632f 100644 --- a/docs/structured.md +++ b/docs/structured.md @@ -3,7 +3,7 @@ Programmer-friendly structured concurrency scopes, building upon the lower-level API available as a preview in Java 25, [JEP 505](https://openjdk.org/jeps/505). -Requires Java 25 (current LTS). +Requires Java 25. Javadocs: [https://javadoc.io](https://javadoc.io/doc/com.softwaremill.jox/structured). @@ -183,8 +183,8 @@ void main(String[] args) throws InterruptedException, TimeoutException { ## Comparing with Java's structured concurrency (JEP 505) -Java 21 and further releases include previews of a structured concurrency API. The latest version of the proposal is in -[JEP 505](https://openjdk.org/jeps/505). How does it compare with Jox's structured concurrency? +[JEP 505](https://openjdk.org/jeps/505) describes a preview of Java's structured concurrency API. How does it compare +with Jox's structured concurrency? Let's examine a simple example of parallelizing two computations, first using JEP 505: From c4de2cba7c1c1e7ed5ac2ec9d082738d67c5631d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sebastian=20=C5=81akomy?= Date: Mon, 10 Aug 2026 17:56:40 +0200 Subject: [PATCH 11/13] fix(json): harden streaming parsing and rendering --- README.md | 1 + docs/json.md | 28 +- .../com/softwaremill/jox/json/JsonFlow.java | 119 ++++++-- .../softwaremill/jox/json/JsonParsing.java | 4 +- .../jox/json/JsonReadSettings.java | 35 +++ .../softwaremill/jox/json/JsonRendering.java | 15 +- .../softwaremill/jox/json/NdjsonFraming.java | 141 ++++++++++ .../softwaremill/jox/json/JsonFlowTest.java | 260 ++++++++++++++++-- 8 files changed, 548 insertions(+), 55 deletions(-) create mode 100644 json/src/main/java/com/softwaremill/jox/json/JsonReadSettings.java create mode 100644 json/src/main/java/com/softwaremill/jox/json/NdjsonFraming.java diff --git a/README.md b/README.md index d5dab4e..03b285a 100644 --- a/README.md +++ b/README.md @@ -14,6 +14,7 @@ Includes: * 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) Find out more in the documentation available at [jox.softwaremill.com](https://jox.softwaremill.com/). diff --git a/docs/json.md b/docs/json.md index 115ca75..e9d4b01 100644 --- a/docs/json.md +++ b/docs/json.md @@ -41,8 +41,9 @@ implementation("com.softwaremill.jox:json:0.1.0") Each method is lazy: parsing, rendering and I/O start only when the returned flow is run. Values are processed one at a time, preserving the backpressure, failure propagation and cancellation behavior of the underlying Jox flow. -Parsing rejects top-level values that Jackson deserializes as Java `null`, as Jox flows do not support `null` elements. -To retain JSON `null` values, deserialize into Jackson's tree model, where they are represented by non-null null nodes. +Parsing rejects an NDJSON record or array element that Jackson deserializes as Java `null`, and rendering rejects raw +Java `null` elements, as Jox flows do not support them. To retain JSON `null` values, use Jackson's tree model, where +they are represented by non-null null nodes. Each operation has overloads accepting a `Class`, a Jackson `TypeReference`, or a configured Jackson `ObjectReader`/`ObjectWriter`. The `Class` and `TypeReference` overloads use the module's default `ObjectMapper`. @@ -51,7 +52,14 @@ Each operation has overloads accepting a `Class`, a Jackson `TypeReference NDJSON parsing accepts LF and CRLF line endings, ignores empty and whitespace-only lines, and accepts a final record without a line ending. Every non-blank line must contain exactly one JSON value; malformed JSON or trailing content on -a record fails the flow when it is run. +a record fails the flow when it is run. Input must be valid UTF-8. One UTF-8 byte-order mark is accepted at the very +beginning of the stream. + +An incomplete record is buffered across source chunks until its LF delimiter, or until end-of-input for the final +unterminated record. The default maximum encoded record size is 32 MiB, excluding the LF delimiter. An initial BOM and +the CR in a CRLF line ending count toward the limit. Use +`JsonReadSettings.defaults().maxNdjsonRecordBytes(...)` to choose another positive byte limit and pass the resulting +settings as the final argument to `parseNdjson`. ```java import java.nio.charset.StandardCharsets; @@ -104,9 +112,11 @@ void main() throws Exception { ## JSON arrays -Array parsing requires exactly one complete top-level array and emits every element as soon as it is decoded. A +Array parsing requires exactly one complete top-level array and incrementally emits its elements while parsing. A different top-level JSON value, an incomplete array, malformed input, or JSON content after the array fails the flow. -An empty array produces an empty flow. +An empty array produces an empty flow. Elements can be emitted before the source ends, but successful completion waits +for end-of-input so that trailing content can be rejected. As array parsing uses an internal supervised scope, failures +are wrapped in `JoxScopeExecutionException`. ```java import java.nio.charset.StandardCharsets; @@ -134,7 +144,8 @@ void main() throws Exception { ``` Array rendering writes `[` and `]` around comma-separated values. Elements are serialized one at a time, and an empty -input flow produces `[]`. +input flow produces `[]`. If serialization or the input flow fails after output starts, already-written bytes can +contain an incomplete array; callers should discard failed output or write transactionally when this matters. ```java import java.nio.file.Path; @@ -178,6 +189,11 @@ For custom Jackson modules, naming strategies, date handling, polymorphism, tree features, configure an `ObjectMapper` and derive an `ObjectReader` or `ObjectWriter`. The reader or writer determines the type and Jackson behavior for each NDJSON record or array element. +The parsing mode controls trailing-token validation: NDJSON enables `FAIL_ON_TRAILING_TOKENS` so that every record +contains exactly one value, while array parsing disables it when reading individual elements. These settings override +the supplied reader's value for that feature. The generic result type of an `ObjectReader` overload is inferred by Java +and cannot be checked against the reader's configured type, so callers must keep them consistent. + ```java import com.softwaremill.jox.flows.Flow; import com.softwaremill.jox.flows.Flow.ByteFlow; diff --git a/json/src/main/java/com/softwaremill/jox/json/JsonFlow.java b/json/src/main/java/com/softwaremill/jox/json/JsonFlow.java index f227489..b541bf6 100644 --- a/json/src/main/java/com/softwaremill/jox/json/JsonFlow.java +++ b/json/src/main/java/com/softwaremill/jox/json/JsonFlow.java @@ -4,6 +4,7 @@ import com.softwaremill.jox.flows.Flow; import com.softwaremill.jox.flows.Flow.ByteFlow; +import com.softwaremill.jox.structured.JoxScopeExecutionException; import tools.jackson.core.type.TypeReference; import tools.jackson.databind.ObjectMapper; @@ -15,8 +16,9 @@ * *

All transformations are lazy and preserve the backpressure and cancellation behavior of the * supplied flow. Values are parsed or rendered one at a time. Parsing fails when Jackson - * deserializes a top-level value as {@code null}, which Jox flows do not support. Use Jackson's - * tree model to represent a JSON {@code null} as a non-null node. + * deserializes an NDJSON record or array element as {@code null}, and rendering fails on a raw Java + * {@code null}, as Jox flows do not support null elements. Use Jackson's tree model to represent a + * JSON {@code null} as a non-null node. */ public final class JsonFlow { @@ -27,7 +29,8 @@ private JsonFlow() {} /** * Parses newline-delimited JSON using a default {@link ObjectMapper}. Empty and whitespace-only * lines are ignored. Both LF and CRLF line endings are accepted, as is a final record without a - * line ending. + * line ending. The input must be valid UTF-8; one initial UTF-8 byte-order mark is accepted. + * Records are limited to 32 MiB by default. * * @param bytes the UTF-8 encoded NDJSON * @param valueType the type of each parsed value @@ -35,14 +38,33 @@ private JsonFlow() {} * @return a flow emitting one value for each non-blank input line */ public static Flow parseNdjson(ByteFlow bytes, Class valueType) { + return parseNdjson(bytes, valueType, JsonReadSettings.defaults()); + } + + /** + * Parses newline-delimited JSON using a default {@link ObjectMapper}. Empty and whitespace-only + * lines are ignored. Both LF and CRLF line endings are accepted, as is a final record without a + * line ending. The input must be valid UTF-8; one initial UTF-8 byte-order mark is accepted. + * + * @param bytes the UTF-8 encoded NDJSON + * @param valueType the type of each parsed value + * @param settings the NDJSON framing settings + * @param the type of parsed values + * @return a flow emitting one value for each non-blank input line + */ + public static Flow parseNdjson( + ByteFlow bytes, Class valueType, JsonReadSettings settings) { return parseNdjson( - bytes, DEFAULT_MAPPER.readerFor(Objects.requireNonNull(valueType, "valueType"))); + bytes, + DEFAULT_MAPPER.readerFor(Objects.requireNonNull(valueType, "valueType")), + settings); } /** * Parses newline-delimited JSON using a default {@link ObjectMapper}. Empty and whitespace-only * lines are ignored. Both LF and CRLF line endings are accepted, as is a final record without a - * line ending. + * line ending. The input must be valid UTF-8; one initial UTF-8 byte-order mark is accepted. + * Records are limited to 32 MiB by default. * * @param bytes the UTF-8 encoded NDJSON * @param valueType the generic type of each parsed value @@ -50,14 +72,36 @@ public static Flow parseNdjson(ByteFlow bytes, Class valueType) { * @return a flow emitting one value for each non-blank input line */ public static Flow parseNdjson(ByteFlow bytes, TypeReference valueType) { + return parseNdjson(bytes, valueType, JsonReadSettings.defaults()); + } + + /** + * Parses newline-delimited JSON using a default {@link ObjectMapper}. Empty and whitespace-only + * lines are ignored. Both LF and CRLF line endings are accepted, as is a final record without a + * line ending. The input must be valid UTF-8; one initial UTF-8 byte-order mark is accepted. + * + * @param bytes the UTF-8 encoded NDJSON + * @param valueType the generic type of each parsed value + * @param settings the NDJSON framing settings + * @param the type of parsed values + * @return a flow emitting one value for each non-blank input line + */ + public static Flow parseNdjson( + ByteFlow bytes, TypeReference valueType, JsonReadSettings settings) { return parseNdjson( - bytes, DEFAULT_MAPPER.readerFor(Objects.requireNonNull(valueType, "valueType"))); + bytes, + DEFAULT_MAPPER.readerFor(Objects.requireNonNull(valueType, "valueType")), + settings); } /** * Parses newline-delimited JSON using the supplied Jackson reader. Empty and whitespace-only * lines are ignored. Both LF and CRLF line endings are accepted, as is a final record without a - * line ending. Each non-blank line must contain exactly one JSON value. + * line ending. Each non-blank line must contain exactly one JSON value. The input must be valid + * UTF-8; one initial UTF-8 byte-order mark is accepted. {@link + * tools.jackson.databind.DeserializationFeature#FAIL_ON_TRAILING_TOKENS} is enabled regardless + * of the reader's configuration. The caller must ensure that {@code T} matches the type + * configured on the reader. Records are limited to 32 MiB by default. * * @param bytes the UTF-8 encoded NDJSON * @param reader the reader used to deserialize each value @@ -65,14 +109,38 @@ public static Flow parseNdjson(ByteFlow bytes, TypeReference valueType * @return a flow emitting one value for each non-blank input line */ public static Flow parseNdjson(ByteFlow bytes, ObjectReader reader) { + return parseNdjson(bytes, reader, JsonReadSettings.defaults()); + } + + /** + * Parses newline-delimited JSON using the supplied Jackson reader. Empty and whitespace-only + * lines are ignored. Both LF and CRLF line endings are accepted, as is a final record without a + * line ending. Each non-blank line must contain exactly one JSON value. The input must be valid + * UTF-8; one initial UTF-8 byte-order mark is accepted. {@link + * tools.jackson.databind.DeserializationFeature#FAIL_ON_TRAILING_TOKENS} is enabled regardless + * of the reader's configuration. The caller must ensure that {@code T} matches the type + * configured on the reader. + * + * @param bytes the UTF-8 encoded NDJSON + * @param reader the reader used to deserialize each value + * @param settings the NDJSON framing settings + * @param the type of parsed values + * @return a flow emitting one value for each non-blank input line + */ + public static Flow parseNdjson( + ByteFlow bytes, ObjectReader reader, JsonReadSettings settings) { return JsonParsing.parseNdjson( - Objects.requireNonNull(bytes, "bytes"), Objects.requireNonNull(reader, "reader")); + Objects.requireNonNull(bytes, "bytes"), + Objects.requireNonNull(reader, "reader"), + Objects.requireNonNull(settings, "settings")); } /** - * Parses a top-level JSON array using a default {@link ObjectMapper}. The returned flow emits - * each array element as soon as it is available. Input that is not one complete top-level - * array, including input containing trailing JSON, fails the flow. + * Parses a top-level JSON array using a default {@link ObjectMapper}. The returned flow + * incrementally emits array elements as they are parsed. Input that is not one complete + * top-level array, including input containing trailing JSON, fails the flow. Elements can be + * emitted before the input ends, but successful completion waits for end-of-input so that + * trailing content can be rejected. Failures are wrapped in {@link JoxScopeExecutionException}. * * @param bytes the UTF-8 encoded JSON array * @param valueType the type of each array element @@ -85,9 +153,11 @@ public static Flow parseArray(ByteFlow bytes, Class valueType) { } /** - * Parses a top-level JSON array using a default {@link ObjectMapper}. The returned flow emits - * each array element as soon as it is available. Input that is not one complete top-level - * array, including input containing trailing JSON, fails the flow. + * Parses a top-level JSON array using a default {@link ObjectMapper}. The returned flow + * incrementally emits array elements as they are parsed. Input that is not one complete + * top-level array, including input containing trailing JSON, fails the flow. Elements can be + * emitted before the input ends, but successful completion waits for end-of-input so that + * trailing content can be rejected. Failures are wrapped in {@link JoxScopeExecutionException}. * * @param bytes the UTF-8 encoded JSON array * @param valueType the generic type of each array element @@ -100,9 +170,15 @@ public static Flow parseArray(ByteFlow bytes, TypeReference valueType) } /** - * Parses a top-level JSON array using the supplied Jackson reader. The returned flow emits each - * array element as soon as it is available. Input that is not one complete top-level array, - * including input containing trailing JSON, fails the flow. + * Parses a top-level JSON array using the supplied Jackson reader. The returned flow + * incrementally emits array elements as they are parsed. Input that is not one complete + * top-level array, including input containing trailing JSON, fails the flow. Elements can be + * emitted before the input ends, but successful completion waits for end-of-input so that + * trailing content can be rejected. {@link + * tools.jackson.databind.DeserializationFeature#FAIL_ON_TRAILING_TOKENS} is disabled while + * reading individual elements, regardless of the reader's configuration. The caller must ensure + * that {@code T} matches the type configured on the reader. Failures are wrapped in {@link + * JoxScopeExecutionException}. * * @param bytes the UTF-8 encoded JSON array * @param reader the reader used to deserialize each array element @@ -159,7 +235,8 @@ public static ByteFlow renderNdjson(Flow values, ObjectWriter writer) { /** * Renders values as one JSON array using a default {@link ObjectMapper}. Elements are - * serialized one at a time. An empty input flow produces {@code []}. + * serialized one at a time. An empty input flow produces {@code []}. If the flow fails after + * output starts, already-emitted bytes can contain an incomplete array. * * @param values the values to render * @param valueType the type of each array element @@ -173,7 +250,8 @@ public static ByteFlow renderArray(Flow values, Class valueType) { /** * Renders values as one JSON array using a default {@link ObjectMapper}. Elements are - * serialized one at a time. An empty input flow produces {@code []}. + * serialized one at a time. An empty input flow produces {@code []}. If the flow fails after + * output starts, already-emitted bytes can contain an incomplete array. * * @param values the values to render * @param valueType the generic type of each array element @@ -187,7 +265,8 @@ public static ByteFlow renderArray(Flow values, TypeReference valueTyp /** * Renders values as one JSON array using the supplied Jackson writer. Elements are serialized - * one at a time. An empty input flow produces {@code []}. + * one at a time. An empty input flow produces {@code []}. If the flow fails after output + * starts, already-emitted bytes can contain an incomplete array. * * @param values the values to render * @param writer the writer used to serialize each array element diff --git a/json/src/main/java/com/softwaremill/jox/json/JsonParsing.java b/json/src/main/java/com/softwaremill/jox/json/JsonParsing.java index 393f875..d907bea 100644 --- a/json/src/main/java/com/softwaremill/jox/json/JsonParsing.java +++ b/json/src/main/java/com/softwaremill/jox/json/JsonParsing.java @@ -15,9 +15,9 @@ final class JsonParsing { private JsonParsing() {} - static Flow parseNdjson(ByteFlow bytes, ObjectReader reader) { + static Flow parseNdjson(ByteFlow bytes, ObjectReader reader, JsonReadSettings settings) { var singleValueReader = reader.with(DeserializationFeature.FAIL_ON_TRAILING_TOKENS); - return bytes.linesUtf8() + return NdjsonFraming.lines(bytes, settings.maxNdjsonRecordBytes()) .filter(line -> !line.isBlank()) .map(line -> requireNonNullValue(singleValueReader.readValue(line))); } diff --git a/json/src/main/java/com/softwaremill/jox/json/JsonReadSettings.java b/json/src/main/java/com/softwaremill/jox/json/JsonReadSettings.java new file mode 100644 index 0000000..a323369 --- /dev/null +++ b/json/src/main/java/com/softwaremill/jox/json/JsonReadSettings.java @@ -0,0 +1,35 @@ +package com.softwaremill.jox.json; + +/** + * Settings used when parsing NDJSON flows. + * + * @param maxNdjsonRecordBytes maximum UTF-8 encoded size of one NDJSON record, excluding the LF + * delimiter; must be positive + */ +public record JsonReadSettings(int maxNdjsonRecordBytes) { + + private static final int DEFAULT_MAX_NDJSON_RECORD_BYTES = 32 * 1024 * 1024; + + /** + * @throws IllegalArgumentException if {@code maxNdjsonRecordBytes} is not positive + */ + public JsonReadSettings { + if (maxNdjsonRecordBytes <= 0) { + throw new IllegalArgumentException("maxNdjsonRecordBytes must be greater than zero"); + } + } + + /** Returns settings with a 32 MiB maximum encoded NDJSON record size. */ + public static JsonReadSettings defaults() { + return new JsonReadSettings(DEFAULT_MAX_NDJSON_RECORD_BYTES); + } + + /** + * Returns a copy with the given maximum UTF-8 encoded NDJSON record size. + * + * @throws IllegalArgumentException if {@code newMaxNdjsonRecordBytes} is not positive + */ + public JsonReadSettings maxNdjsonRecordBytes(int newMaxNdjsonRecordBytes) { + return new JsonReadSettings(newMaxNdjsonRecordBytes); + } +} diff --git a/json/src/main/java/com/softwaremill/jox/json/JsonRendering.java b/json/src/main/java/com/softwaremill/jox/json/JsonRendering.java index 672d9f9..68fbeb9 100644 --- a/json/src/main/java/com/softwaremill/jox/json/JsonRendering.java +++ b/json/src/main/java/com/softwaremill/jox/json/JsonRendering.java @@ -16,18 +16,29 @@ final class JsonRendering { private JsonRendering() {} static ByteFlow renderNdjson(Flow values, ObjectWriter writer) { - return values.map(writer::writeValueAsBytes) + return values.map(value -> writer.writeValueAsBytes(requireNonNullValue(value))) .tap(JsonRendering::requireNoLineBreaks) .map(json -> ByteChunk.fromArray(json).concat(NEW_LINE)) .toByteFlow(); } static ByteFlow renderArray(Flow values, ObjectWriter writer) { - return values.map(value -> ByteChunk.fromArray(writer.writeValueAsBytes(value))) + return values.map( + value -> + ByteChunk.fromArray( + writer.writeValueAsBytes(requireNonNullValue(value)))) .intersperse(ARRAY_START, COMMA, ARRAY_END) .toByteFlow(); } + private static T requireNonNullValue(T value) { + if (value == null) { + throw new IllegalArgumentException( + "Java null cannot be rendered because Jox flows do not support null values"); + } + return value; + } + private static void requireNoLineBreaks(byte[] json) { for (byte value : json) { if (value == '\r' || value == '\n') { diff --git a/json/src/main/java/com/softwaremill/jox/json/NdjsonFraming.java b/json/src/main/java/com/softwaremill/jox/json/NdjsonFraming.java new file mode 100644 index 0000000..24fa7c4 --- /dev/null +++ b/json/src/main/java/com/softwaremill/jox/json/NdjsonFraming.java @@ -0,0 +1,141 @@ +package com.softwaremill.jox.json; + +import java.io.ByteArrayOutputStream; +import java.nio.ByteBuffer; +import java.nio.charset.CharacterCodingException; +import java.nio.charset.CodingErrorAction; +import java.nio.charset.StandardCharsets; +import java.util.Iterator; +import java.util.List; +import java.util.NoSuchElementException; +import java.util.Optional; + +import com.softwaremill.jox.flows.ByteChunk; +import com.softwaremill.jox.flows.Flow; +import com.softwaremill.jox.flows.Flow.ByteFlow; + +final class NdjsonFraming { + + private static final byte[] UTF_8_BOM = {(byte) 0xef, (byte) 0xbb, (byte) 0xbf}; + + private NdjsonFraming() {} + + static Flow lines(ByteFlow bytes, int maxRecordBytes) { + return bytes.mapWithResource( + () -> new State(maxRecordBytes), NdjsonFraming::finish, State::records) + .mapConcat(records -> records) + .map(NdjsonFraming::decode); + } + + private static Optional> finish(State state) { + return state.finish().map(record -> List.of(record)); + } + + private static String decode(Record record) { + var bytes = record.bytes(); + var offset = record.first() && startsWithBom(bytes) ? UTF_8_BOM.length : 0; + var decoder = + StandardCharsets.UTF_8 + .newDecoder() + .onMalformedInput(CodingErrorAction.REPORT) + .onUnmappableCharacter(CodingErrorAction.REPORT); + try { + return decoder.decode(ByteBuffer.wrap(bytes, offset, bytes.length - offset)).toString(); + } catch (CharacterCodingException e) { + throw new IllegalArgumentException("NDJSON input contains malformed UTF-8", e); + } + } + + private static boolean startsWithBom(byte[] bytes) { + if (bytes.length < UTF_8_BOM.length) { + return false; + } + for (int i = 0; i < UTF_8_BOM.length; i++) { + if (bytes[i] != UTF_8_BOM[i]) { + return false; + } + } + return true; + } + + private record Record(byte[] bytes, boolean first) {} + + private static final class State { + private final int maxRecordBytes; + private final ByteArrayOutputStream buffer = new ByteArrayOutputStream(); + private boolean firstRecord = true; + + private State(int maxRecordBytes) { + this.maxRecordBytes = maxRecordBytes; + } + + private Iterable records(ByteChunk chunk) { + return () -> + new Iterator<>() { + private final List arrays = chunk.getArrays(); + private int arrayIndex; + private int offset; + private Record next; + + @Override + public boolean hasNext() { + if (next != null) { + return true; + } + + while (arrayIndex < arrays.size()) { + var array = arrays.get(arrayIndex); + for (int i = offset; i < array.length; i++) { + if (array[i] == '\n') { + append(array, offset, i - offset); + next = completeRecord(); + offset = i + 1; + if (offset == array.length) { + arrayIndex++; + offset = 0; + } + return true; + } + } + + append(array, offset, array.length - offset); + arrayIndex++; + offset = 0; + } + return false; + } + + @Override + public Record next() { + if (!hasNext()) { + throw new NoSuchElementException(); + } + var result = next; + next = null; + return result; + } + }; + } + + private Optional finish() { + return buffer.size() == 0 ? Optional.empty() : Optional.of(completeRecord()); + } + + 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 " + + maxRecordBytes + + " bytes"); + } + buffer.write(bytes, offset, length); + } + + private Record completeRecord() { + var record = new Record(buffer.toByteArray(), firstRecord); + buffer.reset(); + firstRecord = false; + return record; + } + } +} diff --git a/json/src/test/java/com/softwaremill/jox/json/JsonFlowTest.java b/json/src/test/java/com/softwaremill/jox/json/JsonFlowTest.java index a53ed88..b134a9a 100644 --- a/json/src/test/java/com/softwaremill/jox/json/JsonFlowTest.java +++ b/json/src/test/java/com/softwaremill/jox/json/JsonFlowTest.java @@ -10,6 +10,7 @@ import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Path; +import java.util.ArrayList; import java.util.List; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -21,6 +22,7 @@ import com.softwaremill.jox.flows.Flow; import com.softwaremill.jox.flows.Flows; +import tools.jackson.core.JacksonException; import tools.jackson.core.type.TypeReference; import tools.jackson.databind.JsonNode; import tools.jackson.databind.ObjectMapper; @@ -80,6 +82,147 @@ void shouldParseNdjsonAcrossEveryByteBoundaryIncludingUtf8() throws Exception { JsonFlow.parseNdjson(input, Person.class).runToList()); } + @Test + void shouldParseNdjsonWithSplitUtf8Bom() throws Exception { + // given + var input = + Flows.fromByteChunks( + ByteChunk.fromArray(new byte[] {(byte) 0xef}), + ByteChunk.fromArray(new byte[] {(byte) 0xbb}), + ByteChunk.fromArray(new byte[] {(byte) 0xbf, '"', 'o', 'k', '"', '\n'})); + + // when & then + assertEquals(List.of("ok"), JsonFlow.parseNdjson(input, String.class).runToList()); + } + + @Test + void shouldRejectMalformedNdjsonUtf8() { + // given + var input = Flows.fromByteArrays(new byte[] {'"', (byte) 0xc3, '(', '"', '\n'}); + + // when + var exception = + assertThrows( + Exception.class, + () -> JsonFlow.parseNdjson(input, String.class).runToList()); + + // then + assertCauseTypeAndMessage( + exception, IllegalArgumentException.class, "NDJSON input contains malformed UTF-8"); + } + + @Test + void shouldApplyConfiguredNdjsonRecordLimitToEveryReaderOverload() throws Exception { + // given + var settings = JsonReadSettings.defaults().maxNdjsonRecordBytes(3); + TypeReference type = new TypeReference<>() {}; + var reader = MAPPER.readerFor(Integer.class); + + // when + var usingClass = + JsonFlow.parseNdjson(byteFlow("123\n456\n"), Integer.class, settings).runToList(); + var usingTypeReference = + JsonFlow.parseNdjson(byteFlow("123\n456\n"), type, settings).runToList(); + var usingReader = + JsonFlow.parseNdjson(byteFlow("123\n456\n"), reader, settings).runToList(); + var classException = + assertThrows( + Exception.class, + () -> + JsonFlow.parseNdjson(byteFlow("1234\n"), Integer.class, settings) + .runToList()); + var typeReferenceException = + assertThrows( + Exception.class, + () -> JsonFlow.parseNdjson(byteFlow("1234\n"), type, settings).runToList()); + var readerException = + assertThrows( + Exception.class, + () -> + JsonFlow.parseNdjson(byteFlow("1234\n"), reader, settings) + .runToList()); + + // then + assertEquals(List.of(123, 456), usingClass); + assertEquals(List.of(123, 456), usingTypeReference); + assertEquals(List.of(123, 456), usingReader); + assertRecordLimitExceeded(classException, 3); + assertRecordLimitExceeded(typeReferenceException, 3); + assertRecordLimitExceeded(readerException, 3); + } + + @Test + void shouldValidateNdjsonRecordLimitSettings() { + // when & then + assertEquals(32 * 1024 * 1024, JsonReadSettings.defaults().maxNdjsonRecordBytes()); + assertThrows(IllegalArgumentException.class, () -> new JsonReadSettings(0)); + } + + @Test + void shouldCountBomAndCarriageReturnTowardNdjsonRecordLimit() throws Exception { + // given + var fourBytes = JsonReadSettings.defaults().maxNdjsonRecordBytes(4); + var threeBytes = JsonReadSettings.defaults().maxNdjsonRecordBytes(3); + var twoBytes = JsonReadSettings.defaults().maxNdjsonRecordBytes(2); + + // when + var bomAtLimit = + JsonFlow.parseNdjson(byteFlow("\uFEFF1\n"), Integer.class, fourBytes).runToList(); + var bomOverLimit = + assertThrows( + Exception.class, + () -> + JsonFlow.parseNdjson( + byteFlow("\uFEFF1\n"), Integer.class, threeBytes) + .runToList()); + var crAtLimit = + JsonFlow.parseNdjson(byteFlow("12\r\n"), Integer.class, threeBytes).runToList(); + var crOverLimit = + assertThrows( + Exception.class, + () -> + JsonFlow.parseNdjson(byteFlow("12\r\n"), Integer.class, twoBytes) + .runToList()); + + // then + assertEquals(List.of(1), bomAtLimit); + assertRecordLimitExceeded(bomOverLimit, 3); + assertEquals(List.of(12), crAtLimit); + assertRecordLimitExceeded(crOverLimit, 2); + } + + @Test + void shouldEmitValidNdjsonRecordsBeforeLaterRecordInSameChunkFails() { + // given + var emitted = new ArrayList(); + var settings = JsonReadSettings.defaults().maxNdjsonRecordBytes(3); + + // when + var exception = + assertThrows( + Exception.class, + () -> + JsonFlow.parseNdjson(byteFlow("1\n1234\n"), Integer.class, settings) + .runForeach(emitted::add)); + + // then + assertEquals(List.of(1), emitted); + assertRecordLimitExceeded(exception, 3); + } + + @Test + void shouldPreserveNdjsonRecordsAcrossEmptyChunks() throws Exception { + // given + var input = + Flows.fromByteChunks( + ByteChunk.fromArray("12".getBytes(StandardCharsets.UTF_8)), + ByteChunk.empty(), + ByteChunk.fromArray("3\n".getBytes(StandardCharsets.UTF_8))); + + // when & then + assertEquals(List.of(123), JsonFlow.parseNdjson(input, Integer.class).runToList()); + } + @Test void shouldParseArrayAcrossEveryByteBoundaryIncludingUtf8() throws Exception { // given @@ -160,10 +303,7 @@ void shouldRejectDeserializedNullValuesButAllowJsonNullNodes() throws Exception var arrayException = assertThrows( Exception.class, - () -> - JsonFlow.parseArray(byteFlow("[null]"), String.class) - .buffer() - .runToList()); + () -> JsonFlow.parseArray(byteFlow("[null]"), String.class).runToList()); var ndjsonNodes = JsonFlow.parseNdjson(byteFlow("null\n"), JsonNode.class).runToList(); var arrayNodes = JsonFlow.parseArray(byteFlow("[null]"), JsonNode.class).runToList(); @@ -180,9 +320,19 @@ void shouldRejectMalformedNdjsonAndMultipleValuesOnOneLine() { var malformed = byteFlow("{\"name\":}\n"); var multipleValues = byteFlow("{\"name\":\"Ada\",\"age\":36} true\n"); - // when & then - assertFails(() -> JsonFlow.parseNdjson(malformed, Person.class).runToList()); - assertFails(() -> JsonFlow.parseNdjson(multipleValues, Person.class).runToList()); + // when + var malformedException = + assertThrows( + Exception.class, + () -> JsonFlow.parseNdjson(malformed, Person.class).runToList()); + var multipleValuesException = + assertThrows( + Exception.class, + () -> JsonFlow.parseNdjson(multipleValues, Person.class).runToList()); + + // then + assertCauseType(malformedException, JacksonException.class); + assertCauseType(multipleValuesException, JacksonException.class); } @Test @@ -208,19 +358,25 @@ void shouldHandleArrayWhitespaceAndRejectMissingInputAndTrailingCommas() throws assertEquals(List.of(1), result); assertCauseMessage(empty, "Expected one top-level JSON array"); assertCauseMessage(whitespaceOnly, "Expected one top-level JSON array"); - assertFails(() -> JsonFlow.parseArray(trailingCommaInput, Integer.class).runToList()); + var trailingComma = + assertThrows( + Exception.class, + () -> JsonFlow.parseArray(trailingCommaInput, Integer.class).runToList()); + assertCauseType(trailingComma, JacksonException.class); } @Test void shouldRejectMalformedArrayWrongTopLevelShapeAndTrailingContent() { // given - var incomplete = byteFlow("[{\"name\":\"Ada\"}"); + var incomplete = byteFlow("[1"); var wrongShapeInput = byteFlow("{\"name\":\"Ada\"}"); var trailingInput = byteFlow("[] true"); // when - assertFails(() -> JsonFlow.parseArray(incomplete, Person.class).runToList()); - + var incompleteException = + assertThrows( + Exception.class, + () -> JsonFlow.parseArray(incomplete, Integer.class).runToList()); var wrongShape = assertThrows( Exception.class, @@ -231,6 +387,7 @@ void shouldRejectMalformedArrayWrongTopLevelShapeAndTrailingContent() { () -> JsonFlow.parseArray(trailingInput, Person.class).runToList()); // then + assertCauseType(incompleteException, JacksonException.class); assertCauseMessage(wrongShape, "Expected one top-level JSON array"); assertCauseMessage(trailing, "Unexpected content after the top-level JSON array"); } @@ -398,10 +555,11 @@ void shouldStopNdjsonUpstreamAndPropagateDownstreamFailure() throws Exception { @Test void shouldRunParsingFlowsRepeatedly() throws Exception { // given - var ndjson = JsonFlow.parseNdjson(byteFlow("1\n2\n"), Integer.class); + var ndjson = JsonFlow.parseNdjson(oneByteChunks("\uFEFF1\n2"), Integer.class); var array = JsonFlow.parseArray(byteFlow("[1,2]"), Integer.class); // when & then + assertEquals(List.of(1), ndjson.take(1).runToList()); assertEquals(List.of(1, 2), ndjson.runToList()); assertEquals(List.of(1, 2), ndjson.runToList()); assertEquals(List.of(1, 2), array.runToList()); @@ -470,20 +628,48 @@ void shouldRenderGenericTypesUsingTypeReferenceOverloads() throws Exception { void shouldRenderJsonNodesUsingConfiguredWriterOverloads() throws Exception { // given var writer = MAPPER.writerFor(JsonNode.class); - var values = Flows.fromValues(MAPPER.readTree("{\"n\":1}"), MAPPER.readTree("[true,null]")); + var values = + Flows.fromValues( + MAPPER.readTree("{\"n\":1}"), + MAPPER.readTree("[true,null]"), + MAPPER.readTree("null")); // when & then - assertEquals("{\"n\":1}\n[true,null]\n", render(JsonFlow.renderNdjson(values, writer))); assertEquals( - "[{\"n\":1},[true,null]]", + "{\"n\":1}\n[true,null]\nnull\n", render(JsonFlow.renderNdjson(values, writer))); + assertEquals( + "[{\"n\":1},[true,null],null]", render( JsonFlow.renderArray( Flows.fromValues( MAPPER.readTree("{\"n\":1}"), - MAPPER.readTree("[true,null]")), + MAPPER.readTree("[true,null]"), + MAPPER.readTree("null")), writer))); } + @Test + void shouldRejectRawNullValuesWhenRendering() { + // given + var values = Flows.usingEmit(emit -> emit.apply(null)); + + // when + var ndjsonException = + assertThrows( + Exception.class, + () -> JsonFlow.renderNdjson(values, String.class).runToList()); + var arrayException = + assertThrows( + Exception.class, + () -> JsonFlow.renderArray(values, String.class).runToList()); + + // then + var expectedMessage = + "Java null cannot be rendered because Jox flows do not support null values"; + assertCauseTypeAndMessage(ndjsonException, IllegalArgumentException.class, expectedMessage); + assertCauseTypeAndMessage(arrayException, IllegalArgumentException.class, expectedMessage); + } + @Test void shouldRejectRawLineBreaksProducedByNdjsonWriter() { // given @@ -706,10 +892,6 @@ private static String chunksToString(List chunks) { return output.toString(StandardCharsets.UTF_8); } - private static void assertFails(ThrowingRunnable action) { - assertThrows(Exception.class, action::run); - } - private static void assertCauseMessage(Throwable exception, String expectedFragment) { for (Throwable current = exception; current != null; current = current.getCause()) { if (current.getMessage() != null && current.getMessage().contains(expectedFragment)) { @@ -720,6 +902,39 @@ private static void assertCauseMessage(Throwable exception, String expectedFragm "No exception in the cause chain contained: " + expectedFragment, exception); } + private static void assertCauseType( + Throwable exception, Class expectedType) { + for (Throwable current = exception; current != null; current = current.getCause()) { + if (expectedType.isInstance(current)) { + return; + } + } + throw new AssertionError( + "No exception in the cause chain had type: " + expectedType.getName(), exception); + } + + private static void assertCauseTypeAndMessage( + Throwable exception, Class expectedType, String expectedMessage) { + for (Throwable current = exception; current != null; current = current.getCause()) { + if (expectedType.isInstance(current) && expectedMessage.equals(current.getMessage())) { + return; + } + } + throw new AssertionError( + "No exception in the cause chain had type " + + expectedType.getName() + + " and message: " + + expectedMessage, + exception); + } + + private static void assertRecordLimitExceeded(Throwable exception, int maximumBytes) { + assertCauseTypeAndMessage( + exception, + IllegalArgumentException.class, + "NDJSON record exceeds the configured maximum of " + maximumBytes + " bytes"); + } + private static void assertHasCause(Throwable exception, Throwable expected) { for (Throwable current = exception; current != null; current = current.getCause()) { if (current == expected) { @@ -731,11 +946,6 @@ private static void assertHasCause(Throwable exception, Throwable expected) { "Expected exception was not present in the cause chain", exception); } - @FunctionalInterface - private interface ThrowingRunnable { - void run() throws Exception; - } - private record FailingDeserialization(String value) { private FailingDeserialization { throw DESERIALIZATION_FAILURE; From 783b7dc8ef69b955a928db999994375fa626dbf7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sebastian=20=C5=81akomy?= Date: Mon, 10 Aug 2026 18:17:19 +0200 Subject: [PATCH 12/13] docs(json): trim repetitive API documentation --- .../com/softwaremill/jox/json/JsonFlow.java | 154 +++--------------- .../jox/json/JsonReadSettings.java | 8 - 2 files changed, 19 insertions(+), 143 deletions(-) diff --git a/json/src/main/java/com/softwaremill/jox/json/JsonFlow.java b/json/src/main/java/com/softwaremill/jox/json/JsonFlow.java index b541bf6..683d43a 100644 --- a/json/src/main/java/com/softwaremill/jox/json/JsonFlow.java +++ b/json/src/main/java/com/softwaremill/jox/json/JsonFlow.java @@ -26,32 +26,12 @@ public final class JsonFlow { private JsonFlow() {} - /** - * Parses newline-delimited JSON using a default {@link ObjectMapper}. Empty and whitespace-only - * lines are ignored. Both LF and CRLF line endings are accepted, as is a final record without a - * line ending. The input must be valid UTF-8; one initial UTF-8 byte-order mark is accepted. - * Records are limited to 32 MiB by default. - * - * @param bytes the UTF-8 encoded NDJSON - * @param valueType the type of each parsed value - * @param the type of parsed values - * @return a flow emitting one value for each non-blank input line - */ + /** Parses NDJSON using the default mapper and settings. */ public static Flow parseNdjson(ByteFlow bytes, Class valueType) { return parseNdjson(bytes, valueType, JsonReadSettings.defaults()); } - /** - * Parses newline-delimited JSON using a default {@link ObjectMapper}. Empty and whitespace-only - * lines are ignored. Both LF and CRLF line endings are accepted, as is a final record without a - * line ending. The input must be valid UTF-8; one initial UTF-8 byte-order mark is accepted. - * - * @param bytes the UTF-8 encoded NDJSON - * @param valueType the type of each parsed value - * @param settings the NDJSON framing settings - * @param the type of parsed values - * @return a flow emitting one value for each non-blank input line - */ + /** Parses NDJSON using the default mapper and the supplied settings. */ public static Flow parseNdjson( ByteFlow bytes, Class valueType, JsonReadSettings settings) { return parseNdjson( @@ -60,32 +40,12 @@ public static Flow parseNdjson( settings); } - /** - * Parses newline-delimited JSON using a default {@link ObjectMapper}. Empty and whitespace-only - * lines are ignored. Both LF and CRLF line endings are accepted, as is a final record without a - * line ending. The input must be valid UTF-8; one initial UTF-8 byte-order mark is accepted. - * Records are limited to 32 MiB by default. - * - * @param bytes the UTF-8 encoded NDJSON - * @param valueType the generic type of each parsed value - * @param the type of parsed values - * @return a flow emitting one value for each non-blank input line - */ + /** Parses generic NDJSON values using the default mapper and settings. */ public static Flow parseNdjson(ByteFlow bytes, TypeReference valueType) { return parseNdjson(bytes, valueType, JsonReadSettings.defaults()); } - /** - * Parses newline-delimited JSON using a default {@link ObjectMapper}. Empty and whitespace-only - * lines are ignored. Both LF and CRLF line endings are accepted, as is a final record without a - * line ending. The input must be valid UTF-8; one initial UTF-8 byte-order mark is accepted. - * - * @param bytes the UTF-8 encoded NDJSON - * @param valueType the generic type of each parsed value - * @param settings the NDJSON framing settings - * @param the type of parsed values - * @return a flow emitting one value for each non-blank input line - */ + /** Parses generic NDJSON values using the default mapper and the supplied settings. */ public static Flow parseNdjson( ByteFlow bytes, TypeReference valueType, JsonReadSettings settings) { return parseNdjson( @@ -94,29 +54,14 @@ public static Flow parseNdjson( settings); } - /** - * Parses newline-delimited JSON using the supplied Jackson reader. Empty and whitespace-only - * lines are ignored. Both LF and CRLF line endings are accepted, as is a final record without a - * line ending. Each non-blank line must contain exactly one JSON value. The input must be valid - * UTF-8; one initial UTF-8 byte-order mark is accepted. {@link - * tools.jackson.databind.DeserializationFeature#FAIL_ON_TRAILING_TOKENS} is enabled regardless - * of the reader's configuration. The caller must ensure that {@code T} matches the type - * configured on the reader. Records are limited to 32 MiB by default. - * - * @param bytes the UTF-8 encoded NDJSON - * @param reader the reader used to deserialize each value - * @param the type of parsed values - * @return a flow emitting one value for each non-blank input line - */ + /** Parses NDJSON using the supplied reader and default settings. */ public static Flow parseNdjson(ByteFlow bytes, ObjectReader reader) { return parseNdjson(bytes, reader, JsonReadSettings.defaults()); } /** - * Parses newline-delimited JSON using the supplied Jackson reader. Empty and whitespace-only - * lines are ignored. Both LF and CRLF line endings are accepted, as is a final record without a - * line ending. Each non-blank line must contain exactly one JSON value. The input must be valid - * UTF-8; one initial UTF-8 byte-order mark is accepted. {@link + * Parses UTF-8 NDJSON using the supplied reader and framing settings. Blank lines are ignored; + * LF, CRLF, a final unterminated record and one initial byte-order mark are accepted. {@link * tools.jackson.databind.DeserializationFeature#FAIL_ON_TRAILING_TOKENS} is enabled regardless * of the reader's configuration. The caller must ensure that {@code T} matches the type * configured on the reader. @@ -135,46 +80,21 @@ public static Flow parseNdjson( Objects.requireNonNull(settings, "settings")); } - /** - * Parses a top-level JSON array using a default {@link ObjectMapper}. The returned flow - * incrementally emits array elements as they are parsed. Input that is not one complete - * top-level array, including input containing trailing JSON, fails the flow. Elements can be - * emitted before the input ends, but successful completion waits for end-of-input so that - * trailing content can be rejected. Failures are wrapped in {@link JoxScopeExecutionException}. - * - * @param bytes the UTF-8 encoded JSON array - * @param valueType the type of each array element - * @param the type of parsed values - * @return a flow emitting the array elements - */ + /** Parses a top-level JSON array using the default mapper. */ public static Flow parseArray(ByteFlow bytes, Class valueType) { return parseArray( bytes, DEFAULT_MAPPER.readerFor(Objects.requireNonNull(valueType, "valueType"))); } - /** - * Parses a top-level JSON array using a default {@link ObjectMapper}. The returned flow - * incrementally emits array elements as they are parsed. Input that is not one complete - * top-level array, including input containing trailing JSON, fails the flow. Elements can be - * emitted before the input ends, but successful completion waits for end-of-input so that - * trailing content can be rejected. Failures are wrapped in {@link JoxScopeExecutionException}. - * - * @param bytes the UTF-8 encoded JSON array - * @param valueType the generic type of each array element - * @param the type of parsed values - * @return a flow emitting the array elements - */ + /** Parses generic elements from a top-level JSON array using the default mapper. */ public static Flow parseArray(ByteFlow bytes, TypeReference valueType) { return parseArray( bytes, DEFAULT_MAPPER.readerFor(Objects.requireNonNull(valueType, "valueType"))); } /** - * Parses a top-level JSON array using the supplied Jackson reader. The returned flow - * incrementally emits array elements as they are parsed. Input that is not one complete - * top-level array, including input containing trailing JSON, fails the flow. Elements can be - * emitted before the input ends, but successful completion waits for end-of-input so that - * trailing content can be rejected. {@link + * Incrementally parses exactly one top-level JSON array using the supplied reader. Successful + * completion waits for end-of-input to reject trailing content. {@link * tools.jackson.databind.DeserializationFeature#FAIL_ON_TRAILING_TOKENS} is disabled while * reading individual elements, regardless of the reader's configuration. The caller must ensure * that {@code T} matches the type configured on the reader. Failures are wrapped in {@link @@ -190,38 +110,21 @@ public static Flow parseArray(ByteFlow bytes, ObjectReader reader) { Objects.requireNonNull(bytes, "bytes"), Objects.requireNonNull(reader, "reader")); } - /** - * Renders values as newline-delimited JSON using a default {@link ObjectMapper}. Every value is - * followed by an LF byte, including the final value. - * - * @param values the values to render - * @param valueType the type of each value - * @param the type of rendered values - * @return a flow emitting UTF-8 encoded NDJSON - */ + /** Renders values as NDJSON using the default mapper. */ public static ByteFlow renderNdjson(Flow values, Class valueType) { return renderNdjson( values, DEFAULT_MAPPER.writerFor(Objects.requireNonNull(valueType, "valueType"))); } - /** - * Renders values as newline-delimited JSON using a default {@link ObjectMapper}. Every value is - * followed by an LF byte, including the final value. - * - * @param values the values to render - * @param valueType the generic type of each value - * @param the type of rendered values - * @return a flow emitting UTF-8 encoded NDJSON - */ + /** Renders generic values as NDJSON using the default mapper. */ public static ByteFlow renderNdjson(Flow values, TypeReference valueType) { return renderNdjson( values, DEFAULT_MAPPER.writerFor(Objects.requireNonNull(valueType, "valueType"))); } /** - * Renders values as newline-delimited JSON using the supplied Jackson writer. Every value is - * followed by an LF byte, including the final value. Writer output containing a raw CR or LF - * byte is rejected, as it would produce invalid NDJSON records. + * Renders values as UTF-8 NDJSON using the supplied writer. Every value, including the final + * one, is followed by LF. Writer output containing raw CR or LF is rejected. * * @param values the values to render * @param writer the writer used to serialize each value @@ -233,40 +136,21 @@ public static ByteFlow renderNdjson(Flow values, ObjectWriter writer) { Objects.requireNonNull(values, "values"), Objects.requireNonNull(writer, "writer")); } - /** - * Renders values as one JSON array using a default {@link ObjectMapper}. Elements are - * serialized one at a time. An empty input flow produces {@code []}. If the flow fails after - * output starts, already-emitted bytes can contain an incomplete array. - * - * @param values the values to render - * @param valueType the type of each array element - * @param the type of rendered values - * @return a flow emitting one UTF-8 encoded JSON array - */ + /** Renders values as one JSON array using the default mapper. */ public static ByteFlow renderArray(Flow values, Class valueType) { return renderArray( values, DEFAULT_MAPPER.writerFor(Objects.requireNonNull(valueType, "valueType"))); } - /** - * Renders values as one JSON array using a default {@link ObjectMapper}. Elements are - * serialized one at a time. An empty input flow produces {@code []}. If the flow fails after - * output starts, already-emitted bytes can contain an incomplete array. - * - * @param values the values to render - * @param valueType the generic type of each array element - * @param the type of rendered values - * @return a flow emitting one UTF-8 encoded JSON array - */ + /** Renders generic values as one JSON array using the default mapper. */ public static ByteFlow renderArray(Flow values, TypeReference valueType) { return renderArray( values, DEFAULT_MAPPER.writerFor(Objects.requireNonNull(valueType, "valueType"))); } /** - * Renders values as one JSON array using the supplied Jackson writer. Elements are serialized - * one at a time. An empty input flow produces {@code []}. If the flow fails after output - * starts, already-emitted bytes can contain an incomplete array. + * Renders values as one UTF-8 JSON array using the supplied writer. An empty flow produces + * {@code []}; a failed flow can leave an incomplete array. * * @param values the values to render * @param writer the writer used to serialize each array element diff --git a/json/src/main/java/com/softwaremill/jox/json/JsonReadSettings.java b/json/src/main/java/com/softwaremill/jox/json/JsonReadSettings.java index a323369..a9507bf 100644 --- a/json/src/main/java/com/softwaremill/jox/json/JsonReadSettings.java +++ b/json/src/main/java/com/softwaremill/jox/json/JsonReadSettings.java @@ -10,9 +10,6 @@ public record JsonReadSettings(int maxNdjsonRecordBytes) { private static final int DEFAULT_MAX_NDJSON_RECORD_BYTES = 32 * 1024 * 1024; - /** - * @throws IllegalArgumentException if {@code maxNdjsonRecordBytes} is not positive - */ public JsonReadSettings { if (maxNdjsonRecordBytes <= 0) { throw new IllegalArgumentException("maxNdjsonRecordBytes must be greater than zero"); @@ -24,11 +21,6 @@ public static JsonReadSettings defaults() { return new JsonReadSettings(DEFAULT_MAX_NDJSON_RECORD_BYTES); } - /** - * Returns a copy with the given maximum UTF-8 encoded NDJSON record size. - * - * @throws IllegalArgumentException if {@code newMaxNdjsonRecordBytes} is not positive - */ public JsonReadSettings maxNdjsonRecordBytes(int newMaxNdjsonRecordBytes) { return new JsonReadSettings(newMaxNdjsonRecordBytes); } From f34fb9a2fe15407ee34d4a64625f8f3c36e9f075 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sebastian=20=C5=81akomy?= Date: Mon, 10 Aug 2026 18:30:25 +0200 Subject: [PATCH 13/13] refactor(json): simplify NDJSON framing --- .../softwaremill/jox/json/NdjsonFraming.java | 87 ++++++------------- 1 file changed, 28 insertions(+), 59 deletions(-) diff --git a/json/src/main/java/com/softwaremill/jox/json/NdjsonFraming.java b/json/src/main/java/com/softwaremill/jox/json/NdjsonFraming.java index 24fa7c4..6401986 100644 --- a/json/src/main/java/com/softwaremill/jox/json/NdjsonFraming.java +++ b/json/src/main/java/com/softwaremill/jox/json/NdjsonFraming.java @@ -5,14 +5,13 @@ import java.nio.charset.CharacterCodingException; import java.nio.charset.CodingErrorAction; import java.nio.charset.StandardCharsets; -import java.util.Iterator; -import java.util.List; -import java.util.NoSuchElementException; import java.util.Optional; import com.softwaremill.jox.flows.ByteChunk; import com.softwaremill.jox.flows.Flow; import com.softwaremill.jox.flows.Flow.ByteFlow; +import com.softwaremill.jox.flows.FlowEmit; +import com.softwaremill.jox.flows.Flows; final class NdjsonFraming { @@ -21,14 +20,18 @@ final class NdjsonFraming { private NdjsonFraming() {} static Flow lines(ByteFlow bytes, int maxRecordBytes) { - return bytes.mapWithResource( - () -> new State(maxRecordBytes), NdjsonFraming::finish, State::records) - .mapConcat(records -> records) - .map(NdjsonFraming::decode); - } - - private static Optional> finish(State state) { - return state.finish().map(record -> List.of(record)); + return Flows.usingEmit( + output -> { + var framer = new Framer(maxRecordBytes); + FlowEmit emitRecord = record -> output.apply(decode(record)); + + bytes.runToEmit(chunk -> framer.emitRecords(chunk, emitRecord)); + + var finalRecord = framer.finish(); + if (finalRecord.isPresent()) { + emitRecord.apply(finalRecord.get()); + } + }); } private static String decode(Record record) { @@ -60,61 +63,27 @@ private static boolean startsWithBom(byte[] bytes) { private record Record(byte[] bytes, boolean first) {} - private static final class State { + private static final class Framer { private final int maxRecordBytes; private final ByteArrayOutputStream buffer = new ByteArrayOutputStream(); private boolean firstRecord = true; - private State(int maxRecordBytes) { + private Framer(int maxRecordBytes) { this.maxRecordBytes = maxRecordBytes; } - private Iterable records(ByteChunk chunk) { - return () -> - new Iterator<>() { - private final List arrays = chunk.getArrays(); - private int arrayIndex; - private int offset; - private Record next; - - @Override - public boolean hasNext() { - if (next != null) { - return true; - } - - while (arrayIndex < arrays.size()) { - var array = arrays.get(arrayIndex); - for (int i = offset; i < array.length; i++) { - if (array[i] == '\n') { - append(array, offset, i - offset); - next = completeRecord(); - offset = i + 1; - if (offset == array.length) { - arrayIndex++; - offset = 0; - } - return true; - } - } - - append(array, offset, array.length - offset); - arrayIndex++; - offset = 0; - } - return false; - } - - @Override - public Record next() { - if (!hasNext()) { - throw new NoSuchElementException(); - } - var result = next; - next = null; - return result; - } - }; + private void emitRecords(ByteChunk chunk, FlowEmit output) throws Exception { + for (var array : chunk.getArrays()) { + int recordStart = 0; + for (int i = 0; i < array.length; i++) { + if (array[i] == '\n') { + append(array, recordStart, i - recordStart); + output.apply(completeRecord()); + recordStart = i + 1; + } + } + append(array, recordStart, array.length - recordStart); + } } private Optional finish() {