eclipse-vertx / eclipse-vertx/vert.x

JsonParser continuing processing after exception causes issues for RxJava 3 Adapters

Open
#4,623 5 comments 0 reactions 1 assignee Claimed by @vietj View on GitHub
bug
Dominant language
Java
Stars
14.7k
Forks
2.1k
Avg merge
2d 7h
Merged PRs (30d)
28

Description

### Version

[this commit](https://github.com/eclipse-vertx/vert.x/commit/d461719c56f82a4d3c6fd2c6261c650203037ed9)
which fixed related issue: https://github.com/eclipse-vertx/vert.x/issues/4338

### Context

We use JsonParser with HttpServerRequest bodies to process streams but also with the RxJava Adapters.

I think there are 2 issues present:

1. the stream wants to cancel / unsubscribe on the first exception (removing the exception handler)
this causes a NPE exception here:
https://github.com/eclipse-vertx/vert.x/blob/d461719c56f82a4d3c6fd2c6261c650203037ed9/src/main/java/io/vertx/core/parsetools/impl/JsonParserImpl.java#L325-L327

2. if a exception occurs JsonParser first emits successful events and then the exception **including jsonevents AFTER the exception**
https://github.com/eclipse-vertx/vert.x/blob/d461719c56f82a4d3c6fd2c6261c650203037ed9/src/main/java/io/vertx/core/parsetools/impl/JsonParserImpl.java#L161-L162

I could not write a clean reproducer/test-case for this yet.
When consuming a JsonArray containing larger JsonObjects we transform them into DTOs.
The json was malformed `{goodobj},THISISBAD{goodobj}` and instead of the JsonParserException terminating the processing JsonParser emitted only a part of the second `{goodobj}` leading to a wrong DTO.

instead of emitting the json-event for the entire `{goodobj}` we just got `"someproperty":true` from inside said obj

### Steps to reproduce

```
import com.fasterxml.jackson.core.JsonParseException;
import io.reactivex.rxjava3.core.Completable;
import io.reactivex.rxjava3.core.Flowable;
import io.vertx.core.http.HttpClientOptions;
import io.vertx.core.http.HttpMethod;
import io.vertx.core.http.HttpServerOptions;
import io.vertx.core.parsetools.JsonParser;
import io.vertx.junit5.VertxExtension;
import io.vertx.junit5.VertxTestContext;
import io.vertx.rxjava3.CompletableHelper;
import io.vertx.rxjava3.core.Vertx;
import io.vertx.rxjava3.core.buffer.Buffer;
import io.vertx.rxjava3.core.http.HttpClient;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;

import java.util.List;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

@ExtendWith(VertxExtension.class)
class JsonParserTest {

private static final HttpServerOptions serverOptions = new HttpServerOptions()
.setPort(8080)
.setHost("localhost");
private static final HttpClientOptions clientOptions = new HttpClientOptions()
.setDefaultHost("localhost")
.setDefaultPort(8080);

private Vertx vertx;
private HttpClient client;

@BeforeEach
void beforeEach(final io.vertx.core.Vertx vertx) {
this.vertx = Vertx.newInstance(vertx);
this.client = this.vertx.createHttpClient(clientOptions);
}

@Test
void shouldFailSingleBadJsonEvent(final VertxTestContext context) {
final var checkpoint = context.checkpoint(2);
vertx.getOrCreateContext().exceptionHandler(context::failNow);
rxTest(
Flowable.fromArray(
Buffer.buffer("[b]")
),
(status, actual) -> context.verify(() -> {
Assertions.assertEquals(200, status);
Assertions.assertEquals(List.of(),
actual
);
}),
cause -> {
checkpoint.flag();
context.verify(() -> Assertions.assertEquals(JsonParseException.class, cause.getClass()));
}
)
.subscribe(CompletableHelper.toObserver(context.succeeding(nothing -> checkpoint.flag())));
}

@Test
void shouldFailBadJsonEventInbetween(final VertxTestContext context) {
final var checkpoint = context.checkpoint(2);
vertx.getOrCreateContext().exceptionHandler(context::failNow);
rxTest(
Flowable.fromArray(
Buffer.buffer("["),
Buffer.buffer("1,"),
Buffer.buffer("2,"),
Buffer.buffer("3,"),
Buffer.buffer("b,"),
Buffer.buffer("4,"),
Buffer.buffer("5,"),
Buffer.buffer("6,"),
Buffer.buffer("7,"),
Buffer.buffer("8,"),
Buffer.buffer("9,"),
Buffer.buffer("]")
),
(status, actual) -> context.verify(() -> {
Assertions.assertEquals(200, status);
Assertions.assertEquals(List.of(
Buffer.buffer("["),
Buffer.buffer("1"),
Buffer.buffer("2"),
Buffer.buffer("3")
),
actual
);
}),
cause -> {
checkpoint.flag();
context.verify(() -> Assertions.assertEquals(JsonParseException.class, cause.getClass()));
}
)
.subscribe(CompletableHelper.toObserver(context.succeeding(nothing -> checkpoint.flag())));
}

private Completable rxTest(final Flowable given,
final BiConsumer> actual,
final Consumer exception) {
return Completable.concatArray(
vertx.createHttpServer(serverOptions)
.requestHandler(
request -> io.vertx.rxjava3.core.parsetools.JsonParser.newInstance(JsonParser.newParser(request.getDelegate()))
.toFlowable()
.doOnSubscribe(dis -> request.getDelegate().response().setChunked(true))
.map(event -> switch (event.type()) {
case START_OBJECT -> Buffer.buffer("{");
case END_OBJECT -> Buffer.buffer("}");
case START_ARRAY -> Buffer.buffer("[");
case END_ARRAY -> Buffer.buffer("]");
case VALUE -> Buffer.buffer(event.value().toString());
})
.concatMapCompletable(event -> request.response().rxWrite(event))
.doOnError(exception::accept)
.onErrorComplete()
.andThen(request.response().rxEnd())
.subscribe()
)
.rxListen()
.ignoreElement(),
client.rxRequest(HttpMethod.POST, "/path")
.flatMap(request -> request.rxSend(given))
.flatMap(response -> response.toObservable().toList().doOnSuccess(body -> actual.accept(response.statusCode(), body)))
.ignoreElement()
);
}
}

```

### Extra

* Anything that can be relevant such as OS version, JVM version

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.