You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by xi...@apache.org on 2024/10/16 00:53:38 UTC
(tinkerpop) branch master-http updated: Update response message serialization with `{bulked}` flag to indicate bulked result data (#2826)
This is an automated email from the ASF dual-hosted git repository.
xiazcy pushed a commit to branch master-http
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
The following commit(s) were added to refs/heads/master-http by this push:
new 3389bdf66d Update response message serialization with `{bulked}` flag to indicate bulked result data (#2826)
3389bdf66d is described below
commit 3389bdf66d29ce8c04dc283c49272ffdbb191d63
Author: Yang Xia <55...@users.noreply.github.com>
AuthorDate: Tue Oct 15 17:53:33 2024 -0700
Update response message serialization with `{bulked}` flag to indicate bulked result data (#2826)
Added a `bulked` header set by cluster setting, as well as a with `bulked` request option to turn on the bulking of result data.
---
CHANGELOG.asciidoc | 2 +
.../structure/io/binary/GraphBinaryWriter.java | 1 +
.../tinkerpop/gremlin/driver/Channelizer.java | 2 +-
.../apache/tinkerpop/gremlin/driver/Client.java | 1 +
.../apache/tinkerpop/gremlin/driver/Cluster.java | 27 +++++++++++
.../tinkerpop/gremlin/driver/RequestOptions.java | 17 +++++++
.../apache/tinkerpop/gremlin/driver/Settings.java | 8 ++++
.../driver/handler/GremlinResponseHandler.java | 15 +++++-
.../driver/handler/HttpGremlinRequestEncoder.java | 8 +++-
.../handler/HttpGremlinResponseStreamDecoder.java | 5 ++
.../driver/remote/DriverRemoteTraversal.java | 5 +-
.../gremlin/driver/simple/SimpleHttpClient.java | 2 +-
.../tinkerpop/gremlin/driver/SettingsTest.java | 3 ++
.../HttpGremlinResponseStreamDecoderTest.java | 37 ++++++++++++++-
gremlin-python/docker-compose.yml | 1 +
.../main/python/gremlin_python/driver/client.py | 3 +-
.../python/gremlin_python/driver/connection.py | 8 +++-
.../driver/driver_remote_connection.py | 4 +-
.../main/python/gremlin_python/driver/request.py | 2 +-
.../python/gremlin_python/driver/serializer.py | 21 +++++++--
.../python/gremlin_python/process/traversal.py | 21 ++++-----
gremlin-python/src/main/python/radish/terrain.py | 4 +-
.../tests/driver/test_driver_remote_connection.py | 7 +++
.../server/handler/HttpGremlinEndpointHandler.java | 53 +++++++++++++--------
.../GraphBinaryLangBulkedRemoteFeatureTest.java | 51 ++++++++++++++++++++
.../gremlin/driver/remote/RemoteWorld.java | 10 ++++
.../gremlin/server/GremlinServerIntegrateTest.java | 54 ++++++++++++++++++++--
.../io.cucumber.core.backend.ObjectFactory | 1 +
.../org/apache/tinkerpop/gremlin/util/Tokens.java | 7 +++
.../gremlin/util/message/RequestMessage.java | 5 ++
.../gremlin/util/message/ResponseMessage.java | 8 +++-
.../gremlin/util/message/ResponseResult.java | 8 +++-
.../util/ser/GraphBinaryMessageSerializerV4.java | 9 ++++
.../util/ser/binary/RequestMessageSerializer.java | 3 ++
34 files changed, 363 insertions(+), 50 deletions(-)
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 95ab1c2628..9c018a6406 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -66,6 +66,8 @@ image::https://raw.githubusercontent.com/apache/tinkerpop/master/docs/static/ima
* Add `DateTime` serializer for Java and Python according to GraphBinaryV4 IO spec.
* Replaced `Date` with `OffsetDateTime` in Java core.
* Update serializers for `label` as a singleton list of string according to GraphBinaryV4 IO spec.
+* Added `bulked` byte to `Response Message` serialization according to GraphBinaryV4 IO spec.
+* Added a `bulked` header set by cluster setting, as well as a with `bulked` request option to turn on the bulking of result data.
* The `maxContentLength` setting for Gremlin Driver has been renamed to `maxResponseContentLength` and now blocks incoming responses that are too large based on total response size.
* The `maxContentLength` setting for Gremlin Server has been renamed to `maxRequestContentLength`.
* Add missing strategies to the `TraversalStrategies` global cache as well as `CoreImports` in `gremlin-groovy`.
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/binary/GraphBinaryWriter.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/binary/GraphBinaryWriter.java
index 53a2a55868..0a24c1058c 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/binary/GraphBinaryWriter.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/binary/GraphBinaryWriter.java
@@ -47,6 +47,7 @@ public class GraphBinaryWriter {
private final static byte VALUE_FLAG_NONE = 0;
private final static byte VALUE_FLAG_ORDERED = 2;
public final static byte VERSION_BYTE = (byte)0x81;
+ public final static byte BULKED_BYTE = (byte)0x01;
private final static byte[] unspecifiedNullBytes = new byte[] { DataType.UNSPECIFIED_NULL.getCodeByte(), 0x01};
private final static byte[] customTypeCodeBytes = new byte[] { DataType.CUSTOM.getCodeByte() };
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 43c1769c2d..950699f17d 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
@@ -191,7 +191,7 @@ public interface Channelizer extends ChannelHandler {
super.init(connection);
httpCompressionDecoder = new HttpContentDecompressionHandler();
- gremlinRequestEncoder = new HttpGremlinRequestEncoder(cluster.getSerializer(), cluster.getRequestInterceptor(), cluster.isUserAgentOnConnectEnabled());
+ gremlinRequestEncoder = new HttpGremlinRequestEncoder(cluster.getSerializer(), cluster.getRequestInterceptor(), cluster.isUserAgentOnConnectEnabled(), cluster.isBulkingEnabled());
gremlinResponseDecoder = new HttpGremlinResponseStreamDecoder(cluster.getSerializer(), cluster.getMaxResponseContentLength());
}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Client.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Client.java
index 15cb93f570..ebd03ed359 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Client.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Client.java
@@ -238,6 +238,7 @@ public abstract class Client {
options.getG().ifPresent(g -> request.addG(g));
options.getLanguage().ifPresent(lang -> request.addLanguage(lang));
options.getMaterializeProperties().ifPresent(mp -> request.addMaterializeProperties(mp));
+ options.getBulked().ifPresent(bulked -> request.addBulkedResult(Boolean.parseBoolean(bulked)));
return submitAsync(request.create());
}
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 59593d0f14..bfca36f9aa 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
@@ -180,6 +180,7 @@ public final class Cluster {
.minConnectionPoolSize(settings.connectionPool.minSize)
.connectionSetupTimeoutMillis(settings.connectionPool.connectionSetupTimeoutMillis)
.enableUserAgentOnConnect(settings.enableUserAgentOnConnect)
+ .enableBulkedResult(settings.enableBulkedResult)
.validationRequest(settings.connectionPool.validationRequest);
if (!settings.auth.type.isEmpty()) {
@@ -464,6 +465,13 @@ public final class Cluster {
return manager.isUserAgentOnConnectEnabled();
}
+ /**
+ * Checks if cluster is configured to send bulked results
+ */
+ public boolean isBulkingEnabled() {
+ return manager.isBulkedResultEnabled();
+ }
+
public final static class Builder {
private final List<InetAddress> addresses = new ArrayList<>();
private int port = 8182;
@@ -494,6 +502,7 @@ public final class Cluster {
private List<RequestInterceptor> interceptors = new ArrayList<>();
private long connectionSetupTimeoutMillis = Connection.CONNECTION_SETUP_TIMEOUT_MILLIS;
private boolean enableUserAgentOnConnect = true;
+ private boolean enableBulkedResult = false;
private Builder() {
// empty to prevent direct instantiation
@@ -797,6 +806,15 @@ public final class Cluster {
return this;
}
+ /**
+ * Configures whether cluster will enable result bulking to optimize performance.
+ * @param enableBulkedResult true enables bulking.
+ */
+ public Builder enableBulkedResult(final boolean enableBulkedResult) {
+ this.enableBulkedResult = enableBulkedResult;
+ return this;
+ }
+
List<InetSocketAddress> getContactPoints() {
return addresses.stream().map(addy -> new InetSocketAddress(addy, port)).collect(Collectors.toList());
}
@@ -865,6 +883,7 @@ public final class Cluster {
private final int port;
private final String path;
private final boolean enableUserAgentOnConnect;
+ private final boolean enableBulkedResult;
private final AtomicReference<CompletableFuture<Void>> closeFuture = new AtomicReference<>();
@@ -877,6 +896,7 @@ public final class Cluster {
this.contactPoints = builder.getContactPoints();
this.interceptor = builder.interceptors;
this.enableUserAgentOnConnect = builder.enableUserAgentOnConnect;
+ this.enableBulkedResult = builder.enableBulkedResult;
connectionPoolSettings = new Settings.ConnectionPoolSettings();
connectionPoolSettings.maxSize = builder.maxConnectionPoolSize;
@@ -1040,5 +1060,12 @@ public final class Cluster {
public boolean isUserAgentOnConnectEnabled() {
return enableUserAgentOnConnect;
}
+
+ /**
+ * Checks if cluster is configured to send bulked results
+ */
+ public boolean isBulkedResultEnabled() {
+ return enableBulkedResult;
+ }
}
}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/RequestOptions.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/RequestOptions.java
index ef32af2d00..f5e8fcfef9 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/RequestOptions.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/RequestOptions.java
@@ -28,6 +28,7 @@ import java.util.Map;
import java.util.Optional;
import static org.apache.tinkerpop.gremlin.util.Tokens.ARGS_BATCH_SIZE;
+import static org.apache.tinkerpop.gremlin.util.Tokens.BULKED;
import static org.apache.tinkerpop.gremlin.util.Tokens.ARGS_EVAL_TIMEOUT;
import static org.apache.tinkerpop.gremlin.util.Tokens.ARGS_G;
import static org.apache.tinkerpop.gremlin.util.Tokens.ARGS_LANGUAGE;
@@ -48,6 +49,7 @@ public final class RequestOptions {
private final Long timeout;
private final String language;
private final String materializeProperties;
+ private final String bulkedResult;
private RequestOptions(final Builder builder) {
this.graphOrTraversalSource = builder.graphOrTraversalSource;
@@ -56,6 +58,7 @@ public final class RequestOptions {
this.timeout = builder.timeout;
this.language = builder.language;
this.materializeProperties = builder.materializeProperties;
+ this.bulkedResult = builder.bulkedResult;
}
public Optional<String> getG() {
@@ -80,6 +83,8 @@ public final class RequestOptions {
public Optional<String> getMaterializeProperties() { return Optional.ofNullable(materializeProperties); }
+ public Optional<String> getBulked() { return Optional.ofNullable(bulkedResult); }
+
public static Builder build() {
return new Builder();
}
@@ -98,6 +103,9 @@ public final class RequestOptions {
builder.materializeProperties((String) options.get(ARGS_MATERIALIZE_PROPERTIES));
if (options.containsKey(ARGS_LANGUAGE))
builder.language((String) options.get(ARGS_LANGUAGE));
+ if (options.containsKey(BULKED))
+ builder.withBulkedResult((boolean) options.get(BULKED));
+
}
final Map<String, Object> parameters = gremlinLang.getParameters();
@@ -114,6 +122,7 @@ public final class RequestOptions {
private Long timeout = null;
private String materializeProperties = null;
private String language = null;
+ private String bulkedResult;
/**
* The aliases to set on the request.
@@ -177,6 +186,14 @@ public final class RequestOptions {
return this;
}
+ /**
+ * Enables or disables bulked result on the server.
+ */
+ public Builder withBulkedResult(final boolean bulked) {
+ this.bulkedResult = String.valueOf(bulked);
+ return this;
+ }
+
public RequestOptions create() {
return new RequestOptions(this);
}
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 58933806da..3f5f5e6f4a 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
@@ -90,6 +90,11 @@ public final class Settings {
*/
public boolean enableUserAgentOnConnect = true;
+ /**
+ * Toggles if result from server is bulked. Default is false.
+ */
+ public boolean enableBulkedResult = false;
+
/**
* Read configuration from a file into a new {@link Settings} object.
*
@@ -127,6 +132,9 @@ public final class Settings {
if (conf.containsKey("enableUserAgentOnConnect"))
settings.enableUserAgentOnConnect = conf.getBoolean("enableUserAgentOnConnect");
+ if (conf.containsKey("enableBulkedResult"))
+ settings.enableBulkedResult = conf.getBoolean("enableBulkedResult");
+
if (conf.containsKey("hosts"))
settings.hosts = conf.getList("hosts").stream().map(Object::toString).collect(Collectors.toList());
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/GremlinResponseHandler.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/GremlinResponseHandler.java
index c10b43823a..d32acecd2e 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/GremlinResponseHandler.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/GremlinResponseHandler.java
@@ -27,6 +27,7 @@ import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.tinkerpop.gremlin.driver.Result;
import org.apache.tinkerpop.gremlin.driver.ResultQueue;
import org.apache.tinkerpop.gremlin.driver.exception.ResponseException;
+import org.apache.tinkerpop.gremlin.process.remote.traversal.DefaultRemoteTraverser;
import org.apache.tinkerpop.gremlin.util.ExceptionHelper;
import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
import org.apache.tinkerpop.gremlin.util.message.ResponseMessage;
@@ -34,6 +35,7 @@ import org.apache.tinkerpop.gremlin.util.ser.SerializationException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.Iterator;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;
@@ -72,8 +74,19 @@ public class GremlinResponseHandler extends SimpleChannelInboundHandler<Response
if ((null == statusCode) || (statusCode == HttpResponseStatus.OK)) {
final List<Object> data = response.getResult().getData();
+ final boolean bulked = channelHandlerContext.channel().attr(HttpGremlinResponseStreamDecoder.IS_BULKED).get();
// unrolls the collection into individual results to be handled by the queue.
- data.forEach(item -> queue.add(new Result(item)));
+ if (bulked) {
+ for (Iterator<Object> iter = data.iterator(); iter.hasNext(); ) {
+ final Object obj = iter.next();
+ final long bulk = (long) iter.next();
+ DefaultRemoteTraverser<Object> item = new DefaultRemoteTraverser<>(obj, bulk);
+ queue.add(new Result(item));
+ }
+ } else {
+ data.forEach(item -> queue.add(new Result(item)));
+ }
+
} else {
// this is a "success" but represents no results otherwise it is an error
if (statusCode != HttpResponseStatus.NO_CONTENT) {
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinRequestEncoder.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinRequestEncoder.java
index 9d130ab4cd..4e08f6def7 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinRequestEncoder.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinRequestEncoder.java
@@ -36,6 +36,7 @@ import org.apache.tinkerpop.gremlin.driver.auth.Auth.AuthenticationException;
import org.apache.tinkerpop.gremlin.driver.exception.ResponseException;
import org.apache.tinkerpop.gremlin.process.traversal.GremlinLang;
import org.apache.tinkerpop.gremlin.util.MessageSerializer;
+import org.apache.tinkerpop.gremlin.util.Tokens;
import org.apache.tinkerpop.gremlin.util.message.RequestMessage;
import org.apache.tinkerpop.gremlin.util.ser.SerializationException;
@@ -52,12 +53,14 @@ public final class HttpGremlinRequestEncoder extends MessageToMessageEncoder<Req
private final MessageSerializer<?> serializer;
private final boolean userAgentEnabled;
+ private final boolean bulkedResultEnabled;
private final List<RequestInterceptor> interceptors;
- public HttpGremlinRequestEncoder(final MessageSerializer<?> serializer, final List<RequestInterceptor> interceptors, boolean userAgentEnabled) {
+ public HttpGremlinRequestEncoder(final MessageSerializer<?> serializer, final List<RequestInterceptor> interceptors, boolean userAgentEnabled, boolean bulkedResultEnabled) {
this.serializer = serializer;
this.interceptors = interceptors;
this.userAgentEnabled = userAgentEnabled;
+ this.bulkedResultEnabled = bulkedResultEnabled;
}
@Override
@@ -81,6 +84,9 @@ public final class HttpGremlinRequestEncoder extends MessageToMessageEncoder<Req
if (userAgentEnabled) {
request.headers().add(HttpHeaderNames.USER_AGENT, UserAgent.USER_AGENT);
}
+ if (bulkedResultEnabled) {
+ request.headers().add(Tokens.BULKED, "true");
+ }
for (final RequestInterceptor interceptor : interceptors) {
request = interceptor.apply(request);
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinResponseStreamDecoder.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinResponseStreamDecoder.java
index 5c8c38edf9..36f08e5d3b 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinResponseStreamDecoder.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinResponseStreamDecoder.java
@@ -45,6 +45,7 @@ import static org.apache.tinkerpop.gremlin.driver.Channelizer.HttpChannelizer.LA
public class HttpGremlinResponseStreamDecoder extends MessageToMessageDecoder<DefaultHttpObject> {
+ public static final AttributeKey<Boolean> IS_BULKED = AttributeKey.valueOf("isBulked");
private static final AttributeKey<Boolean> IS_FIRST_CHUNK = AttributeKey.valueOf("isFirstChunk");
private static final AttributeKey<HttpResponseStatus> RESPONSE_STATUS = AttributeKey.valueOf("responseStatus");
private static final AttributeKey<String> RESPONSE_ENCODING = AttributeKey.valueOf("responseSerializer");
@@ -62,6 +63,7 @@ public class HttpGremlinResponseStreamDecoder extends MessageToMessageDecoder<De
@Override
protected void decode(ChannelHandlerContext ctx, DefaultHttpObject msg, List<Object> out) throws Exception {
final Attribute<Boolean> isFirstChunk = ((AttributeMap) ctx).attr(IS_FIRST_CHUNK);
+ final Attribute<Boolean> isBulked = ((AttributeMap) ctx).attr(IS_BULKED);
final Attribute<HttpResponseStatus> responseStatus = ((AttributeMap) ctx).attr(RESPONSE_STATUS);
final Attribute<String> responseEncoding = ((AttributeMap) ctx).attr(RESPONSE_ENCODING);
@@ -95,6 +97,9 @@ public class HttpGremlinResponseStreamDecoder extends MessageToMessageDecoder<De
out.add(response);
} else {
final ResponseMessage chunk = serializer.readChunk(content, isFirstChunk.get());
+ if (isFirstChunk.get()){
+ isBulked.set(chunk.getResult().isBulked());
+ }
isFirstChunk.set(false);
out.add(chunk);
}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTraversal.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTraversal.java
index 4c2dc41225..6521f1ec97 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTraversal.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTraversal.java
@@ -113,7 +113,10 @@ public class DriverRemoteTraversal<S, E> extends AbstractRemoteTraversal<S, E> {
@Override
public Traverser.Admin<E> next() {
- return new DefaultRemoteTraverser<>((E)inner.next().getObject(), 1);
+ final Object next = inner.next().getObject();
+ return next instanceof RemoteTraverser
+ ? (RemoteTraverser<E>) next
+ : new DefaultRemoteTraverser<>((E) next, 1);
}
}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/simple/SimpleHttpClient.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/simple/SimpleHttpClient.java
index fc1d622eb3..8a8cc623f3 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/simple/SimpleHttpClient.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/simple/SimpleHttpClient.java
@@ -105,7 +105,7 @@ public class SimpleHttpClient extends AbstractClient {
new HttpClientCodec(),
new HttpContentDecompressionHandler(),
new HttpGremlinResponseStreamDecoder(serializer, Integer.MAX_VALUE),
- new HttpGremlinRequestEncoder(serializer, new ArrayList<>(), false),
+ new HttpGremlinRequestEncoder(serializer, new ArrayList<>(), false, false),
callbackResponseHandler);
}
});
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 42966bf989..a2d47ff1db 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
@@ -21,6 +21,7 @@ package org.apache.tinkerpop.gremlin.driver;
import static org.hamcrest.core.Is.is;
import static org.junit.Assert.assertEquals;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.assertTrue;
import org.apache.commons.configuration2.BaseConfiguration;
import org.apache.commons.configuration2.Configuration;
@@ -46,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("enableBulkedResult", true);
conf.setProperty("connectionPool.enableSsl", true);
conf.setProperty("connectionPool.keyStore", "server.jks");
conf.setProperty("connectionPool.keyStorePassword", "password2");
@@ -82,6 +84,7 @@ public class SettingsTest {
assertEquals("my.serializers.MySerializer", settings.serializer.className);
assertEquals("thing", settings.serializer.config.get("any"));
assertEquals(false, settings.enableUserAgentOnConnect);
+ assertTrue(settings.enableBulkedResult);
assertThat(settings.connectionPool.enableSsl, is(true));
assertEquals("server.jks", settings.connectionPool.keyStore);
assertEquals("password2", settings.connectionPool.keyStorePassword);
diff --git a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinResponseStreamDecoderTest.java b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinResponseStreamDecoderTest.java
index f2e7be88b2..caa91c3299 100644
--- a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinResponseStreamDecoderTest.java
+++ b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinResponseStreamDecoderTest.java
@@ -30,6 +30,8 @@ import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.HttpVersion;
import java.util.Collections;
+
+import io.netty.util.AttributeKey;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.tinkerpop.gremlin.util.message.ResponseMessage;
import org.apache.tinkerpop.gremlin.util.ser.SerializationException;
@@ -37,6 +39,8 @@ import org.apache.tinkerpop.gremlin.util.ser.Serializers;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public class HttpGremlinResponseStreamDecoderTest {
@@ -83,16 +87,47 @@ public class HttpGremlinResponseStreamDecoderTest {
testChannel.writeInbound(httpResponse);
fail("Expected TooLongFrameException");
} catch (TooLongFrameException e) {
- assertEquals("Response exceeded 59 bytes.", e.getMessage());
+ assertEquals("Response exceeded 60 bytes.", e.getMessage());
}
}
+ @Test
+ public void shouldSetBulkedFlagCtxValueWithEachResponse() throws SerializationException {
+ final EmbeddedChannel testChannel = initializeChannel(Integer.MAX_VALUE);
+
+ final FullHttpResponse httpResponse1 = createBulkedResponse(true);
+ testChannel.writeInbound(httpResponse1);
+ final boolean bulked1 = (boolean) testChannel.pipeline().channel().attr(AttributeKey.valueOf("isBulked")).get();
+ assertTrue(bulked1);
+
+ final FullHttpResponse httpResponse2 = createBulkedResponse(false);
+ testChannel.writeInbound(httpResponse2);
+ final boolean bulked2 = (boolean) testChannel.pipeline().channel().attr(AttributeKey.valueOf("isBulked")).get();
+ assertFalse(bulked2);
+
+ final FullHttpResponse httpResponse3 = createBulkedResponse(true);
+ testChannel.writeInbound(httpResponse3);
+ final boolean bulked3 = (boolean) testChannel.pipeline().channel().attr(AttributeKey.valueOf("isBulked")).get();
+ assertTrue(bulked3);
+
+ final FullHttpResponse httpResponse4 = createBulkedResponse(false);
+ testChannel.writeInbound(httpResponse4);
+ final boolean bulked4 = (boolean) testChannel.pipeline().channel().attr(AttributeKey.valueOf("isBulked")).get();
+ assertFalse(bulked4);
+ }
+
private FullHttpResponse createResponse(String content) throws SerializationException {
final ResponseMessage response = ResponseMessage.build().code(HttpResponseStatus.OK).result(Collections.singletonList(content)).create();
final ByteBuf buffer = Serializers.GRAPHBINARY_V4.simpleInstance().serializeResponseAsBinary(response, ByteBufAllocator.DEFAULT);
return new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, buffer, new DefaultHttpHeaders(), new DefaultHttpHeaders());
}
+ private FullHttpResponse createBulkedResponse(final boolean bulkedFlag) throws SerializationException {
+ final ResponseMessage response = ResponseMessage.build().code(HttpResponseStatus.OK).bulked(bulkedFlag).result(Collections.singletonList("test bulked")).create();
+ final ByteBuf buffer = Serializers.GRAPHBINARY_V4.simpleInstance().serializeResponseAsBinary(response, ByteBufAllocator.DEFAULT);
+ return new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, buffer, new DefaultHttpHeaders(), new DefaultHttpHeaders());
+ }
+
private EmbeddedChannel initializeChannel(final long maxResponseContentLength) {
final HttpGremlinResponseStreamDecoder decoder = new HttpGremlinResponseStreamDecoder(Serializers.GRAPHBINARY_V4.simpleInstance(), maxResponseContentLength);
final EmbeddedChannel testChannel = new EmbeddedChannel(new HttpServerCodec(), new HttpObjectAggregator(Integer.MAX_VALUE), decoder);
diff --git a/gremlin-python/docker-compose.yml b/gremlin-python/docker-compose.yml
index 973d946db9..f67f2e9dad 100644
--- a/gremlin-python/docker-compose.yml
+++ b/gremlin-python/docker-compose.yml
@@ -61,6 +61,7 @@ services:
&& python3 ./setup.py build
&& python3 ./setup.py test
&& python3 ./setup.py install
+ && radish -f dots -e -t -b ./radish ./gremlin-test --user-data='serializer=application/vnd.graphbinary-v4.0' --user-data='bulked=true'
&& radish -f dots -e -t -b ./radish ./gremlin-test --user-data='serializer=application/vnd.graphbinary-v4.0';
EXIT_CODE=$$?; chown -R `stat -c "%u:%g" .` .; exit $$EXIT_CODE"
depends_on:
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 16bd6f8270..73fd96588f 100644
--- a/gremlin-python/src/main/python/gremlin_python/driver/client.py
+++ b/gremlin-python/src/main/python/gremlin_python/driver/client.py
@@ -42,13 +42,14 @@ class Client:
def __init__(self, url, traversal_source, protocol_factory=None,
transport_factory=None, pool_size=None, max_workers=None,
message_serializer=None, auth=None, headers=None,
- enable_user_agent_on_connect=True, **transport_kwargs):
+ enable_user_agent_on_connect=True, enable_bulked_result=False, **transport_kwargs):
log.info("Creating Client with url '%s'", url)
self._closed = False
self._url = url
self._headers = headers
self._enable_user_agent_on_connect = enable_user_agent_on_connect
+ self._enable_bulked_result = enable_bulked_result
self._traversal_source = traversal_source
if "max_content_length" not in transport_kwargs:
transport_kwargs["max_content_length"] = 10 * 1024 * 1024
diff --git a/gremlin-python/src/main/python/gremlin_python/driver/connection.py b/gremlin-python/src/main/python/gremlin_python/driver/connection.py
index de9a81034d..5ebfb1cf1a 100644
--- a/gremlin-python/src/main/python/gremlin_python/driver/connection.py
+++ b/gremlin-python/src/main/python/gremlin_python/driver/connection.py
@@ -25,7 +25,8 @@ __author__ = 'David M. Brown (davebshow@gmail.com)'
class Connection:
def __init__(self, url, traversal_source, protocol, transport_factory,
- executor, pool, headers=None, enable_user_agent_on_connect=True):
+ executor, pool, headers=None, enable_user_agent_on_connect=True,
+ enable_bulked_result=False):
self._url = url
self._headers = headers
self._traversal_source = traversal_source
@@ -41,6 +42,11 @@ class Connection:
if self._headers is None:
self._headers = dict()
self._headers[useragent.userAgentHeader] = useragent.userAgent
+ self._enable_bulked_result = enable_bulked_result
+ if self._enable_bulked_result:
+ if self._headers is None:
+ self._headers = dict()
+ self._headers["bulked"] = "true"
def connect(self):
if self._transport:
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 a48dc0087b..dc8b95eb7a 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
@@ -35,7 +35,7 @@ class DriverRemoteConnection(RemoteConnection):
transport_factory=None, pool_size=None, max_workers=None,
auth=None,
message_serializer=None, headers=None,
- enable_user_agent_on_connect=True, **transport_kwargs):
+ enable_user_agent_on_connect=True, enable_bulked_result=False, **transport_kwargs):
log.info("Creating DriverRemoteConnection with url '%s'", str(url))
self.__url = url
self.__traversal_source = traversal_source
@@ -47,6 +47,7 @@ class DriverRemoteConnection(RemoteConnection):
self.__message_serializer = message_serializer
self.__headers = headers
self.__enable_user_agent_on_connect = enable_user_agent_on_connect
+ self.__enable_bulked_result = enable_bulked_result
self.__transport_kwargs = transport_kwargs
if message_serializer is None:
@@ -60,6 +61,7 @@ class DriverRemoteConnection(RemoteConnection):
auth=auth,
headers=headers,
enable_user_agent_on_connect=enable_user_agent_on_connect,
+ enable_bulked_result=enable_bulked_result,
**transport_kwargs)
self._url = self._client._url
self._traversal_source = self._client._traversal_source
diff --git a/gremlin-python/src/main/python/gremlin_python/driver/request.py b/gremlin-python/src/main/python/gremlin_python/driver/request.py
index 5e04f0bb85..db524d6642 100644
--- a/gremlin-python/src/main/python/gremlin_python/driver/request.py
+++ b/gremlin-python/src/main/python/gremlin_python/driver/request.py
@@ -24,4 +24,4 @@ RequestMessage = collections.namedtuple(
'RequestMessage', ['fields', 'gremlin'])
Tokens = ['batchSize', 'bindings', 'g', 'gremlin', 'language',
- 'evaluationTimeout', 'materializeProperties', 'timeoutMs', 'userAgent']
+ 'evaluationTimeout', 'materializeProperties', 'timeoutMs', 'userAgent', 'bulked']
diff --git a/gremlin-python/src/main/python/gremlin_python/driver/serializer.py b/gremlin-python/src/main/python/gremlin_python/driver/serializer.py
index 3fa80a055b..94099f4c5d 100644
--- a/gremlin-python/src/main/python/gremlin_python/driver/serializer.py
+++ b/gremlin-python/src/main/python/gremlin_python/driver/serializer.py
@@ -21,6 +21,9 @@ import base64
import logging
import struct
import io
+
+from gremlin_python.process.traversal import Traverser
+
try:
import ujson as json
if int(json.__version__[0]) < 2:
@@ -58,6 +61,7 @@ class GraphBinarySerializersV4(object):
if not writer:
writer = self.DEFAULT_WRITER_CLASS()
self._graphbinary_writer = writer
+ self._bulked = False
@property
def version(self):
@@ -105,6 +109,7 @@ class GraphBinarySerializersV4(object):
if is_first_chunk:
b.read(1) # version
+ self._bulked = b.read(1)[0] == 0x01
result, readable = self.read_payload(b)
if not readable:
@@ -134,9 +139,17 @@ class GraphBinarySerializersV4(object):
if buffer.tell() == len(buffer.getvalue()):
readable = False
break
- data = self._graphbinary_reader.to_object(buffer)
- if data == Marker.end_of_stream():
- break
- results.append(data)
+ if self._bulked:
+ item = self._graphbinary_reader.to_object(buffer)
+ if item == Marker.end_of_stream():
+ self._bulked = False # no more data expected, reset bulked flag
+ break
+ bulk = self._graphbinary_reader.to_object(buffer)
+ results.append(Traverser(item, bulk))
+ else:
+ data = self._graphbinary_reader.to_object(buffer)
+ if data == Marker.end_of_stream():
+ break
+ results.append(data)
return results, readable
diff --git a/gremlin-python/src/main/python/gremlin_python/process/traversal.py b/gremlin-python/src/main/python/gremlin_python/process/traversal.py
index dce8524b98..8eac3e8349 100644
--- a/gremlin-python/src/main/python/gremlin_python/process/traversal.py
+++ b/gremlin-python/src/main/python/gremlin_python/process/traversal.py
@@ -55,15 +55,16 @@ class Traversal(object):
self.traversal_strategies.apply_strategies(self)
if self.last_traverser is None:
self.last_traverser = next(self.traversers)
- # TODO affected by bulking, revisit after bulking is added back
- # object = self.last_traverser.object
- # self.last_traverser.bulk = self.last_traverser.bulk - 1
- # if self.last_traverser.bulk <= 0:
- # self.last_traverser = None
- # return object
res = self.last_traverser
- self.last_traverser = None
- return res
+ if isinstance(res, Traverser):
+ obj = res.object
+ self.last_traverser.bulk = self.last_traverser.bulk - 1
+ if self.last_traverser.bulk <= 0:
+ self.last_traverser = None
+ return obj
+ else:
+ self.last_traverser = None
+ return res
def toList(self):
warnings.warn(
@@ -100,7 +101,6 @@ class Traversal(object):
DeprecationWarning)
return self.next_traverser()
- # TODO: affected by bulking, revisit after bulking is added back
def next_traverser(self):
if self.traversers is None:
self.traversal_strategies.apply_strategies(self)
@@ -118,7 +118,6 @@ class Traversal(object):
DeprecationWarning)
return self.has_next()
- # TODO: revisit after bulking is added back
def has_next(self):
if self.traversers is None:
self.traversal_strategies.apply_strategies(self)
@@ -127,7 +126,7 @@ class Traversal(object):
self.last_traverser = next(self.traversers)
except StopIteration:
return False
- return not (self.last_traverser is None) # and self.last_traverser.bulk > 0
+ return not (self.last_traverser is None) and self.last_traverser.bulk > 0 if isinstance(self.last_traverser, Traverser) else True
def next(self, amount=None):
if amount is None:
diff --git a/gremlin-python/src/main/python/radish/terrain.py b/gremlin-python/src/main/python/radish/terrain.py
index 51935d1bf9..cc000ef881 100644
--- a/gremlin-python/src/main/python/radish/terrain.py
+++ b/gremlin-python/src/main/python/radish/terrain.py
@@ -99,4 +99,6 @@ def __create_remote(server_graph_name):
else:
raise ValueError('serializer not found - ' + world.config.user_data["serializer"])
- return DriverRemoteConnection(test_no_auth_url, server_graph_name, message_serializer=s)
+ bulked = world.config.user_data["bulked"] == "true" if "bulked" in world.config.user_data else False
+
+ return DriverRemoteConnection(test_no_auth_url, server_graph_name, message_serializer=s, enable_bulked_result=bulked)
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 49f63f606e..607501935a 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
@@ -35,6 +35,13 @@ test_no_auth_url = gremlin_server_url.format(45940)
class TestDriverRemoteConnection(object):
+ def test_bulked_request_option(self, remote_connection):
+ g = traversal().with_(remote_connection)
+ result = g.inject(1,2,3,2,1).to_list()
+ assert 5 == len(result)
+ bulked_results = g.with_("language", "gremlin-lang").with_("bulked", True).inject(1,2,3,2,1).to_list()
+ assert 5 == len(bulked_results)
+
def test_extract_request_options(self, remote_connection):
g = traversal().with_(remote_connection)
t = g.with_("evaluationTimeout", 1000).with_("batchSize", 100).V().count()
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandler.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandler.java
index 622c979fe7..040b1e9edf 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandler.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandler.java
@@ -40,6 +40,7 @@ import org.apache.tinkerpop.gremlin.process.traversal.Pop;
import org.apache.tinkerpop.gremlin.process.traversal.Scope;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalSource;
+import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalInterruptedException;
import org.apache.tinkerpop.gremlin.server.Context;
import org.apache.tinkerpop.gremlin.server.GraphManager;
@@ -76,6 +77,7 @@ import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.UUID;
@@ -311,7 +313,21 @@ public class HttpGremlinEndpointHandler extends SimpleChannelInboundHandler<Requ
final Bindings mergedBindings = mergeBindingsFromRequest(context, new SimpleBindings(graphManager.getAsBindings()));
final Object result = scriptEngine.eval(message.getGremlin(), mergedBindings);
- handleIterator(context, IteratorUtils.asIterator(result), serializer);
+ final String bulkingSetting = context.getChannelHandlerContext().channel().attr(StateKey.REQUEST_HEADERS).get().get(Tokens.BULKED);
+ // bulking only applies if it's gremlin-lang, and per request token setting takes precedence over header setting.
+ final boolean bulking = Objects.equals(language, "gremlin-lang") ?
+ (args.containsKey(Tokens.BULKED) ?
+ Objects.equals(args.get(Tokens.BULKED), "true") :
+ Objects.equals(bulkingSetting, "true")) :
+ false;
+
+ if (bulking) {
+ // optimization for driver requests
+ ((Traversal.Admin<?, ?>) result).applyStrategies();
+ handleIterator(context, new TraverserIterator((Traversal.Admin<?, ?>) result), serializer, true);
+ } else {
+ handleIterator(context, IteratorUtils.asIterator(result), serializer, false);
+ }
}
@Override
@@ -361,17 +377,7 @@ public class HttpGremlinEndpointHandler extends SimpleChannelInboundHandler<Requ
return bindings;
}
- private void iterateTraversal(final Context context, MessageSerializer<?> serializer, Traversal.Admin<?, ?> traversal)
- throws InterruptedException {
- final UUID requestId = context.getChannelHandlerContext().attr(StateKey.REQUEST_ID).get();
- logger.debug("Traversal request {} for in thread {}", requestId, Thread.currentThread().getName());
-
- // compile the traversal - without it getEndStep() has nothing in it
- traversal.applyStrategies();
- handleIterator(context, new TraverserIterator(traversal), serializer);
- }
-
- private void handleIterator(final Context context, final Iterator itty, final MessageSerializer<?> serializer) throws InterruptedException {
+ private void handleIterator(final Context context, final Iterator itty, final MessageSerializer<?> serializer, final boolean bulking) throws InterruptedException {
final ChannelHandlerContext nettyContext = context.getChannelHandlerContext();
final RequestMessage msg = context.getRequestMessage();
final Settings settings = context.getSettings();
@@ -381,7 +387,7 @@ public class HttpGremlinEndpointHandler extends SimpleChannelInboundHandler<Requ
if (!itty.hasNext()) {
ByteBuf chunk = null;
try {
- chunk = makeChunk(context, msg, serializer, new ArrayList<>(), false);
+ chunk = makeChunk(context, serializer, new ArrayList<>(), false, bulking);
nettyContext.writeAndFlush(new DefaultHttpContent(chunk));
} catch (Exception ex) {
// Bytebuf is a countable release - if it does not get written downstream
@@ -418,7 +424,15 @@ public class HttpGremlinEndpointHandler extends SimpleChannelInboundHandler<Requ
// this could be placed inside the isWriteable() portion of the if-then below but it seems better to
// allow iteration to continue into a batch if that is possible rather than just doing nothing at all
// while waiting for the client to catch up
- if (aggregate.size() < resultIterationBatchSize && itty.hasNext()) aggregate.add(itty.next());
+ if (aggregate.size() < resultIterationBatchSize && itty.hasNext()) {
+ if (bulking) {
+ Traverser traverser = (Traverser) itty.next();
+ aggregate.add(traverser.get());
+ aggregate.add(traverser.bulk());
+ } else {
+ aggregate.add(itty.next());
+ }
+ }
// Don't keep executor busy if client has already given up; there is no way to catch up if the channel is
// not active, and hence we should break the loop.
@@ -438,7 +452,7 @@ public class HttpGremlinEndpointHandler extends SimpleChannelInboundHandler<Requ
if (aggregate.size() == resultIterationBatchSize || !itty.hasNext()) {
ByteBuf chunk = null;
try {
- chunk = makeChunk(context, msg, serializer, aggregate, itty.hasNext());
+ chunk = makeChunk(context, serializer, aggregate, itty.hasNext(), bulking);
} catch (Exception ex) {
// Bytebuf is a countable release - if it does not get written downstream
// it needs to be released here
@@ -505,9 +519,9 @@ public class HttpGremlinEndpointHandler extends SimpleChannelInboundHandler<Requ
return false;
}
- private static ByteBuf makeChunk(final Context ctx, final RequestMessage msg,
- final MessageSerializer<?> serializer, final List<Object> aggregate,
- final boolean hasMore) throws Exception {
+ private static ByteBuf makeChunk(final Context ctx, final MessageSerializer<?> serializer,
+ final List<Object> aggregate, final boolean hasMore,
+ final boolean bulking) throws Exception {
try {
final ChannelHandlerContext nettyContext = ctx.getChannelHandlerContext();
@@ -528,6 +542,8 @@ public class HttpGremlinEndpointHandler extends SimpleChannelInboundHandler<Requ
builder.code(HttpResponseStatus.OK);
}
+ builder.bulked(bulking);
+
responseMessage = builder.create();
}
@@ -541,6 +557,7 @@ public class HttpGremlinEndpointHandler extends SimpleChannelInboundHandler<Requ
return serializer.serializeResponseAsBinary(ResponseMessage.build()
.result(aggregate)
+ .bulked(bulking)
.code(HttpResponseStatus.OK)
.create(), nettyContext.alloc());
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/remote/GraphBinaryLangBulkedRemoteFeatureTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/remote/GraphBinaryLangBulkedRemoteFeatureTest.java
new file mode 100644
index 0000000000..1e5b0a7317
--- /dev/null
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/remote/GraphBinaryLangBulkedRemoteFeatureTest.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.tinkerpop.gremlin.driver.remote;
+
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Stage;
+import io.cucumber.guice.CucumberModules;
+import io.cucumber.junit.Cucumber;
+import io.cucumber.junit.CucumberOptions;
+import org.apache.tinkerpop.gremlin.features.AbstractGuiceFactory;
+import org.apache.tinkerpop.gremlin.features.World;
+import org.junit.runner.RunWith;
+
+@RunWith(Cucumber.class)
+@CucumberOptions(
+ tags = "not @RemoteOnly and not @GraphComputerOnly and not @AllowNullPropertyValues",
+ glue = { "org.apache.tinkerpop.gremlin.features" },
+ objectFactory = GraphBinaryLangBulkedRemoteFeatureTest.RemoteGuiceFactory.class,
+ features = { "classpath:/org/apache/tinkerpop/gremlin/test/features" },
+ plugin = {"progress", "junit:target/cucumber.xml"})
+public class GraphBinaryLangBulkedRemoteFeatureTest extends AbstractFeatureTest {
+ public static class RemoteGuiceFactory extends AbstractGuiceFactory {
+ public RemoteGuiceFactory() {
+ super(Guice.createInjector(Stage.PRODUCTION, CucumberModules.createScenarioModule(), new ServiceModule()));
+ }
+ }
+
+ public static final class ServiceModule extends AbstractModule {
+ @Override
+ protected void configure() {
+ bind(World.class).to(RemoteWorld.GraphBinaryLangBulkedRemoteWorld.class);
+ }
+ }
+}
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/remote/RemoteWorld.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/remote/RemoteWorld.java
index b1c66c1d39..84109b080e 100644
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/remote/RemoteWorld.java
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/driver/remote/RemoteWorld.java
@@ -169,6 +169,16 @@ public abstract class RemoteWorld implements World {
}
}
+ public static class GraphBinaryLangBulkedRemoteWorld extends RemoteWorld {
+ public GraphBinaryLangBulkedRemoteWorld() { super(createTestCluster(Serializers.GRAPHBINARY_V4)); }
+
+ @Override
+ public GraphTraversalSource getGraphTraversalSource(final LoadGraphWith.GraphData graphData) {
+ final GraphTraversalSource g = super.getGraphTraversalSource(graphData);
+ return g.with("language", "gremlin-lang").with("bulked", true);
+ }
+ }
+
public static class GraphBinaryGroovyRemoteWorld extends RemoteWorld {
public GraphBinaryGroovyRemoteWorld() { super(createTestCluster(Serializers.GRAPHBINARY_V4)); }
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java
index baa54a5b98..9ee754ce8e 100644
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java
@@ -83,10 +83,7 @@ import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.core.AllOf.allOf;
import static org.hamcrest.core.IsInstanceOf.instanceOf;
import static org.hamcrest.core.StringStartsWith.startsWith;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import static org.junit.Assert.*;
/**
* Integration tests for server-side settings and processing.
@@ -964,4 +961,53 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration
assertEquals(HttpResponseStatus.INTERNAL_SERVER_ERROR, ((ResponseException) t).getResponseStatusCode());
}
}
+
+ @Test
+ public void shouldEnableBulkedResultWithClusterOptions() {
+ final Cluster cluster = TestClientFactory.build().enableBulkedResult(true).create();
+ try {
+ final GraphTraversalSource g = traversal().with(DriverRemoteConnection.using(cluster));
+ final List resultBulked = g.with("language", "gremlin-lang").inject(1, 2, 3, 2, 1).toList();
+ assertTrue(cluster.isBulkingEnabled());
+ assertEquals(5, resultBulked.size());
+
+ // the result itself should not be affected by bulking through the wire
+ final List result = g.with("language", "gremlin-lang").with("bulked", false).inject(1, 2, 3, 2, 1).toList();
+ assertEquals(5, result.size());
+
+ assertTrue(cluster.isBulkingEnabled());
+
+ // the result itself should not be affected by bulking through the wire
+ final List resultBulked2 = g.with("language", "gremlin-lang").with("bulked", true).inject(1, 2, 3, 2, 1).toList();
+ assertEquals(5, resultBulked2.size());
+
+ } finally {
+ cluster.close();
+ }
+ }
+
+ @Test
+ public void shouldEnableBulkedResultWithRequestOptions() {
+ final Cluster cluster = TestClientFactory.build().create();
+ try {
+ final GraphTraversalSource g = traversal().with(DriverRemoteConnection.using(cluster));
+ final List result = g.inject(1, 2, 3, 2, 1).toList();
+ assertEquals(5, result.size());
+ assertFalse(cluster.isBulkingEnabled());
+
+ // the result itself should not be affected by bulking through the wire
+ final List resultBulked = g.with("language", "gremlin-lang").with("bulked", true).inject(1, 2, 3, 2, 1).toList();
+ assertEquals(5, resultBulked.size());
+
+ assertFalse(cluster.isBulkingEnabled());
+
+ // the result itself should not be affected by bulking through the wire
+ final List result2 = g.with("language", "gremlin-lang").with("bulked", false).inject(1, 2, 3, 2, 1).toList();
+ assertEquals(5, result2.size());
+
+ } finally {
+ cluster.close();
+ }
+ }
+
}
diff --git a/gremlin-server/src/test/resources/META-INF/services/io.cucumber.core.backend.ObjectFactory b/gremlin-server/src/test/resources/META-INF/services/io.cucumber.core.backend.ObjectFactory
index 5b6b888d13..7b70b113b0 100644
--- a/gremlin-server/src/test/resources/META-INF/services/io.cucumber.core.backend.ObjectFactory
+++ b/gremlin-server/src/test/resources/META-INF/services/io.cucumber.core.backend.ObjectFactory
@@ -1,3 +1,4 @@
org.apache.tinkerpop.gremlin.driver.remote.GraphBinaryLangRemoteFeatureTest$RemoteGuiceFactory
+org.apache.tinkerpop.gremlin.driver.remote.GraphBinaryLangBulkedRemoteFeatureTest$RemoteGuiceFactory
org.apache.tinkerpop.gremlin.driver.remote.GraphBinaryGroovyRemoteFeatureTest$RemoteGuiceFactory
org.apache.tinkerpop.gremlin.driver.remote.GraphBinaryRemoteComputerFeatureTest$RemoteGuiceFactory
diff --git a/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/Tokens.java b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/Tokens.java
index 79aa5df6d4..d1eba60985 100644
--- a/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/Tokens.java
+++ b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/Tokens.java
@@ -62,6 +62,7 @@ public final class Tokens {
* request to execute on the server.
*/
public static final String ARGS_EVAL_TIMEOUT = "evaluationTimeout";
+
/**
* The name of the argument that allows to control the serialization of properties on the server.
*/
@@ -73,6 +74,7 @@ public final class Tokens {
*/
public static final String MATERIALIZE_PROPERTIES_ALL = "all";
+
/**
* The name of the value denoting that only `ID` and `Label` of Element should be returned.
* Should be used with {@code ARGS_MATERIALIZE_PROPERTIES}
@@ -84,6 +86,11 @@ public final class Tokens {
*/
public static final String TIMEOUT_MS = "timeoutMs";
+ /**
+ * The key for turning on bulked result optimization for driver requests.
+ */
+ public static final String BULKED = "bulked";
+
/**
* A value that is a custom string that the user can pass to a server that might accept it for purpose of
* identifying the kind of client it came from.
diff --git a/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/message/RequestMessage.java b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/message/RequestMessage.java
index 16c168b304..31c10ade67 100644
--- a/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/message/RequestMessage.java
+++ b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/message/RequestMessage.java
@@ -161,6 +161,11 @@ public final class RequestMessage {
return this;
}
+ public Builder addBulkedResult(final boolean bulked) {
+ this.fields.put(Tokens.BULKED, String.valueOf(bulked));
+ return this;
+ }
+
/**
* Create the request message given the settings provided to the {@link Builder}.
*/
diff --git a/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/message/ResponseMessage.java b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/message/ResponseMessage.java
index dc502cee44..6251f6d613 100644
--- a/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/message/ResponseMessage.java
+++ b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/message/ResponseMessage.java
@@ -99,6 +99,7 @@ public final class ResponseMessage {
public final static class Builder {
private HttpResponseStatus code = null;
private List<Object> result = Collections.emptyList();
+ private boolean bulked = false;
private String statusMessage = null;
private String exception = null;
@@ -124,8 +125,13 @@ public final class ResponseMessage {
return this;
}
+ public Builder bulked(final boolean bulked) {
+ this.bulked = bulked;
+ return this;
+ }
+
public ResponseMessage create() {
- final ResponseResult responseResult = new ResponseResult(result);
+ final ResponseResult responseResult = new ResponseResult(result, bulked);
// skip null values
if (code == null && statusMessage == null) {
return new ResponseMessage(null, responseResult);
diff --git a/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/message/ResponseResult.java b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/message/ResponseResult.java
index 6f7ad8fe6e..3d8ec519b3 100644
--- a/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/message/ResponseResult.java
+++ b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/message/ResponseResult.java
@@ -25,8 +25,10 @@ import java.util.List;
*/
public final class ResponseResult {
private final List<Object> data;
+ private final boolean bulked;
- public ResponseResult(final List<Object> data) {
+ public ResponseResult(final List<Object> data, final boolean bulked) {
+ this.bulked = bulked;
this.data = data;
}
@@ -34,6 +36,10 @@ public final class ResponseResult {
return data;
}
+ public boolean isBulked() {
+ return bulked;
+ }
+
@Override
public String toString() {
return "ResponseResult{" +
diff --git a/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/GraphBinaryMessageSerializerV4.java b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/GraphBinaryMessageSerializerV4.java
index 63ed101c0c..e26d02a017 100644
--- a/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/GraphBinaryMessageSerializerV4.java
+++ b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/GraphBinaryMessageSerializerV4.java
@@ -221,6 +221,11 @@ public class GraphBinaryMessageSerializerV4 extends AbstractMessageSerializer<Gr
if (parts.contains(MessageParts.HEADER)) {
// Version
buffer.writeByte(GraphBinaryWriter.VERSION_BYTE);
+ if (responseMessage.getResult().isBulked()) {
+ buffer.writeByte(GraphBinaryWriter.BULKED_BYTE);
+ } else {
+ buffer.writeByte((byte) 0);
+ }
}
if (parts.contains(MessageParts.DATA)) {
@@ -282,6 +287,7 @@ public class GraphBinaryMessageSerializerV4 extends AbstractMessageSerializer<Gr
@Override
public ResponseMessage readChunk(final ByteBuf byteBuf, final boolean isFirstChunk) throws SerializationException {
final Buffer buffer = bufferFactory.create(byteBuf);
+ boolean bulking = false;
try {
// empty input buffer
@@ -297,6 +303,7 @@ public class GraphBinaryMessageSerializerV4 extends AbstractMessageSerializer<Gr
// Or the buffer offsets are wrong
throw new SerializationException("The most significant bit should be set according to the format");
}
+ bulking = (buffer.readByte() & 1) == 1;
}
final List<Object> result = readPayload(buffer);
@@ -305,12 +312,14 @@ public class GraphBinaryMessageSerializerV4 extends AbstractMessageSerializer<Gr
if (buffer.readableBytes() == 0) {
return ResponseMessage.build()
.result(result)
+ .bulked(bulking)
.create();
}
final Triplet<HttpResponseStatus, String, String> footer = readFooter(buffer);
return ResponseMessage.build()
.result(result)
+ .bulked(bulking)
.code(footer.getValue0())
.statusMessage(footer.getValue1())
.exception(footer.getValue2())
diff --git a/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/binary/RequestMessageSerializer.java b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/binary/RequestMessageSerializer.java
index bbf6d70ba4..986dbc8479 100644
--- a/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/binary/RequestMessageSerializer.java
+++ b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/binary/RequestMessageSerializer.java
@@ -70,6 +70,9 @@ public class RequestMessageSerializer {
if (fields.containsKey(Tokens.ARGS_BATCH_SIZE)) {
builder.addChunkSize((int) fields.get(Tokens.ARGS_BATCH_SIZE));
}
+ if (fields.containsKey(Tokens.BULKED)) {
+ builder.addBulkedResult(Boolean.parseBoolean(fields.get(Tokens.BULKED).toString()));
+ }
return builder.create();
} catch (IOException ex) {