Skip to content

Commit

Permalink
Fixes #323 - IllegalStateException in ContentSource.demand(). (#336)
Browse files Browse the repository at this point in the history 
Before, the application BiFunction passed to `request.response(BiFunction)` was called at the "headers" event.

Eventually, a `Publisher<T>` was returned to the application that could subscribe to it, triggering demand to the internal `Publisher<Chunk>`, which is triggering the read of `Chunk`s from the `Content.Source`.

At the same time, the "content source event" may be delivered, triggering also the read of `Chunk`s from the `Content.Source`.

The solution was to delay the call of the application BiFunction at the "content source event".
In this way, there is only one trigger to read `Chunk`s from the `Content.Source`: when the application demands on the `Publisher<T>` returned to the application.

Signed-off-by: Simone Bordet <[email protected]>
  • Loading branch information
@sbordet
sbordet authored Dec 18, 2023
1 parent 6d56e67 commit c55c98a
Show file tree
Hide file tree
Showing 6 changed files with 111 additions and 25 deletions.
6 changes: 6 additions & 0 deletions pom.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -370,6 +370,12 @@
<version>1.0.7</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.awaitility</groupId>
<artifactId>awaitility</artifactId>
<version>4.2.0</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter</artifactId>
Expand Down
1 change: 0 additions & 1 deletion src/main/java/org/eclipse/jetty/reactive/client/internal/AbstractBufferingProcessor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@

import java.util.ArrayList;
import java.util.List;

import org.eclipse.jetty.io.Content;
import org.eclipse.jetty.reactive.client.ReactiveResponse;

Expand Down
1 change: 0 additions & 1 deletion src/main/java/org/eclipse/jetty/reactive/client/internal/AbstractSingleProcessor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@
package org.eclipse.jetty.reactive.client.internal;

import java.util.Objects;

import org.eclipse.jetty.util.MathUtils;
import org.eclipse.jetty.util.thread.AutoLock;
import org.reactivestreams.Processor;
Expand Down
28 changes: 17 additions & 11 deletions src/main/java/org/eclipse/jetty/reactive/client/internal/ResponseListenerProcessor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@

import java.util.concurrent.CancellationException;
import java.util.function.BiFunction;

import org.eclipse.jetty.client.Response;
import org.eclipse.jetty.client.Result;
import org.eclipse.jetty.io.Content;
Expand Down Expand Up @@ -73,23 +72,31 @@ public void onHeaders(Response response) {
if (logger.isDebugEnabled()) {
logger.debug("received response headers {} on {}", response, this);
}
responseReceived = true;
Publisher<T> publisher = contentFn.apply(request.getReactiveResponse(), content);
// Links the publisher/subscriber chain.
// ContentPublisher (reads Chunks)
// `- application Processor (receives Chunks, emits T) [optional]
// `- Publisher (emits Ts)
// `- ResponseListenerProcessor (receives Ts, emits Ts)
// `- application Subscriber (receives Ts)
publisher.subscribe(this);
}

@Override
public void onContentSource(Response response, Content.Source source) {
if (logger.isDebugEnabled()) {
logger.debug("received response content source {} {} on {}", response, source, this);
}

// Link the source of Chunks with the Publisher of Chunks.
content.accept(source);

responseReceived = true;

// Call the application to obtain a response content transformer.
Publisher<T> appPublisher = contentFn.apply(request.getReactiveResponse(), content);

// Links the publisher/subscriber chain.
// Content Chunks flow from top (upstream) to bottom (downstream).
//
// ContentPublisher (reads Chunks)
// `- application Processor (receives Chunks, transforms them and emits Ts) [optional]
// `- application Publisher (emits Ts)
// `- ResponseListenerProcessor (receives Ts, emits Ts)
// `- application Subscriber (receives Ts)
appPublisher.subscribe(this);
}

@Override
Expand Down Expand Up @@ -167,7 +174,6 @@ private static class ContentPublisher extends QueuedSinglePublisher<Content.Chun

private void accept(Content.Source source) {
contentSource = source;
tryProduce(this);
}

@Override
Expand Down
1 change: 0 additions & 1 deletion src/test/java/org/eclipse/jetty/reactive/client/ReactorTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@
import java.time.Duration;
import java.util.Random;
import java.util.concurrent.TimeoutException;

import org.eclipse.jetty.io.Content;
import org.eclipse.jetty.server.Handler;
import org.eclipse.jetty.server.Request;
Expand Down
99 changes: 88 additions & 11 deletions src/test/java/org/eclipse/jetty/reactive/client/RxJava2Test.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30,9 +30,9 @@
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import io.reactivex.rxjava3.core.Emitter;
import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.core.Single;
Expand Down Expand Up @@ -61,6 +61,10 @@
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import static java.nio.charset.StandardCharsets.UTF_8;
import static org.awaitility.Awaitility.await;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
Expand All @@ -84,7 +88,7 @@ public boolean handle(Request request, Response response, Callback callback) {
Flowable.fromPublisher(request.response((reactiveResponse, chunkPublisher) -> Flowable.fromPublisher(chunkPublisher)
.map(chunk -> {
ByteBuffer byteBuffer = chunk.getByteBuffer();
CharBuffer charBuffer = StandardCharsets.UTF_8.decode(byteBuffer);
CharBuffer charBuffer = UTF_8.decode(byteBuffer);
chunk.release();
return charBuffer.toString();
}).doOnComplete(contentLatch::countDown)))
Expand Down Expand Up @@ -251,7 +255,7 @@ public boolean handle(Request request, Response response, Callback callback) {

String text = "Γειά σου Κόσμε";
ReactiveRequest request = ReactiveRequest.newBuilder(httpClient(), uri())
.content(ReactiveRequest.Content.fromString(text, "text/plain", StandardCharsets.UTF_8))
.content(ReactiveRequest.Content.fromString(text, "text/plain", UTF_8))
.build();

String content = Single.fromPublisher(request.response(ReactiveResponse.Content.asString()))
Expand Down Expand Up @@ -280,7 +284,7 @@ public boolean handle(Request request, Response response, Callback callback) {
Flowable<String> stream = Flowable.fromArray(data.split("(?!^)"));

// Transform it to chunks, showing what you can use the callback for.
Charset charset = StandardCharsets.UTF_8;
Charset charset = UTF_8;
ByteBufferPool bufferPool = httpClient().getByteBufferPool();
Flowable<Content.Chunk> chunks = stream.map(item -> item.getBytes(charset))
.map(bytes -> {
Expand Down Expand Up @@ -380,6 +384,79 @@ public boolean handle(Request request, Response response, Callback callback) {
assertEquals(responseContent, pangram);
}

@ParameterizedTest
@MethodSource("protocols")
public void testFlowableResponseBodyBackPressure(String protocol) throws Exception {
prepare(protocol, new Handler.Abstract() {
@Override
public boolean handle(Request request, Response response, Callback callback) {
response.getHeaders().put(HttpHeader.CONTENT_TYPE, "text/plain");
Callback.Completable completable = new Callback.Completable();
Content.Sink.write(response, false, "hello", completable);
completable.thenRun(() -> Content.Sink.write(response, true, "world", callback));
return true;
}
});

AtomicInteger chunks = new AtomicInteger();
ReactiveRequest request = ReactiveRequest.newBuilder(httpClient(), uri()).build();
request.getRequest().onResponseContent((response, chunk) -> chunks.incrementAndGet());

Publisher<Content.Chunk> publisher = request.response((response, content) -> content);

CountDownLatch completeLatch = new CountDownLatch(1);
var subscriber = new Subscriber<Content.Chunk>() {
private Subscription subscription;
private Content.Chunk chunk;

@Override
public void onSubscribe(Subscription subscription) {
this.subscription = subscription;
}

@Override
public void onNext(Content.Chunk chunk) {
this.chunk = chunk;
}

@Override
public void onError(Throwable throwable) {
}

@Override
public void onComplete() {
completeLatch.countDown();
}
};
publisher.subscribe(subscriber);

// There is no demand yet, so there should be no chunks.
assertEquals(0, chunks.get());

// Send the request and demand 1 chunk of content.
subscriber.subscription.request(1);

// There should be 1 chunk only.
await().during(1, TimeUnit.SECONDS).atMost(5, TimeUnit.SECONDS).until(chunks::get, is(1));
Content.Chunk chunk = await().atMost(5, TimeUnit.SECONDS).until(() -> subscriber.chunk, notNullValue());
subscriber.chunk = null;
assertEquals("hello", UTF_8.decode(chunk.getByteBuffer()).toString());

// Wait to be sure there is backpressure.
await().during(1, TimeUnit.SECONDS).atMost(5, TimeUnit.SECONDS).until(chunks::get, is(1));

// Demand 1 more chunk.
subscriber.subscription.request(1);
chunk = await().atMost(5, TimeUnit.SECONDS).until(() -> subscriber.chunk, notNullValue());
subscriber.chunk = null;
assertEquals("world", UTF_8.decode(chunk.getByteBuffer()).toString());

// Demand completion.
subscriber.subscription.request(1);

assertTrue(completeLatch.await(5, TimeUnit.SECONDS));
}

@ParameterizedTest
@MethodSource("protocols")
public void testFlowableResponsePipedToRequest(String protocol) throws Exception {
Expand Down Expand Up @@ -500,7 +577,7 @@ public boolean handle(Request request, Response response, Callback callback) {
@MethodSource("protocols")
public void testResponseContentTimeout(String protocol) throws Exception {
long timeout = 500;
byte[] data = "hello".getBytes(StandardCharsets.UTF_8);
byte[] data = "hello".getBytes(UTF_8);
prepare(protocol, new Handler.Abstract() {
@Override
public boolean handle(Request request, Response response, Callback callback) throws Exception {
Expand Down Expand Up @@ -571,8 +648,8 @@ public boolean handle(Request request, Response response, Callback callback) thr
.blockingGet();

assertEquals(result.response().getStatus(), HttpStatus.OK_200);
;
String expected = StandardCharsets.UTF_8.decode(ByteBuffer.allocate(original.length)

String expected = UTF_8.decode(ByteBuffer.allocate(original.length)
.put(original)
.flip()).toString();
assertEquals(expected, result.content());
Expand Down Expand Up @@ -640,7 +717,7 @@ public boolean handle(Request request, Response response, Callback callback) {

String text = "hello world";
ReactiveRequest request = ReactiveRequest.newBuilder(httpClient().newRequest(uri()).method(HttpMethod.POST))
.content(ReactiveRequest.Content.fromString(text, "text/plain", StandardCharsets.UTF_8))
.content(ReactiveRequest.Content.fromString(text, "text/plain", UTF_8))
.build();
String content = Single.fromPublisher(request.response(ReactiveResponse.Content.asString()))
.blockingGet();
Expand Down Expand Up @@ -668,9 +745,9 @@ public boolean handle(Request request, Response response, Callback callback) {
String text2 = "world";
ChunkListSinglePublisher publisher = new ChunkListSinglePublisher();
// Offer content to trigger the sending of the request and the processing on the server.
publisher.offer(StandardCharsets.UTF_8.encode(text1));
publisher.offer(UTF_8.encode(text1));
ReactiveRequest request = ReactiveRequest.newBuilder(httpClient().newRequest(uri()).method(HttpMethod.POST))
.content(ReactiveRequest.Content.fromPublisher(publisher, "text/plain", StandardCharsets.UTF_8))
.content(ReactiveRequest.Content.fromPublisher(publisher, "text/plain", UTF_8))
.build();
Single<String> flow = Single.fromPublisher(request.response(ReactiveResponse.Content.asString()));
// Send the request by subscribing as a CompletableFuture.
Expand All @@ -679,7 +756,7 @@ public boolean handle(Request request, Response response, Callback callback) {
// Allow the redirect to happen.
Thread.sleep(1000);

publisher.offer(StandardCharsets.UTF_8.encode(text2));
publisher.offer(UTF_8.encode(text2));
publisher.complete();

assertEquals(text1 + text2, completable.get(5, TimeUnit.SECONDS));
Expand Down

0 comments on commit c55c98a

Please sign in to comment.