How to consume an InputStream in an HttpResponse.BodyHandler?

Question

I'm trying to build a JSON handler for the Java 11 HttpClient. The problem I'm having is that trying to compose with BodySubscribers.ofInputStream() results in never reading any data from the stream, and hanging forever.

@Override
public HttpResponse.BodySubscriber<Void> apply(HttpResponse.ResponseInfo responseInfo) {
    return HttpResponse.BodySubscribers.mapping(
            HttpResponse.BodySubscribers.ofInputStream(),
            inputStream -> {
                try {
                    System.out.print(inputStream.read());
                } catch (IOException e) {
                    throw new UncheckedIOException(e);
                }
                return null;
            }
    );
}

In contrast, returning the stream and reading it afterwards does work.

InputStream inputStream = client.send(request, HttpResponse.BodyHandlers.ofInputStream()).body();
System.out.print(inputStream.read());  // out: 123

What am I missing?

Answer 1

The problem here lays in invoking a blocking operation in a thread that is not supposed to block. And really - there are two issues:

1. The Java 11 API documentation hints that you can do this - and that's really not a good idea.
2. The implementation wasn't prepared to handle a blocking operation in the mapper's function.

Both issues have been fixed in 13 by JDK-8217264: HttpClient: Blocking operations in mapper function do not work as documented

If you look at a more recent version of the API documentation you will see that it has now been updated to cover this case. Although with 13 onward doing so will no longer wedge the request, blocking in the mapper's function is still discouraged, as it will block one of the HttpClient's executor thread until the response is fully received, which could lead to thread starvation.

To complement my answer - you should consider following the recommendation of the newer API documentation and return a Supplier<JSONObject> instead of a JSONObject, which would allow to delay the reading of the stream (and the blocking operation) until Supplier::get is called by the caller. That gives you the flexibility of choosing which thread will block waiting for the response.