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/11 15:10:48 UTC
(tinkerpop) 01/01: Added `bulked` as token for cluster setting to add a header, as well as a per-request option for server to return bulked result data.
This is an automated email from the ASF dual-hosted git repository.
xiazcy pushed a commit to branch graphbinary-bulked-response
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit 93e6d49455d2ba12cc88d288ed3d4db7724ef1bd
Author: Yang Xia <55...@users.noreply.github.com>
AuthorDate: Fri Oct 11 08:10:37 2024 -0700
Added `bulked` as token for cluster setting to add a header, as well as a per-request option for server to return bulked 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 | 28 ++++++++++++
.../tinkerpop/gremlin/driver/RequestOptions.java | 14 ++++++
.../apache/tinkerpop/gremlin/driver/Settings.java | 8 ++++
.../driver/handler/GremlinResponseHandler.java | 15 ++++++-
.../driver/handler/HttpGremlinRequestEncoder.java | 8 +++-
.../handler/HttpGremlinResponseStreamDecoder.java | 5 +++
.../driver/remote/DriverRemoteConnection.java | 2 +-
.../driver/remote/DriverRemoteTraversal.java | 7 ++-
.../gremlin/driver/simple/SimpleHttpClient.java | 2 +-
.../tinkerpop/gremlin/driver/SettingsTest.java | 3 ++
.../HttpGremlinResponseStreamDecoderTest.java | 2 +-
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 | 6 ++-
.../tests/driver/test_driver_remote_connection.py | 8 ++++
.../server/handler/HttpGremlinEndpointHandler.java | 50 ++++++++++++++--------
.../gremlin/driver/remote/RemoteWorld.java | 10 +++++
.../gremlin/server/GremlinServerIntegrateTest.java | 50 ++++++++++++++++++++--
.../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, 272 insertions(+), 53 deletions(-)
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 95ab1c2628..1b9cc520ae 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 `bulked` as a request option in body and as a cluster setting that adds a `bulked` header to toggle on bulked 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..7e08c1c056 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));
+ if (options.isBulkedResult()) request.addBulkedResult(true);
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..c27c8542f2 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,14 @@ public final class Cluster {
return manager.isUserAgentOnConnectEnabled();
}
+ /**
+ * Checks if cluster is configured to send bulked results
+ * in the web socket handshake
+ */
+ public boolean isBulkingEnabled() {
+ return manager.isBulkedResultEnabled();
+ }
+
public final static class Builder {
private final List<InetAddress> addresses = new ArrayList<>();
private int port = 8182;
@@ -494,6 +503,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 +807,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 +884,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 +897,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 +1061,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..ee5a253d81 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 boolean 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 boolean isBulkedResult() { return 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 boolean bulkedResult = false;
/**
* The aliases to set on the request.
@@ -177,6 +186,11 @@ public final class RequestOptions {
return this;
}
+ public Builder withBulkedResult(final boolean bulked) {
+ this.bulkedResult = 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..3dfd983481 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 = (boolean) channelHandlerContext.channel().attr(AttributeKey.valueOf("isBulked")).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(); ) {
+ Object obj = iter.next();
+ 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..16400ad9e6 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
@@ -49,6 +49,7 @@ public class HttpGremlinResponseStreamDecoder extends MessageToMessageDecoder<De
private static final AttributeKey<HttpResponseStatus> RESPONSE_STATUS = AttributeKey.valueOf("responseStatus");
private static final AttributeKey<String> RESPONSE_ENCODING = AttributeKey.valueOf("responseSerializer");
private static final AttributeKey<Long> BYTES_READ = AttributeKey.valueOf("bytesRead");
+ private static final AttributeKey<Boolean> IS_BULKED = AttributeKey.valueOf("isBulked");
private final MessageSerializer<?> serializer;
private final long maxResponseContentLength;
@@ -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/DriverRemoteConnection.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteConnection.java
index b779e57ab3..62b0dd89f6 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteConnection.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteConnection.java
@@ -219,7 +219,7 @@ public class DriverRemoteConnection implements RemoteConnection {
try {
gremlinLang.addG(remoteTraversalSourceName);
return client.submitAsync(gremlinLang.getGremlin(), getRequestOptions(gremlinLang))
- .thenApply(rs -> new DriverRemoteTraversal<>(rs, client, attachElements, conf));
+ .thenApply(rs -> new DriverRemoteTraversal<>(rs, attachElements, conf));
} catch (Exception ex) {
throw new RemoteConnectionException(ex);
}
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..b764c4fb7e 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
@@ -53,7 +53,7 @@ public class DriverRemoteTraversal<S, E> extends AbstractRemoteTraversal<S, E> {
private final Iterator<Traverser.Admin<E>> traversers;
private Traverser.Admin<E> lastTraverser = EmptyTraverser.instance();
- public DriverRemoteTraversal(final ResultSet rs, final Client client, final boolean attach, final Optional<Configuration> conf) {
+ public DriverRemoteTraversal(final ResultSet rs, final boolean attach, final Optional<Configuration> conf) {
// attaching is really just for testing purposes. it doesn't make sense in any real-world scenario as it would
// require that the client have access to the Graph instance that produced the result. tests need that
// attachment process to properly execute in full hence this little hack.
@@ -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);
+ 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..95b45bf611 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
@@ -83,7 +83,7 @@ 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());
}
}
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..c6664b558d 100644
--- a/gremlin-python/src/main/python/radish/terrain.py
+++ b/gremlin-python/src/main/python/radish/terrain.py
@@ -95,8 +95,10 @@ def __create_remote(server_graph_name):
raise ValueError('test configuration requires setting of --user-data="serializer={mime-type}"')
if world.config.user_data["serializer"] == "application/vnd.graphbinary-v4.0":
- s = serializer.GraphBinarySerializersV4()
+ s = serializer.GraphBinarySerializersV4
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..c227527592 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,14 @@ 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_("bulked", True).inject(1,2,3,2,1).to_list()
+ assert 5 == len(bulked_results)
+ print(g.with_("bulked", True).inject(1,2,3,2,1).to_list())
+
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..c5b4f1f6e7 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,18 @@ 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 boolean bulking = ((Objects.equals(context.getChannelHandlerContext().channel().attr(StateKey.REQUEST_HEADERS).get().get(Tokens.BULKED), "true")
+ && result instanceof Traversal)
+ || (args.containsKey(Tokens.BULKED) && (boolean) args.get(Tokens.BULKED))
+ && Objects.equals(language, "gremlin-lang"));
+
+ 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 +374,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 +384,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 +421,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 +449,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 +516,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 +539,8 @@ public class HttpGremlinEndpointHandler extends SimpleChannelInboundHandler<Requ
builder.code(HttpResponseStatus.OK);
}
+ builder.bulked(bulking);
+
responseMessage = builder.create();
}
@@ -541,6 +554,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/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..2da6204c73 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,49 @@ 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.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("bulked", false).inject(1, 2, 3, 2, 1).toList();
+ assertEquals(5, result.size());
+
+ assertTrue(cluster.isBulkingEnabled());
+
+ } catch (Exception ex) {
+ throw ex;
+ } 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("bulked", true).inject(1, 2, 3, 2, 1).toList();
+ assertEquals(5, resultBulked.size());
+
+ assertFalse(cluster.isBulkingEnabled());
+
+ } catch (Exception ex) {
+ throw ex;
+ } 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..cb0fe713ec 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, 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..2d43a94b39 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) fields.get(Tokens.BULKED));
+ }
return builder.create();
} catch (IOException ex) {