Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 0 additions & 4 deletions java-client/build.gradle.kts
Original file line number Diff line number Diff line change
Expand Up @@ -187,10 +187,6 @@ dependencies {
compileOnly("org.elasticsearch.client", "elasticsearch-rest-client", elasticsearchVersion)
testImplementation("org.elasticsearch.client", "elasticsearch-rest-client", elasticsearchVersion)

// Apache 2.0
// https://hc.apache.org/httpcomponents-client-ga/
api("org.apache.httpcomponents.client5","httpclient5","5.4.4")

// Apache 2.0
// https://search.maven.org/artifact/com.google.code.findbugs/jsr305
api("com.google.code.findbugs:jsr305:3.0.2")
Expand Down
2 changes: 1 addition & 1 deletion rest5-client/build.gradle.kts
Original file line number Diff line number Diff line change
Expand Up @@ -140,7 +140,7 @@ signing {
dependencies {
// Apache 2.0
// https://hc.apache.org/httpcomponents-client-ga/
api("org.apache.httpcomponents.client5","httpclient5","5.4.4")
api("org.apache.httpcomponents.client5","httpclient5","5.6")

// Apache 2.0
// http://commons.apache.org/logging/
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -19,13 +19,11 @@

package co.elastic.clients.transport.rest5_client.low_level;


import org.apache.hc.core5.http.ClassicHttpResponse;
import org.apache.hc.core5.http.ContentType;
import org.apache.hc.core5.http.HttpResponse;
import org.apache.hc.core5.http.io.entity.ByteArrayEntity;
import org.apache.hc.core5.http.message.BasicClassicHttpResponse;
import org.apache.hc.core5.http.nio.entity.AbstractBinAsyncEntityConsumer;
import org.apache.hc.core5.http.nio.support.AbstractAsyncResponseConsumer;
import org.apache.hc.core5.http.protocol.HttpContext;

Expand All @@ -39,7 +37,7 @@ class BasicAsyncResponseConsumer extends AbstractAsyncResponseConsumer<ClassicHt
/**
* Creates a new instance of this consumer with the provided buffer limit
*/
BasicAsyncResponseConsumer(AbstractBinAsyncEntityConsumer consumer) {
BasicAsyncResponseConsumer(BufferedByteConsumer consumer) {
super(consumer);
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,21 +18,28 @@
*/
package co.elastic.clients.transport.rest5_client.low_level;

import org.apache.hc.core5.concurrent.FutureCallback;
import org.apache.hc.core5.http.ContentTooLongException;
import org.apache.hc.core5.http.ContentType;
import org.apache.hc.core5.http.EntityDetails;
import org.apache.hc.core5.http.io.entity.ByteArrayEntity;
import org.apache.hc.core5.http.nio.entity.AbstractBinAsyncEntityConsumer;
import org.apache.hc.core5.http.nio.AsyncEntityConsumer;
import org.apache.hc.core5.http.nio.entity.AbstractBinDataConsumer;
import org.apache.hc.core5.util.ByteArrayBuffer;

import java.io.IOException;
import java.nio.ByteBuffer;

import static co.elastic.clients.transport.rest5_client.low_level.Constants.DEFAULT_BUFFER_INITIAL_CAPACITY;

public class BufferedByteConsumer extends AbstractBinAsyncEntityConsumer<ByteArrayEntity> {
public class BufferedByteConsumer extends AbstractBinDataConsumer implements AsyncEntityConsumer<ByteArrayEntity> {

private volatile ByteArrayBuffer buffer;
private final int limit;
private ContentType contentType;
private volatile ByteArrayBuffer buffer;
private volatile FutureCallback<ByteArrayEntity> resultCallback;
private volatile ContentType contentType;
private volatile String contentEncoding;
private volatile ByteArrayEntity result;

public BufferedByteConsumer(int bufferLimit) {
super();
Expand All @@ -44,8 +51,11 @@ public BufferedByteConsumer(int bufferLimit) {
}

@Override
protected void streamStart(final ContentType contentType) {
this.contentType = contentType;
public void streamStart(final EntityDetails entityDetails,
final FutureCallback<ByteArrayEntity> resultCallback) {
this.contentType = entityDetails != null ? ContentType.parse(entityDetails.getContentType()) : null;
this.contentEncoding = entityDetails != null ? entityDetails.getContentEncoding() : null;
this.resultCallback = resultCallback;
}

@Override
Expand All @@ -64,8 +74,25 @@ protected void data(final ByteBuffer src, final boolean endOfStream) throws Cont
}

@Override
protected ByteArrayEntity generateContent() {
return new ByteArrayEntity(buffer.toByteArray(), contentType);
protected final void completed() throws IOException {
result = new ByteArrayEntity(buffer.toByteArray(), contentType, contentEncoding);
if (resultCallback != null) {
resultCallback.completed(result);
}
releaseResources();
}

@Override
public ByteArrayEntity getContent() {
return result;
}

@Override
public final void failed(final Exception cause) {
if (resultCallback != null) {
resultCallback.failed(cause);
}
releaseResources();
}

@Override
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,6 @@
import org.apache.hc.core5.http.HttpEntity;
import org.apache.hc.core5.http.HttpHost;
import org.apache.hc.core5.http.HttpRequest;
import org.apache.hc.core5.http.ProtocolException;
import org.apache.hc.core5.http.message.RequestLine;
import org.apache.hc.core5.http.nio.AsyncEntityProducer;
import org.apache.hc.core5.http.nio.AsyncRequestProducer;
Expand Down Expand Up @@ -339,13 +338,7 @@ private ResponseOrResponseException convertResponse(InternalRequest request, Nod

HttpEntity entity = httpResponse.getEntity();
if (entity != null) {
Header encoding = null;
try {
encoding = httpResponse.getHeader(CONTENT_ENCODING);
} catch (ProtocolException e) {
throw new IOException("Couldn't retrieve content encoding: " + e);
}
if (encoding != null && "gzip".equals(encoding.getValue())) {
if ("gzip".equals(entity.getContentEncoding())) {
// Decompress and cleanup response headers
httpResponse.setEntity(new GzipDecompressingEntity(entity));
httpResponse.removeHeaders(CONTENT_ENCODING);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import com.sun.net.httpserver.HttpExchange;
import com.sun.net.httpserver.HttpHandler;
import com.sun.net.httpserver.HttpServer;
import org.apache.hc.client5.http.impl.async.HttpAsyncClientBuilder;
import org.apache.hc.core5.http.ContentType;
import org.apache.hc.core5.http.HttpEntity;
import org.apache.hc.core5.http.HttpHost;
Expand Down Expand Up @@ -122,10 +123,16 @@ private static byte[] readAll(InputStream in) throws IOException {
return bos.toByteArray();
}

// From org.apache.httpcomponents.client5.httpclient5.5.6, the client decompresses responses
// by default, so to test the java client's decompression logic, a custom http client with decompression
// disabled must be used.
private Rest5Client createClient(boolean enableCompression) {
InetSocketAddress address = httpServer.getAddress();
return Rest5Client.builder(new HttpHost("http", address.getHostString(), address.getPort()))
.setCompressionEnabled(enableCompression)
.setHttpClient(HttpAsyncClientBuilder.create()
.disableContentCompression()
.build())
.build();
}

Expand Down Expand Up @@ -205,6 +212,9 @@ public void testCompressingClientAsync() throws Exception {
Rest5Client restClient = Rest5Client.builder(new HttpHost("http", address.getHostString(),
address.getPort()))
.setCompressionEnabled(true)
.setHttpClient(HttpAsyncClientBuilder.create()
.disableContentCompression()
.build())
.build();

Request request = new Request("POST", "/");
Expand Down