You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by co...@apache.org on 2024/10/15 16:45:11 UTC
(tinkerpop) 01/03: [TINKERPOP-2700] Add `useCompression` config
This is an automated email from the ASF dual-hosted git repository.
colegreer pushed a commit to branch TINKERPOP-2700-3.7
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit 41ff2990d58d010eae0dd5c367cbe27438f20459
Author: Cole-Greer <Co...@improving.com>
AuthorDate: Mon Oct 14 15:10:52 2024 -0700
[TINKERPOP-2700] Add `useCompression` config
Adds new useCompression configuration to toggle use of permessage deflate
compression in Java, Python, and JS. This was brings these GLVs inline with
.Net and Go which were previously configurable.
This change does not modify the default behaviour regarding compression in any
of the GLVs.
---
.../tinkerpop/gremlin/driver/Channelizer.java | 6 +++--
.../apache/tinkerpop/gremlin/driver/Cluster.java | 26 ++++++++++++++++++++++
.../tinkerpop/gremlin/driver/Connection.java | 2 +-
.../apache/tinkerpop/gremlin/driver/Settings.java | 8 +++++++
.../tinkerpop/gremlin/driver/SettingsTest.java | 2 ++
.../gremlin-javascript/lib/driver/client.js | 1 +
.../gremlin-javascript/lib/driver/connection.js | 2 ++
.../lib/driver/driver-remote-connection.js | 1 +
.../main/python/gremlin_python/driver/client.py | 6 ++++-
.../driver/driver_remote_connection.py | 4 +++-
.../tests/driver/test_driver_remote_connection.py | 22 ++++++++++++++++++
11 files changed, 75 insertions(+), 5 deletions(-)
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java
index a626e0160b..7b9cb27a60 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java
@@ -210,8 +210,10 @@ public interface Channelizer extends ChannelHandler {
pipeline.addLast("http-codec", new HttpClientCodec());
pipeline.addLast("aggregator", new HttpObjectAggregator(maxContentLength));
- // Add compression extension for WebSocket defined in https://tools.ietf.org/html/rfc7692
- pipeline.addLast(WebSocketClientCompressionHandler.INSTANCE);
+ if (connection.getCluster().useCompression()) {
+ // Add compression extension for WebSocket defined in https://tools.ietf.org/html/rfc7692
+ pipeline.addLast(WebSocketClientCompressionHandler.INSTANCE);
+ }
pipeline.addLast("idle-state-Handler", new IdleStateHandler(0, keepAliveInterval, 0));
pipeline.addLast("ws-handler", handler);
pipeline.addLast("gremlin-encoder", gremlinRequestEncoder);
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Cluster.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Cluster.java
index 870bb8bdcc..bd6d642245 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Cluster.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Cluster.java
@@ -210,6 +210,7 @@ public final class Cluster {
.minConnectionPoolSize(settings.connectionPool.minSize)
.connectionSetupTimeoutMillis(settings.connectionPool.connectionSetupTimeoutMillis)
.enableUserAgentOnConnect(settings.enableUserAgentOnConnect)
+ .useCompression(settings.useCompression)
.validationRequest(settings.connectionPool.validationRequest);
if (settings.username != null && settings.password != null)
@@ -578,6 +579,13 @@ public final class Cluster {
return manager.isUserAgentOnConnectEnabled();
}
+ /**
+ * Checks if cluster is configured to use per-message deflate compression
+ */
+ public boolean useCompression() {
+ return manager.useCompression();
+ }
+
public final static class Builder {
private List<InetAddress> addresses = new ArrayList<>();
private int port = 8182;
@@ -615,6 +623,7 @@ public final class Cluster {
private AuthProperties authProps = new AuthProperties();
private long connectionSetupTimeoutMillis = Connection.CONNECTION_SETUP_TIMEOUT_MILLIS;
private boolean enableUserAgentOnConnect = true;
+ private boolean useCompression = true;
private Builder() {
// empty to prevent direct instantiation
@@ -1031,6 +1040,14 @@ public final class Cluster {
return this;
}
+ /**
+ * Configures use of per-message deflate compression. Defaults to true.
+ */
+ public Builder useCompression(final boolean useCompression) {
+ this.useCompression = useCompression;
+ return this;
+ }
+
List<InetSocketAddress> getContactPoints() {
return addresses.stream().map(addy -> new InetSocketAddress(addy, port)).collect(Collectors.toList());
}
@@ -1100,6 +1117,7 @@ public final class Cluster {
private final int port;
private final String path;
private final boolean enableUserAgentOnConnect;
+ private final boolean useCompression;
private final AtomicReference<CompletableFuture<Void>> closeFuture = new AtomicReference<>();
@@ -1113,6 +1131,7 @@ public final class Cluster {
this.contactPoints = builder.getContactPoints();
this.interceptor = builder.interceptor;
this.enableUserAgentOnConnect = builder.enableUserAgentOnConnect;
+ this.useCompression = builder.useCompression;
connectionPoolSettings = new Settings.ConnectionPoolSettings();
connectionPoolSettings.maxInProcessPerConnection = builder.maxInProcessPerConnection;
@@ -1307,5 +1326,12 @@ public final class Cluster {
public boolean isUserAgentOnConnectEnabled() {
return enableUserAgentOnConnect;
}
+
+ /**
+ * Checks if cluster is configured to use per-message deflate compression
+ */
+ public boolean useCompression() {
+ return useCompression;
+ }
}
}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
index db79fe065f..749789a86c 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
@@ -215,7 +215,7 @@ final class Connection {
throw new IllegalStateException(String.format("There is already a request pending with an id of: %s", requestMessage.getRequestId()));
// once there is a completed write, then create a traverser for the result set and complete
- // the promise so that the client knows that that it can start checking for results.
+ // the promise so that the client knows that it can start checking for results.
final Connection thisConnection = this;
final ChannelPromise requestPromise = channel.newPromise()
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Settings.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Settings.java
index 0845dbf547..e8005f8b68 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Settings.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Settings.java
@@ -105,6 +105,11 @@ final class Settings {
*/
public boolean enableUserAgentOnConnect = true;
+ /**
+ * Configures use of per-message deflate compression. Defaults to true.
+ */
+ public boolean useCompression = true;
+
/**
* Read configuration from a file into a new {@link Settings} object.
*
@@ -154,6 +159,9 @@ final class Settings {
if (conf.containsKey("enableUserAgentOnConnect"))
settings.enableUserAgentOnConnect = conf.getBoolean("enableUserAgentOnConnect");
+ if (conf.containsKey("useCompression"))
+ settings.useCompression = conf.getBoolean("useCompression");
+
if (conf.containsKey("hosts"))
settings.hosts = conf.getList("hosts").stream().map(Object::toString).collect(Collectors.toList());
diff --git a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/SettingsTest.java b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/SettingsTest.java
index 2f74829bca..6a7b2a004a 100644
--- a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/SettingsTest.java
+++ b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/SettingsTest.java
@@ -47,6 +47,7 @@ public class SettingsTest {
conf.setProperty("serializer.className", "my.serializers.MySerializer");
conf.setProperty("serializer.config.any", "thing");
conf.setProperty("enableUserAgentOnConnect", false);
+ conf.setProperty("useCompression", false);
conf.setProperty("connectionPool.enableSsl", true);
conf.setProperty("connectionPool.keyStore", "server.jks");
conf.setProperty("connectionPool.keyStorePassword", "password2");
@@ -84,6 +85,7 @@ public class SettingsTest {
assertEquals("my.serializers.MySerializer", settings.serializer.className);
assertEquals("thing", settings.serializer.config.get("any"));
assertEquals(false, settings.enableUserAgentOnConnect);
+ assertEquals(false, settings.useCompression);
assertThat(settings.connectionPool.enableSsl, is(true));
assertEquals("server.jks", settings.connectionPool.keyStore);
assertEquals("password2", settings.connectionPool.keyStorePassword);
diff --git a/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/driver/client.js b/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/driver/client.js
index acd71b504d..c781d9c69e 100644
--- a/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/driver/client.js
+++ b/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/driver/client.js
@@ -44,6 +44,7 @@ class Client {
* @param {String} [options.processor] The name of the opProcessor to use, leave it undefined or set 'session' when session mode.
* @param {String} [options.session] The sessionId of Client in session mode. Defaults to null means session-less Client.
* @param {http.Agent} [options.agent] The http.Agent implementation to use.
+ * @param {Boolean} [options.useCompression] Enable per-message deflate compression. Defaults to: false.
* @constructor
*/
constructor(url, options = {}) {
diff --git a/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/driver/connection.js b/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/driver/connection.js
index 5a39a773a0..6f9c2fee03 100644
--- a/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/driver/connection.js
+++ b/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/driver/connection.js
@@ -65,6 +65,7 @@ class Connection extends EventEmitter {
* @param {Object} [options.headers] An associative array containing the additional header key/values for the initial request.
* @param {Boolean} [options.enableUserAgentOnConnect] Determines if a user agent will be sent during connection handshake. Defaults to: true
* @param {http.Agent} [options.agent] The http.Agent implementation to use.
+ * @param {Boolean} [options.useCompression] Enable per-message deflate compression. Defaults to: false.
* @constructor
*/
constructor(url, options) {
@@ -135,6 +136,7 @@ class Connection extends EventEmitter {
pfx: this.options.pfx,
rejectUnauthorized: this.options.rejectUnauthorized,
agent: this.options.agent,
+ perMessageDeflate: this.options.useCompression,
}
: undefined,
);
diff --git a/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/driver/driver-remote-connection.js b/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/driver/driver-remote-connection.js
index 25c7cf0e4d..9b60593b81 100644
--- a/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/driver/driver-remote-connection.js
+++ b/gremlin-javascript/src/main/javascript/gremlin-javascript/lib/driver/driver-remote-connection.js
@@ -50,6 +50,7 @@ class DriverRemoteConnection extends RemoteConnection {
* @param {Object} [options.headers] An associative array containing the additional header key/values for the initial request.
* @param {Boolean} [options.enableUserAgentOnConnect] Determines if a user agent will be sent during connection handshake. Defaults to: true
* @param {http.Agent} [options.agent] The http.Agent implementation to use.
+ * @param {Boolean} [options.useCompression] Enable per-message deflate compression. Defaults to: false.
* @constructor
*/
constructor(url, options = {}) {
diff --git a/gremlin-python/src/main/python/gremlin_python/driver/client.py b/gremlin-python/src/main/python/gremlin_python/driver/client.py
index 74d90a32d0..a88d893075 100644
--- a/gremlin-python/src/main/python/gremlin_python/driver/client.py
+++ b/gremlin-python/src/main/python/gremlin_python/driver/client.py
@@ -44,7 +44,8 @@ class Client:
transport_factory=None, pool_size=None, max_workers=None,
message_serializer=None, username="", password="",
kerberized_service="", headers=None, session=None,
- enable_user_agent_on_connect=True, **transport_kwargs):
+ enable_user_agent_on_connect=True, use_compression=False,
+ **transport_kwargs):
log.info("Creating Client with url '%s'", url)
# check via url that we are using http protocol
@@ -55,8 +56,11 @@ class Client:
self._headers = headers
self._enable_user_agent_on_connect = enable_user_agent_on_connect
self._traversal_source = traversal_source
+ self._use_compression = use_compression
if not self._use_http and "max_content_length" not in transport_kwargs:
transport_kwargs["max_content_length"] = 10 * 1024 * 1024
+ if self._use_compression and "compress" not in transport_kwargs:
+ transport_kwargs["compress"] = 15 # enable per-message deflate compression with max 32k sliding window size
if message_serializer is None:
message_serializer = serializer.GraphBinarySerializersV1()
diff --git a/gremlin-python/src/main/python/gremlin_python/driver/driver_remote_connection.py b/gremlin-python/src/main/python/gremlin_python/driver/driver_remote_connection.py
index 6058a6ba4b..f4be58c110 100644
--- a/gremlin-python/src/main/python/gremlin_python/driver/driver_remote_connection.py
+++ b/gremlin-python/src/main/python/gremlin_python/driver/driver_remote_connection.py
@@ -39,7 +39,7 @@ class DriverRemoteConnection(RemoteConnection):
username="", password="", kerberized_service='',
message_serializer=None, graphson_reader=None,
graphson_writer=None, headers=None, session=None,
- enable_user_agent_on_connect=True, **transport_kwargs):
+ enable_user_agent_on_connect=True, use_compression=False, **transport_kwargs):
log.info("Creating DriverRemoteConnection with url '%s'", str(url))
self.__url = url
self.__traversal_source = traversal_source
@@ -56,6 +56,7 @@ class DriverRemoteConnection(RemoteConnection):
self.__headers = headers
self.__session = session
self.__enable_user_agent_on_connect = enable_user_agent_on_connect
+ self.__use_compression = use_compression
self.__transport_kwargs = transport_kwargs
# keeps a list of sessions that have been spawned from this DriverRemoteConnection
@@ -78,6 +79,7 @@ class DriverRemoteConnection(RemoteConnection):
headers=headers,
session=session,
enable_user_agent_on_connect=enable_user_agent_on_connect,
+ use_compression=use_compression,
**transport_kwargs)
self._url = self._client._url
self._traversal_source = self._client._traversal_source
diff --git a/gremlin-python/src/main/python/tests/driver/test_driver_remote_connection.py b/gremlin-python/src/main/python/tests/driver/test_driver_remote_connection.py
index f6feff1ffc..da41121fc1 100644
--- a/gremlin-python/src/main/python/tests/driver/test_driver_remote_connection.py
+++ b/gremlin-python/src/main/python/tests/driver/test_driver_remote_connection.py
@@ -19,6 +19,7 @@
import os
from gremlin_python import statics
+from gremlin_python.driver.driver_remote_connection import DriverRemoteConnection
from gremlin_python.driver.protocol import GremlinServerError
from gremlin_python.statics import long
from gremlin_python.process.traversal import Traverser
@@ -34,6 +35,8 @@ from gremlin_python.driver.serializer import GraphSONSerializersV2d0
__author__ = 'Marko A. Rodriguez (http://markorodriguez.com)'
+from tests.conftest import anonymous_url
+
gremlin_server_url = os.environ.get('GREMLIN_SERVER_URL', 'ws://localhost:{}/gremlin')
test_no_auth_url = gremlin_server_url.format(45940)
@@ -286,3 +289,22 @@ class TestDriverRemoteConnection(object):
g = traversal().withRemote(remote_connection_authenticated)
assert long(6) == g.V().count().toList()[0]
+
+ def test_should_propagate_use_compression(self):
+ # Use Compression
+ drc = DriverRemoteConnection(anonymous_url, use_compression=True)
+
+ aiohttp_kwargs = drc._client._transport_factory()._aiohttp_kwargs
+ assert aiohttp_kwargs.get("compress") != 0
+
+ # Disable Compression
+ drc = DriverRemoteConnection(anonymous_url, use_compression=False)
+
+ aiohttp_kwargs = drc._client._transport_factory()._aiohttp_kwargs
+ assert aiohttp_kwargs.get("compress") is None or aiohttp_kwargs.get("compress") == 0
+
+ # Default (no compression)
+ drc = DriverRemoteConnection(anonymous_url)
+
+ aiohttp_kwargs = drc._client._transport_factory()._aiohttp_kwargs
+ assert aiohttp_kwargs.get("compress") is None or aiohttp_kwargs.get("compress") == 0