Merge pull request #1443 from gentics/hotfix-1.8.x-sup-12629

Port of Fix for SUP-12629

LTS-CHANGELOG.adoc

@@ -17,6 +17,13 @@ include::content/docs/variables.adoc-include[]
 The LTS changelog lists releases which are only accessible via a commercial subscription.
 All fixes and changes in LTS releases will be released the next minor release. Changes from LTS 1.4.x will be included in release 1.5.0.
 
+[[v1.8.10]]
+== 1.8.10 (TBD)
+
+icon:check[] Java Rest Client: The classes `MeshWebrootResponse` and `MeshWebrootFieldResponse` now extend `Closeable` and calling `close()` on implementations
+will close the underlying response. See link:{{< relref "java-rest-client.asciidoc" >}}#_connection_leaks[Gentics Mesh Java Client] for examples of how to properly
+close the responses.
+
 [[v1.8.9]]
 == 1.8.9 (07.09.2022)
 
@@ -53,6 +60,13 @@ enforced in the UI, and is now enforced in the REST API. The check is not perfor
 
 icon:check[] Core: When a project was deleted, its associated version purge job were not. This has been fixed.
 
+[[v1.6.34]]
+== 1.6.34 (TBD)
+
+icon:check[] Java Rest Client: The classes `MeshWebrootResponse` and `MeshWebrootFieldResponse` now extend `Closeable` and calling `close()` on implementations
+will close the underlying response. See link:{{< relref "java-rest-client.asciidoc" >}}#_connection_leaks[Gentics Mesh Java Client] for examples of how to properly
+close the responses.
+
 [[v1.6.33]]
 == 1.6.33 (07.09.2022)
 

core/src/main/java/com/gentics/mesh/rest/MeshLocalRequestImpl.java

@@ -63,6 +63,10 @@ public T getBody() {
 			public List<String> getCookies() {
 				throw new RuntimeException("There are no cookies in local requests");
 			}
+
+			@Override
+			public void close() {
+			}
 		});
 	}
 

doc/src/main/docs/java-rest-client.asciidoc

@@ -96,6 +96,48 @@ Method to set the hostname verification checks:
 
   * `MeshRestClientConfig.Builder#setHostnameVerification(boolean flag)`
 
+=== Connection Leaks
+
+In most cases, the connection to Mesh will implicitly be closed when the response object is created. The only exceptions are when loading binary data or doing a webroot or webrootField request (where the response might contain binary data).
+
+For those requests, it is important to make sure, that the response is properly closed in all cases, especially when the response data is not (fully) consumed (e.g. when an error happens while consuming the binary data).
+
+Incorrect example:
+[source,java]
+----
+client.downloadBinaryField("demo", "01ecd6048ee21471bb90af6deea40d2c", "en", "image")
+	.toSingle()
+	.doOnSuccess(response -> {
+		// an exception might be thrown here, which would leave the response open
+	})
+	.subscribe();
+----
+
+Better example:
+[source,java]
+----
+client.downloadBinaryField("demo", "01ecd6048ee21471bb90af6deea40d2c", "en", "image")
+	.toSingle()
+	.doAfterSuccess(response -> {
+		// we explicitly need to close the response here
+		response.close();
+	})
+	.doOnSuccess(response -> {
+		// an exception might be thrown here
+	})
+	.subscribe();
+----
+
+Example with blocking code:
+[source,java]
+----
+// try-with-resource will make sure, that close() is called on the response
+try (MeshBinaryResponse response = client.downloadBinaryField("demo", "01ecd6048ee21471bb90af6deea40d2c", "en", "image")
+	.toSingle().blockingGet()) {
+	// an exception might be thrown here
+}
+----
+
 == Monitoring Client
 
 The monitoring client can be used to interact with the link:{{< relref "monitoring.asciidoc" >}}#_endpoints[Monitoring Endpoints].

rest-client/src/main/java/com/gentics/mesh/rest/client/MeshBinaryResponse.java

@@ -26,7 +26,7 @@ public interface MeshBinaryResponse extends Closeable {
 	 * @return
 	 */
 	default Flowable<byte[]> getFlowable() {
-		return Flowable.defer(() -> {
+		Flowable<byte[]> f = Flowable.defer(() -> {
 			InputStream stream = getStream();
 			return Flowable.generate(emitter -> {
 				byte[] buffer = new byte[FLOWABLE_BUFFER_SIZE];
@@ -42,6 +42,8 @@ default Flowable<byte[]> getFlowable() {
 				}
 			});
 		});
+
+		return f.doFinally(() -> close());
 	}
 
 	/**

rest-client/src/main/java/com/gentics/mesh/rest/client/MeshResponse.java

@@ -1,5 +1,6 @@
 package com.gentics.mesh.rest.client;
 
+import java.io.Closeable;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -10,8 +11,9 @@
  * 
  * @param <T>
  *            Response type
+ * @implNote It is important to close the response by calling {@link #close()} when the response is no longer needed. Failing to do so might lead to a connection leak.
  */
-public interface MeshResponse<T> {
+public interface MeshResponse<T> extends Closeable {
 	/**
 	 * Retrieve the response headers
 	 * 
@@ -77,4 +79,7 @@ default List<String> getCookies() {
 	 * @return The body as the specified type
 	 */
 	T getBody();
+
+	@Override
+	void close();
 }

rest-client/src/main/java/com/gentics/mesh/rest/client/MeshWebrootFieldResponse.java

@@ -1,12 +1,15 @@
 package com.gentics.mesh.rest.client;
 
+import java.io.Closeable;
+
 /**
  * Response for {@link WebRootFieldClientMethods}
  * 
  * @author plyhun
  *
+ * @implNote It is important to close the response by calling {@link #close()} when the response is no longer needed. Failing to do so might lead to a connection leak.
  */
-public interface MeshWebrootFieldResponse {
+public interface MeshWebrootFieldResponse extends Closeable {
 
 	/**
 	 * Tests if the response is binary data.
@@ -63,4 +66,7 @@ public interface MeshWebrootFieldResponse {
 	 * @return
 	 */
 	String getResponseAsPlainText();
+
+	@Override
+	void close();
 }

rest-client/src/main/java/com/gentics/mesh/rest/client/MeshWebrootResponse.java

@@ -1,12 +1,15 @@
 package com.gentics.mesh.rest.client;
 
+import java.io.Closeable;
+
 import com.gentics.mesh.core.rest.node.NodeResponse;
 
 /**
  * Definition of a WebRootResponse. The webroot response is special since it can return JSON of the {@link NodeResponse} or a {@link MeshBinaryResponse} of the
  * binary data of a binary field. This behaviour is controlled by the Accept header of the client request and the queried node (e.g. whether it uses a binary field for the segment).
+ * @implNote It is important to close the response by calling {@link #close()} when the response is no longer needed. Failing to do so might lead to a connection leak.
  */
-public interface MeshWebrootResponse {
+public interface MeshWebrootResponse extends Closeable {
 
 	/**
 	 * Tests if the response is binary data.
@@ -35,4 +38,7 @@ public interface MeshWebrootResponse {
 	 * @return
 	 */
 	String getNodeUuid();
+
+	@Override
+	void close();
 }

rest-client/src/main/java/com/gentics/mesh/rest/client/impl/MeshOkHttpRequestImpl.java

@@ -362,6 +362,11 @@ public String getBodyAsString() {
 			public T getBody() {
 				return body.get();
 			}
+
+			@Override
+			public void close() {
+				Optional.ofNullable(response).map(Response::body).ifPresent(ResponseBody::close);
+			}
 		});
 	}
 

rest-client/src/main/java/com/gentics/mesh/rest/client/impl/OkHttpWebrootFieldResponse.java

@@ -1,12 +1,13 @@
 package com.gentics.mesh.rest.client.impl;
 
+import static com.gentics.mesh.http.HttpConstants.TEXT_HTML_UTF8;
+import static com.gentics.mesh.http.HttpConstants.TEXT_PLAIN_UTF8;
 import static com.gentics.mesh.http.MeshHeaders.WEBROOT_NODE_UUID;
 import static com.gentics.mesh.http.MeshHeaders.WEBROOT_RESPONSE_TYPE;
-import static io.vertx.core.http.HttpHeaders.CONTENT_TYPE;
-import static com.gentics.mesh.http.HttpConstants.TEXT_PLAIN_UTF8;
-import static com.gentics.mesh.http.HttpConstants.TEXT_HTML_UTF8;
 import static com.gentics.mesh.rest.client.impl.Util.lazily;
+import static io.vertx.core.http.HttpHeaders.CONTENT_TYPE;
 
+import java.util.Optional;
 import java.util.function.Supplier;
 
 import org.apache.commons.lang.StringUtils;
@@ -17,6 +18,7 @@
 
 import okhttp3.OkHttpClient;
 import okhttp3.Response;
+import okhttp3.ResponseBody;
 
 /**
  * {@link OkHttpClient} implementation of {@link MeshWebrootFieldResponse}.
@@ -87,5 +89,10 @@ public String getResponseAsPlainText() {
 			return null;
 		}
 		return jsonStringResponse.get();
-	}	
+	}
+
+	@Override
+	public void close() {
+		Optional.ofNullable(response).map(Response::body).ifPresent(ResponseBody::close);
+	}
 }

rest-client/src/main/java/com/gentics/mesh/rest/client/impl/OkHttpWebrootResponse.java

@@ -1,19 +1,21 @@
 package com.gentics.mesh.rest.client.impl;
 
-import com.gentics.mesh.core.rest.node.NodeResponse;
-
-import com.gentics.mesh.json.JsonUtil;
-import com.gentics.mesh.rest.client.MeshBinaryResponse;
-import com.gentics.mesh.rest.client.MeshWebrootResponse;
-import okhttp3.Response;
+import static com.gentics.mesh.http.MeshHeaders.WEBROOT_NODE_UUID;
+import static com.gentics.mesh.http.MeshHeaders.WEBROOT_RESPONSE_TYPE;
+import static com.gentics.mesh.rest.client.impl.Util.lazily;
 
+import java.util.Optional;
 import java.util.function.Supplier;
 
 import org.apache.commons.lang.StringUtils;
 
-import static com.gentics.mesh.http.MeshHeaders.WEBROOT_RESPONSE_TYPE;
-import static com.gentics.mesh.http.MeshHeaders.WEBROOT_NODE_UUID;
-import static com.gentics.mesh.rest.client.impl.Util.lazily;
+import com.gentics.mesh.core.rest.node.NodeResponse;
+import com.gentics.mesh.json.JsonUtil;
+import com.gentics.mesh.rest.client.MeshBinaryResponse;
+import com.gentics.mesh.rest.client.MeshWebrootResponse;
+
+import okhttp3.Response;
+import okhttp3.ResponseBody;
 
 /**
  * OkHttp specific webroot response implementation.
@@ -57,4 +59,9 @@ public NodeResponse getNodeResponse() {
 			? null
 			: nodeResponse.get();
 	}
+
+	@Override
+	public void close() {
+		Optional.ofNullable(response).map(Response::body).ifPresent(ResponseBody::close);
+	}
 }

tests/common/src/main/java/com/gentics/mesh/test/ConnectionVerifier.java

@@ -0,0 +1,95 @@
+package com.gentics.mesh.test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.rules.Verifier;
+
+import io.reactivex.Flowable;
+import io.reactivex.disposables.Disposable;
+import okhttp3.ConnectionPool;
+import okhttp3.OkHttpClient;
+
+/**
+ * Testrule, that verifies that for the given OkHttpClient, the number of used
+ * connections in the pool do not change between creation of the instance (at
+ * start of the test) and end of the test
+ */
+public class ConnectionVerifier extends Verifier {
+	/**
+	 * Connection pool to check
+	 */
+	protected ConnectionPool connectionPool;
+
+	/**
+	 * Number of used connections at creation time
+	 */
+	protected int usedConnectionsBeforeTest = 0;
+
+	/**
+	 * Timeout for waiting for the connection to be freed
+	 */
+	protected long timeout = 10;
+
+	/**
+	 * Unit for the waiting timeout
+	 */
+	protected TimeUnit timeoutUnit = TimeUnit.SECONDS;
+
+	/**
+	 * Create an instance for the given client
+	 * @param client client
+	 */
+	public ConnectionVerifier(OkHttpClient client) {
+		this.connectionPool = client.connectionPool();
+		usedConnectionsBeforeTest = connectionPool.connectionCount() - connectionPool.idleConnectionCount();
+	}
+
+	/**
+	 * Set the timeout
+	 * @param timeout maximum time to wait
+	 * @param timeoutUnit unit
+	 * @return fluent API
+	 */
+	public ConnectionVerifier withTimeout(long timeout, TimeUnit timeoutUnit) {
+		this.timeout = timeout;
+		this.timeoutUnit = timeoutUnit;
+		return this;
+	}
+
+	@Override
+	protected void verify() throws Throwable {
+		waitForConnectionFreed();
+		assertThat(getConnectionsUsedByTest()).as("Connections used (and not ended) by test").isEqualTo(0);
+	}
+
+	/**
+	 * Wait for the connection used by the test to become free
+	 * @throws InterruptedException
+	 */
+	protected void waitForConnectionFreed() throws InterruptedException {
+		// now wait for the connection to be freed
+		CountDownLatch waitLatch = new CountDownLatch(1);
+		Disposable disp = Flowable.interval(100, TimeUnit.MILLISECONDS).forEach(ignore -> {
+			if (getConnectionsUsedByTest() == 0) {
+				waitLatch.countDown();
+			}
+		});
+		try {
+			waitLatch.await(timeout, timeoutUnit);
+		} finally {
+			disp.dispose();
+		}
+	}
+
+	/**
+	 * Determine the number of connections used by the test
+	 * @return number of used connections
+	 */
+	protected int getConnectionsUsedByTest() {
+		int usedConnectionsAfterTest = connectionPool.connectionCount() - connectionPool.idleConnectionCount();
+		return usedConnectionsAfterTest - usedConnectionsBeforeTest;
+	}
+}