You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by va...@apache.org on 2024/03/26 16:28:07 UTC
(tinkerpop) branch valentyn/streaming-serializers created (now b56b2c0c4a)
This is an automated email from the ASF dual-hosted git repository.
valentyn pushed a change to branch valentyn/streaming-serializers
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
at b56b2c0c4a draft for GraphSON4
This branch includes the following new commits:
new b56b2c0c4a draft for GraphSON4
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(tinkerpop) 01/01: draft for GraphSON4
Posted by va...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
valentyn pushed a commit to branch valentyn/streaming-serializers
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit b56b2c0c4ae9db7da59093666a85f3cbb08a91c2
Author: Valentyn Kahamlyk <Va...@improving.com>
AuthorDate: Tue Mar 26 09:27:57 2024 -0700
draft for GraphSON4
---
.../structure/io/graphson/GraphSONMapper.java | 14 +-
.../structure/io/graphson/GraphSONModule.java | 253 +++++++++++++++
.../structure/io/graphson/GraphSONVersion.java | 3 +-
.../apache/tinkerpop/gremlin/server/Context.java | 19 ++
.../server/handler/HttpGremlinEndpointHandler.java | 86 +++++-
.../gremlin/server/handler/HttpHandlerUtil.java | 44 ++-
.../gremlin/server/handler/OpSelectorHandler.java | 4 +-
.../tinkerpop/gremlin/server/ContextTest.java | 1 +
.../server/GremlinServerHttpIntegrateTest.java | 86 +++++-
.../gremlin/util/message/ResponseMessage.java | 24 ++
.../ser/AbstractGraphSONMessageSerializerV2.java | 20 +-
.../util/ser/GraphSONMessageSerializerV4.java | 340 +++++++++++++++++++++
.../ser/GraphSONUntypedMessageSerializerV3.java | 1 +
.../gremlin/util/ser/MessageChunkSerializer.java | 34 +++
.../tinkerpop/gremlin/util/ser/SerTokens.java | 1 +
.../util/ser/GraphSONMessageSerializerV3Test.java | 6 +-
.../util/ser/GraphSONMessageSerializerV4Test.java | 141 +++++++++
17 files changed, 1025 insertions(+), 52 deletions(-)
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/graphson/GraphSONMapper.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/graphson/GraphSONMapper.java
index 9e655941ac..04174c429d 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/graphson/GraphSONMapper.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/graphson/GraphSONMapper.java
@@ -94,7 +94,8 @@ public class GraphSONMapper implements Mapper<ObjectMapper> {
if (loadCustomSerializers)
om.findAndRegisterModules();
- if ((version == GraphSONVersion.V3_0 || version == GraphSONVersion.V2_0) && typeInfo != TypeInfo.NO_TYPES) {
+ if ((version == GraphSONVersion.V4_0 || version == GraphSONVersion.V3_0 || version == GraphSONVersion.V2_0) &&
+ typeInfo != TypeInfo.NO_TYPES) {
final GraphSONTypeIdResolver graphSONTypeIdResolver = new GraphSONTypeIdResolver();
final TypeResolverBuilder typer = new GraphSONTypeResolverBuilder(version)
.typesEmbedding(this.typeInfo)
@@ -134,7 +135,7 @@ public class GraphSONMapper implements Mapper<ObjectMapper> {
.typeProperty(GraphSONTokens.CLASS);
om.setDefaultTyping(typer);
}
- } else if (version == GraphSONVersion.V3_0) {
+ } else if (version == GraphSONVersion.V3_0 || version == GraphSONVersion.V4_0) {
} else {
throw new IllegalStateException("Unknown GraphSONVersion: " + version);
@@ -199,7 +200,7 @@ public class GraphSONMapper implements Mapper<ObjectMapper> {
private boolean loadCustomModules = false;
private boolean normalize = false;
private List<IoRegistry> registries = new ArrayList<>();
- private GraphSONVersion version = GraphSONVersion.V3_0;
+ private GraphSONVersion version = GraphSONVersion.V4_0;
private boolean includeDefaultXModule = false;
private StreamReadConstraints.Builder streamReadConstraintsBuilder = StreamReadConstraints.builder()
.maxNumberLength(DEFAULT_MAX_NUMBER_LENGTH);
@@ -218,7 +219,7 @@ public class GraphSONMapper implements Mapper<ObjectMapper> {
}
/**
- * Set the version of GraphSON to use. The default is {@link GraphSONVersion#V3_0}.
+ * Set the version of GraphSON to use. The default is {@link GraphSONVersion#V4_0}.
*/
public Builder version(final GraphSONVersion version) {
this.version = version;
@@ -253,7 +254,7 @@ public class GraphSONMapper implements Mapper<ObjectMapper> {
}
/**
- * Supply a default extension module of V2_0 and V3_0 for serialization/deserialization.
+ * Supply a default extension module of V2_0, V3_0 and V4_0 for serialization/deserialization.
*/
public Builder addDefaultXModule(final boolean includeDefaultXModule) {
this.includeDefaultXModule = includeDefaultXModule;
@@ -321,6 +322,9 @@ public class GraphSONMapper implements Mapper<ObjectMapper> {
this.addCustomModule(GraphSONXModuleV2.build().create(this.normalize, typeInfo));
} else if (this.version == GraphSONVersion.V3_0) {
this.addCustomModule(GraphSONXModuleV3.build().create(this.normalize, typeInfo));
+ } else if (this.version == GraphSONVersion.V4_0) {
+ // todo:
+ this.addCustomModule(GraphSONXModuleV3.build().create(this.normalize, typeInfo));
}
}
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/graphson/GraphSONModule.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/graphson/GraphSONModule.java
index a274ec9abb..d2266edcfd 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/graphson/GraphSONModule.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/graphson/GraphSONModule.java
@@ -135,6 +135,259 @@ abstract class GraphSONModule extends TinkerPopJacksonModule {
}
}
+ /**
+ * Version 4.0 of GraphSON.
+ */
+ static final class GraphSONModuleV4 extends GraphSONModule {
+
+ private static final Map<Class, String> TYPE_DEFINITIONS = Collections.unmodifiableMap(
+ new LinkedHashMap<Class, String>() {{
+ // Those don't have deserializers because handled by Jackson,
+ // but we still want to rename them in GraphSON
+ put(Integer.class, "Int32");
+ put(Long.class, "Int64");
+ put(Double.class, "Double");
+ put(Float.class, "Float");
+
+ put(Map.class, "Map");
+ put(List.class, "List");
+ put(Set.class, "Set");
+
+ // TinkerPop Graph objects
+ put(Lambda.class, "Lambda");
+ put(Vertex.class, "Vertex");
+ put(Edge.class, "Edge");
+ put(Property.class, "Property");
+ put(Path.class, "Path");
+ put(VertexProperty.class, "VertexProperty");
+ put(Metrics.class, "Metrics");
+ put(TraversalMetrics.class, "TraversalMetrics");
+ put(TraversalExplanation.class, "TraversalExplanation");
+ put(Traverser.class, "Traverser");
+ put(Tree.class, "Tree");
+ put(BulkSet.class, "BulkSet");
+ put(Bytecode.class, "Bytecode");
+ put(Bytecode.Binding.class, "Binding");
+ put(AndP.class, "P");
+ put(OrP.class, "P");
+ put(P.class, "P");
+ put(TextP.class, "TextP");
+ Stream.of(
+ VertexProperty.Cardinality.class,
+ Column.class,
+ Direction.class,
+ DT.class,
+ Merge.class,
+ Operator.class,
+ Order.class,
+ Pop.class,
+ SackFunctions.Barrier.class,
+ Pick.class,
+ Scope.class,
+ T.class).forEach(e -> put(e, e.getSimpleName()));
+ Arrays.asList(
+ ConnectiveStrategy.class,
+ ElementIdStrategy.class,
+ EventStrategy.class,
+ HaltedTraverserStrategy.class,
+ PartitionStrategy.class,
+ SubgraphStrategy.class,
+ SeedStrategy.class,
+ LazyBarrierStrategy.class,
+ MatchAlgorithmStrategy.class,
+ AdjacentToIncidentStrategy.class,
+ ByModulatorOptimizationStrategy.class,
+ ProductiveByStrategy.class,
+ CountStrategy.class,
+ FilterRankingStrategy.class,
+ IdentityRemovalStrategy.class,
+ IncidentToAdjacentStrategy.class,
+ InlineFilterStrategy.class,
+ MatchPredicateStrategy.class,
+ OrderLimitStrategy.class,
+ OptionsStrategy.class,
+ PathProcessorStrategy.class,
+ PathRetractionStrategy.class,
+ RepeatUnrollStrategy.class,
+ ComputerVerificationStrategy.class,
+ LambdaRestrictionStrategy.class,
+ ReadOnlyStrategy.class,
+ StandardVerificationStrategy.class,
+ EarlyLimitStrategy.class,
+ EdgeLabelVerificationStrategy.class,
+ ReservedKeysVerificationStrategy.class,
+ //
+ GraphFilterStrategy.class,
+ VertexProgramStrategy.class
+ ).forEach(strategy -> put(strategy, strategy.getSimpleName()));
+
+ GraphSONModule.tryLoadSparqlStrategy().ifPresent(s -> put(s, s.getSimpleName()));
+ }});
+
+ /**
+ * Constructs a new object.
+ */
+ protected GraphSONModuleV4(final boolean normalize, final TypeInfo typeInfo) {
+ super("graphson-4.0");
+
+ /////////////////////// SERIALIZERS ////////////////////////////
+
+ // graph
+ addSerializer(Edge.class, new GraphSONSerializersV3.EdgeJacksonSerializer(normalize, typeInfo));
+ addSerializer(Vertex.class, new GraphSONSerializersV3.VertexJacksonSerializer(normalize, typeInfo));
+ addSerializer(VertexProperty.class, new GraphSONSerializersV3.VertexPropertyJacksonSerializer(normalize, true));
+ addSerializer(Property.class, new GraphSONSerializersV3.PropertyJacksonSerializer());
+ addSerializer(Metrics.class, new GraphSONSerializersV3.MetricsJacksonSerializer());
+ addSerializer(TraversalMetrics.class, new GraphSONSerializersV3.TraversalMetricsJacksonSerializer());
+ addSerializer(TraversalExplanation.class, new GraphSONSerializersV3.TraversalExplanationJacksonSerializer());
+ addSerializer(Path.class, new GraphSONSerializersV3.PathJacksonSerializer());
+ addSerializer(DirectionalStarGraph.class, new StarGraphGraphSONSerializerV3(normalize));
+ addSerializer(Tree.class, new GraphSONSerializersV3.TreeJacksonSerializer());
+
+ // java.util - use the standard jackson serializers for collections when types aren't embedded
+ if (typeInfo != TypeInfo.NO_TYPES) {
+ addSerializer(Map.Entry.class, new JavaUtilSerializersV3.MapEntryJacksonSerializer());
+ addSerializer(Map.class, new JavaUtilSerializersV3.MapJacksonSerializer());
+ addSerializer(List.class, new JavaUtilSerializersV3.ListJacksonSerializer());
+ addSerializer(Set.class, new JavaUtilSerializersV3.SetJacksonSerializer());
+ }
+
+ // need to explicitly add serializers for these types because Jackson doesn't do it at all.
+ addSerializer(Integer.class, new GraphSONSerializersV3.IntegerGraphSONSerializer());
+ addSerializer(Double.class, new GraphSONSerializersV3.DoubleGraphSONSerializer());
+
+ // traversal
+ addSerializer(BulkSet.class, new TraversalSerializersV3.BulkSetJacksonSerializer());
+ addSerializer(Traversal.class, new TraversalSerializersV3.TraversalJacksonSerializer());
+ addSerializer(Bytecode.class, new TraversalSerializersV3.BytecodeJacksonSerializer());
+ Stream.of(VertexProperty.Cardinality.class,
+ Column.class,
+ Direction.class,
+ DT.class,
+ Merge.class,
+ Operator.class,
+ Order.class,
+ Pop.class,
+ SackFunctions.Barrier.class,
+ Scope.class,
+ Pick.class,
+ T.class).forEach(e -> addSerializer(e, new TraversalSerializersV3.EnumJacksonSerializer()));
+ addSerializer(P.class, new TraversalSerializersV3.PJacksonSerializer());
+ addSerializer(Lambda.class, new TraversalSerializersV3.LambdaJacksonSerializer());
+ addSerializer(Bytecode.Binding.class, new TraversalSerializersV3.BindingJacksonSerializer());
+ addSerializer(Traverser.class, new TraversalSerializersV3.TraverserJacksonSerializer());
+ addSerializer(TraversalStrategy.class, new TraversalSerializersV3.TraversalStrategyJacksonSerializer());
+
+ /////////////////////// DESERIALIZERS ////////////////////////////
+
+ // Tinkerpop Graph
+ addDeserializer(Vertex.class, new GraphSONSerializersV3.VertexJacksonDeserializer());
+ addDeserializer(Edge.class, new GraphSONSerializersV3.EdgeJacksonDeserializer());
+ addDeserializer(Property.class, new GraphSONSerializersV3.PropertyJacksonDeserializer());
+ addDeserializer(Path.class, new GraphSONSerializersV3.PathJacksonDeserializer());
+ addDeserializer(TraversalExplanation.class, new GraphSONSerializersV3.TraversalExplanationJacksonDeserializer());
+ addDeserializer(VertexProperty.class, new GraphSONSerializersV3.VertexPropertyJacksonDeserializer());
+ addDeserializer(Metrics.class, new GraphSONSerializersV3.MetricsJacksonDeserializer());
+ addDeserializer(TraversalMetrics.class, new GraphSONSerializersV3.TraversalMetricsJacksonDeserializer());
+ addDeserializer(Tree.class, new GraphSONSerializersV3.TreeJacksonDeserializer());
+
+ // java.util - use the standard jackson serializers for collections when types aren't embedded
+ if (typeInfo != TypeInfo.NO_TYPES) {
+ addDeserializer(Map.class, new JavaUtilSerializersV3.MapJacksonDeserializer());
+ addDeserializer(List.class, new JavaUtilSerializersV3.ListJacksonDeserializer());
+ addDeserializer(Set.class, new JavaUtilSerializersV3.SetJacksonDeserializer());
+ }
+
+ // numbers
+ addDeserializer(Integer.class, new GraphSONSerializersV3.IntegerJackonsDeserializer());
+ addDeserializer(Double.class, new GraphSONSerializersV3.DoubleJacksonDeserializer());
+
+ // traversal
+ addDeserializer(BulkSet.class, new TraversalSerializersV3.BulkSetJacksonDeserializer());
+ addDeserializer(Bytecode.class, new TraversalSerializersV3.BytecodeJacksonDeserializer());
+ addDeserializer(Bytecode.Binding.class, new TraversalSerializersV3.BindingJacksonDeserializer());
+ Stream.of(VertexProperty.Cardinality.values(),
+ Column.values(),
+ Direction.values(),
+ DT.values(),
+ Merge.values(),
+ Operator.values(),
+ Order.values(),
+ Pop.values(),
+ SackFunctions.Barrier.values(),
+ Scope.values(),
+ Pick.values(),
+ T.values()).flatMap(Stream::of).forEach(e -> addDeserializer(e.getClass(), new TraversalSerializersV3.EnumJacksonDeserializer(e.getDeclaringClass())));
+ addDeserializer(P.class, new TraversalSerializersV3.PJacksonDeserializer());
+ addDeserializer(TextP.class, new TraversalSerializersV3.TextPJacksonDeserializer());
+ addDeserializer(Lambda.class, new TraversalSerializersV3.LambdaJacksonDeserializer());
+ addDeserializer(Traverser.class, new TraversalSerializersV3.TraverserJacksonDeserializer());
+ Arrays.asList(
+ ConnectiveStrategy.class,
+ ElementIdStrategy.class,
+ EventStrategy.class,
+ HaltedTraverserStrategy.class,
+ PartitionStrategy.class,
+ SubgraphStrategy.class,
+ SeedStrategy.class,
+ LazyBarrierStrategy.class,
+ MatchAlgorithmStrategy.class,
+ AdjacentToIncidentStrategy.class,
+ ByModulatorOptimizationStrategy.class,
+ ProductiveByStrategy.class,
+ CountStrategy.class,
+ FilterRankingStrategy.class,
+ IdentityRemovalStrategy.class,
+ IncidentToAdjacentStrategy.class,
+ InlineFilterStrategy.class,
+ MatchPredicateStrategy.class,
+ OrderLimitStrategy.class,
+ OptionsStrategy.class,
+ PathProcessorStrategy.class,
+ PathRetractionStrategy.class,
+ RepeatUnrollStrategy.class,
+ ComputerVerificationStrategy.class,
+ LambdaRestrictionStrategy.class,
+ ReadOnlyStrategy.class,
+ StandardVerificationStrategy.class,
+ EarlyLimitStrategy.class,
+ EdgeLabelVerificationStrategy.class,
+ ReservedKeysVerificationStrategy.class,
+ //
+ GraphFilterStrategy.class,
+ VertexProgramStrategy.class
+ ).forEach(strategy -> addDeserializer(strategy, new TraversalSerializersV3.TraversalStrategyProxyJacksonDeserializer(strategy)));
+
+ GraphSONModule.tryLoadSparqlStrategy().ifPresent(s -> addDeserializer(s, new TraversalSerializersV3.TraversalStrategyProxyJacksonDeserializer(s)));
+ }
+
+ public static Builder build() {
+ return new Builder();
+ }
+
+ @Override
+ public Map<Class, String> getTypeDefinitions() {
+ return TYPE_DEFINITIONS;
+ }
+
+ @Override
+ public String getTypeNamespace() {
+ return GraphSONTokens.GREMLIN_TYPE_NAMESPACE;
+ }
+
+ static final class Builder implements GraphSONModuleBuilder {
+
+ private Builder() {
+ }
+
+ @Override
+ public GraphSONModule create(final boolean normalize, final TypeInfo typeInfo) {
+ return new GraphSONModuleV4(normalize, typeInfo);
+ }
+
+ }
+ }
+
/**
* Version 3.0 of GraphSON.
*/
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/graphson/GraphSONVersion.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/graphson/GraphSONVersion.java
index 9738496171..09cfb699d4 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/graphson/GraphSONVersion.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/io/graphson/GraphSONVersion.java
@@ -26,7 +26,8 @@ package org.apache.tinkerpop.gremlin.structure.io.graphson;
public enum GraphSONVersion {
V1_0(GraphSONModule.GraphSONModuleV1.build(), "1.0"),
V2_0(GraphSONModule.GraphSONModuleV2.build(), "2.0"),
- V3_0(GraphSONModule.GraphSONModuleV3.build(), "3.0");
+ V3_0(GraphSONModule.GraphSONModuleV3.build(), "3.0"),
+ V4_0(GraphSONModule.GraphSONModuleV4.build(), "4.0");
private final GraphSONModule.GraphSONModuleBuilder builder;
private final String versionNumber;
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/Context.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/Context.java
index 8636954ad5..236704c2be 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/Context.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/Context.java
@@ -19,6 +19,7 @@
package org.apache.tinkerpop.gremlin.server;
import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.AbstractTraverser;
+import org.apache.tinkerpop.gremlin.server.handler.HttpGremlinEndpointHandler;
import org.apache.tinkerpop.gremlin.structure.Element;
import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceFactory;
import org.apache.tinkerpop.gremlin.util.Tokens;
@@ -60,6 +61,7 @@ public class Context {
private final String materializeProperties;
private final RequestContentType requestContentType;
private final Object gremlinArgument;
+ private HttpGremlinEndpointHandler.RequestState requestState;
private final AtomicBoolean startedResponse = new AtomicBoolean(false);
private ScheduledFuture<?> timeoutExecutor = null;
private boolean timeoutExecutorGrabbed = false;
@@ -88,6 +90,14 @@ public class Context {
public Context(final RequestMessage requestMessage, final ChannelHandlerContext ctx,
final Settings settings, final GraphManager graphManager,
final GremlinExecutor gremlinExecutor, final ScheduledExecutorService scheduledExecutorService) {
+ this(requestMessage, ctx, settings, graphManager, gremlinExecutor, scheduledExecutorService,
+ HttpGremlinEndpointHandler.RequestState.CHUNKING_NOT_SUPPORTED);
+ }
+
+ public Context(final RequestMessage requestMessage, final ChannelHandlerContext ctx,
+ final Settings settings, final GraphManager graphManager,
+ final GremlinExecutor gremlinExecutor, final ScheduledExecutorService scheduledExecutorService,
+ final HttpGremlinEndpointHandler.RequestState requestState) {
this.requestMessage = requestMessage;
this.channelHandlerContext = ctx;
this.settings = settings;
@@ -97,6 +107,7 @@ public class Context {
// order of calls matter as one depends on the next
this.gremlinArgument = requestMessage.getArgs().get(Tokens.ARGS_GREMLIN);
+ this.requestState = requestState;
this.requestContentType = determineRequestContents();
this.requestTimeout = determineTimeout();
this.materializeProperties = determineMaterializeProperties();
@@ -335,4 +346,12 @@ public class Context {
}
}
}
+
+ public HttpGremlinEndpointHandler.RequestState getRequestState() {
+ return requestState;
+ }
+
+ public void setRequestState(HttpGremlinEndpointHandler.RequestState requestState) {
+ this.requestState = requestState;
+ }
}
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 d2d8e8619a..00e88c72ab 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
@@ -67,6 +67,7 @@ import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
import org.apache.tinkerpop.gremlin.util.message.RequestMessage;
import org.apache.tinkerpop.gremlin.util.message.ResponseMessage;
import org.apache.tinkerpop.gremlin.util.message.ResponseStatusCode;
+import org.apache.tinkerpop.gremlin.util.ser.MessageChunkSerializer;
import org.apache.tinkerpop.gremlin.util.ser.MessageTextSerializer;
import org.apache.tinkerpop.gremlin.util.ser.SerializationException;
import org.codehaus.groovy.control.MultipleCompilationErrorsException;
@@ -95,6 +96,7 @@ import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
@@ -114,6 +116,7 @@ import static io.netty.handler.codec.http.HttpResponseStatus.NOT_FOUND;
import static io.netty.handler.codec.http.HttpResponseStatus.OK;
import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
import static io.netty.handler.codec.http.LastHttpContent.EMPTY_LAST_CONTENT;
+import static org.apache.tinkerpop.gremlin.server.handler.HttpHandlerUtil.sendTrailingHeaders;
import static org.apache.tinkerpop.gremlin.server.handler.HttpHandlerUtil.writeErrorFrame;
/**
@@ -241,8 +244,12 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter {
// not using the req anywhere below here - assume it is safe to release at this point.
ReferenceCountUtil.release(msg);
+ final RequestState requestState = serializer.getValue1() instanceof MessageChunkSerializer
+ ? RequestState.NOT_STARTED
+ : RequestState.CHUNKING_NOT_SUPPORTED;
+
final Context requestCtx = new Context(requestMessage, ctx, settings, graphManager, gremlinExecutor,
- gremlinExecutor.getScheduledExecutorService());
+ gremlinExecutor.getScheduledExecutorService(), requestState);
final Timer.Context timerContext = evalOpTimer.time();
// timeout override - handle both deprecated and newly named configuration. earlier logic should prevent
@@ -309,21 +316,21 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter {
}
} catch (OpProcessorException ope) {
logger.warn(ope.getMessage(), ope);
- writeErrorFrame(ctx, ope.getResponseMessage(), serializer.getValue1());
+ writeErrorFrame(ctx, requestCtx, ope.getResponseMessage(), serializer.getValue1());
}
} catch (Exception ex) {
// send the error response here and don't rely on exception caught because it might not have the
// context on whether to close the connection or not, based on keepalive.
final Throwable t = ExceptionHelper.getRootCause(ex);
if (t instanceof TooLongFrameException) {
- writeErrorFrame(ctx,
+ writeErrorFrame(ctx, requestCtx,
ResponseMessage.build(requestId)
.code(ResponseStatusCode.SERVER_ERROR)
.statusMessage(t.getMessage() + " - increase the maxContentLength")
.create(),
serializer.getValue1());
} else {
- writeErrorFrame(ctx,
+ writeErrorFrame(ctx, requestCtx,
ResponseMessage.build(requestId)
.code(ResponseStatusCode.SERVER_ERROR)
.statusMessage((t != null) ? t.getMessage() : ex.getMessage())
@@ -357,7 +364,7 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter {
}, seto, TimeUnit.MILLISECONDS));
}
} catch (RejectedExecutionException ree) {
- writeErrorFrame(requestCtx.getChannelHandlerContext(),
+ writeErrorFrame(requestCtx.getChannelHandlerContext(), requestCtx,
ResponseMessage.build(requestMessage).code(ResponseStatusCode.TOO_MANY_REQUESTS) .statusMessage("Rate limiting").create(),
serializer.getValue1());
}
@@ -559,11 +566,12 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter {
specialResponseMsg.code(ResponseStatusCode.SERVER_ERROR_FAIL_STEP).
statusAttribute(Tokens.STATUS_ATTRIBUTE_FAIL_STEP_MESSAGE, failure.format());
}
- writeErrorFrame(context.getChannelHandlerContext(), specialResponseMsg.create(), serializer);
+ writeErrorFrame(context.getChannelHandlerContext(), context, specialResponseMsg.create(), serializer);
} else if (t instanceof InterruptedException || t instanceof TraversalInterruptedException) {
final String errorMessage = String.format("A timeout occurred during traversal evaluation of [%s] - consider increasing the limit given to evaluationTimeout", msg);
logger.warn(errorMessage);
writeErrorFrame(context.getChannelHandlerContext(),
+ context,
ResponseMessage.build(msg)
.code(ResponseStatusCode.SERVER_ERROR_TIMEOUT)
.statusMessage(errorMessage)
@@ -575,6 +583,7 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter {
final String errorMessage = String.format("A timeout occurred within the script during evaluation of [%s] - consider increasing the limit given to TimedInterruptCustomizerProvider", msg);
logger.warn(errorMessage);
writeErrorFrame(context.getChannelHandlerContext(),
+ context,
ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_TIMEOUT)
.statusMessage("Timeout during script evaluation triggered by TimedInterruptCustomizerProvider")
.statusAttributeException(t).create(),
@@ -583,6 +592,7 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter {
final String errorMessage = String.format("Script evaluation exceeded the configured threshold for request [%s]", msg);
logger.warn(errorMessage, t);
writeErrorFrame(context.getChannelHandlerContext(),
+ context,
ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_TIMEOUT)
.statusMessage(t.getMessage())
.statusAttributeException(t).create(),
@@ -592,6 +602,7 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter {
final String errorMessage = String.format("The Gremlin statement that was submitted exceeds the maximum compilation size allowed by the JVM, please split it into multiple smaller statements - %s", trimMessage(msg));
logger.warn(errorMessage);
writeErrorFrame(context.getChannelHandlerContext(),
+ context,
ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_EVALUATION)
.statusMessage(errorMessage)
.statusAttributeException(t).create(),
@@ -599,6 +610,7 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter {
} else {
logger.warn(String.format("Exception processing a Traversal on iteration for request [%s].", msg.getRequestId()), ex);
writeErrorFrame(context.getChannelHandlerContext(),
+ context,
ResponseMessage.build(msg)
.code(ResponseStatusCode.SERVER_ERROR)
.statusMessage(ex.getMessage())
@@ -610,6 +622,7 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter {
} catch (Throwable t) {
logger.warn(String.format("Exception processing a Traversal on request [%s].", msg.getRequestId()), t);
writeErrorFrame(context.getChannelHandlerContext(),
+ context,
ResponseMessage.build(msg)
.code(ResponseStatusCode.SERVER_ERROR)
.statusMessage(t.getMessage())
@@ -642,9 +655,10 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter {
return;
}
- // the batch size can be overridden by the request
- final int resultIterationBatchSize = (Integer) msg.optionalArgs(Tokens.ARGS_BATCH_SIZE)
- .orElse(settings.resultIterationBatchSize);
+ // the batch size can be overridden by the request !!!
+ final int resultIterationBatchSize = 2;
+// (Integer) msg.optionalArgs(Tokens.ARGS_BATCH_SIZE)
+// .orElse(settings.resultIterationBatchSize);
List<Object> aggregate = new ArrayList<>(resultIterationBatchSize);
// use an external control to manage the loop as opposed to just checking hasNext() in the while. this
@@ -729,7 +743,7 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter {
context.writeAndFlush(code, new DefaultHttpContent((ByteBuf) frame.getMsg()));
if (!hasMore) {
- nettyContext.writeAndFlush(EMPTY_LAST_CONTENT);
+ sendTrailingHeaders(nettyContext, 200, "OK");
}
}
} else {
@@ -783,11 +797,43 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter {
ctx.handleDetachment(aggregate);
if (useBinary) {
- return new Frame(serializer.serializeResponseAsBinary(ResponseMessage.build(msg)
- .code(code)
- .statusAttributes(statusAttributes)
- .responseMetaData(responseMetaData)
- .result(aggregate).create(), nettyContext.alloc()));
+ if (code == ResponseStatusCode.SUCCESS && ctx.getRequestState() == RequestState.STREAMING) {
+ ctx.setRequestState(RequestState.FINISHING);
+ }
+
+ final ResponseMessage responseMessage = ctx.getRequestState() == RequestState.STREAMING ? null :
+ ResponseMessage.build(msg)
+ // always return 200, error can be added later
+ .code(ResponseStatusCode.SUCCESS)
+ .statusAttributes(statusAttributes)
+ .responseMetaData(responseMetaData)
+ .result(aggregate)
+ .create();
+
+ if (ctx.getRequestState() == RequestState.CHUNKING_NOT_SUPPORTED) {
+ return new Frame(serializer.serializeResponseAsBinary(responseMessage, nettyContext.alloc()));
+ }
+
+ final MessageChunkSerializer<?> chunkSerializer = (MessageChunkSerializer) serializer;
+
+ switch (ctx.getRequestState()) {
+ case NOT_STARTED:
+ if (code == ResponseStatusCode.SUCCESS) {
+ ctx.setRequestState(RequestState.FINISHED);
+ return new Frame(serializer.serializeResponseAsBinary(responseMessage, nettyContext.alloc()));
+ }
+
+ ctx.setRequestState(RequestState.STREAMING);
+ return new Frame(chunkSerializer.writeResponseHeader(responseMessage, nettyContext.alloc()));
+ case STREAMING:
+ return new Frame(chunkSerializer.writeResponseChunk(aggregate, nettyContext.alloc()));
+ case FINISHING:
+ ctx.setRequestState(RequestState.FINISHED);
+ return new Frame(chunkSerializer.writeResponseFooter(responseMessage, nettyContext.alloc()));
+ }
+
+ // todo: just throw?
+ return new Frame(serializer.serializeResponseAsBinary(responseMessage, nettyContext.alloc()));
} else {
// the expectation is that the GremlinTextRequestDecoder will have placed a MessageTextSerializer
// instance on the channel.
@@ -822,4 +868,14 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter {
return trimmedMsg;
}
+
+ public enum RequestState {
+ CHUNKING_NOT_SUPPORTED,
+ NOT_STARTED,
+ STREAMING,
+ // last portion of data
+ FINISHING,
+ FINISHED,
+ ERROR
+ }
}
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpHandlerUtil.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpHandlerUtil.java
index 13dfe4ad67..3ce6cb3f3a 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpHandlerUtil.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpHandlerUtil.java
@@ -26,28 +26,25 @@ import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.DefaultHttpContent;
-import io.netty.handler.codec.http.DefaultHttpHeaders;
import io.netty.handler.codec.http.DefaultLastHttpContent;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpHeaderNames;
-import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpUtil;
-import io.netty.handler.codec.http.LastHttpContent;
import io.netty.handler.codec.http.QueryStringDecoder;
import io.netty.util.CharsetUtil;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.tinkerpop.gremlin.server.Context;
-import org.apache.tinkerpop.gremlin.util.ExceptionHelper;
-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.server.GremlinServer;
import org.apache.tinkerpop.gremlin.server.op.standard.StandardOpProcessor;
import org.apache.tinkerpop.gremlin.server.util.MetricManager;
+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.message.ResponseMessage;
-import org.apache.tinkerpop.gremlin.util.message.ResponseStatusCode;
+import org.apache.tinkerpop.gremlin.util.ser.MessageChunkSerializer;
+import org.apache.tinkerpop.gremlin.util.ser.SerTokens;
import org.apache.tinkerpop.gremlin.util.ser.SerializationException;
import org.apache.tinkerpop.shaded.jackson.databind.JsonNode;
import org.apache.tinkerpop.shaded.jackson.databind.ObjectMapper;
@@ -57,13 +54,9 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
-import java.io.UnsupportedEncodingException;
-import java.net.URLEncoder;
-import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
@@ -301,13 +294,34 @@ public class HttpHandlerUtil {
}
}
- static void writeErrorFrame(final ChannelHandlerContext ctx, final ResponseMessage responseMessage, final MessageSerializer<?> serializer) {
+ static void writeErrorFrame(final ChannelHandlerContext ctx, final ResponseMessage responseMessage, final MessageSerializer<?> serializer) {
try {
- ctx.writeAndFlush(new DefaultHttpContent(
- (ByteBuf) new Frame(serializer.serializeResponseAsBinary(responseMessage, ctx.alloc())).getMsg()));
+ ctx.writeAndFlush(new DefaultHttpContent(serializer.serializeResponseAsBinary(responseMessage, ctx.alloc())));
ctx.writeAndFlush(EMPTY_LAST_CONTENT);
} catch (SerializationException se) {
logger.warn("Unable to serialize ResponseMessage: {} ", responseMessage);
}
}
+
+ static void writeErrorFrame(final ChannelHandlerContext ctx, final Context context, ResponseMessage responseMessage, final MessageSerializer<?> serializer) {
+ try {
+ final ByteBuf ByteBuf = context.getRequestState() == HttpGremlinEndpointHandler.RequestState.STREAMING
+ ? ((MessageChunkSerializer) serializer).writeError(responseMessage, ctx.alloc())
+ : serializer.serializeResponseAsBinary(responseMessage, ctx.alloc());
+
+ context.setRequestState(HttpGremlinEndpointHandler.RequestState.ERROR);
+ ctx.writeAndFlush(new DefaultHttpContent(ByteBuf));
+
+ sendTrailingHeaders(ctx, responseMessage.getStatus().getCode().getValue(), responseMessage.getStatus().getMessage());
+ } catch (SerializationException se) {
+ logger.warn("Unable to serialize ResponseMessage: {} ", responseMessage);
+ }
+ }
+
+ static void sendTrailingHeaders(final ChannelHandlerContext ctx, final int statusCode, final String message) {
+ final DefaultLastHttpContent defaultLastHttpContent = new DefaultLastHttpContent();
+ defaultLastHttpContent.trailingHeaders().add(SerTokens.TOKEN_CODE, statusCode);
+ defaultLastHttpContent.trailingHeaders().add(SerTokens.TOKEN_MESSAGE, message);
+ ctx.writeAndFlush(defaultLastHttpContent);
+ }
}
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/OpSelectorHandler.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/OpSelectorHandler.java
index 340b913165..d3a92e7aaa 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/OpSelectorHandler.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/OpSelectorHandler.java
@@ -68,8 +68,8 @@ public class OpSelectorHandler extends MessageToMessageDecoder<RequestMessage> {
@Override
protected void decode(final ChannelHandlerContext ctx, final RequestMessage msg,
final List<Object> objects) throws Exception {
- final Context gremlinServerContext = new Context(msg, ctx, settings,
- graphManager, gremlinExecutor, this.scheduledExecutorService);
+ final Context gremlinServerContext = new Context(msg, ctx,
+ settings, graphManager, gremlinExecutor, this.scheduledExecutorService);
try {
// choose a processor to do the work based on the request message.
final Optional<OpProcessor> processor = OpLoader.getProcessor(msg.getProcessor());
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/ContextTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/ContextTest.java
index 2092713dc3..db75a12de8 100644
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/ContextTest.java
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/ContextTest.java
@@ -20,6 +20,7 @@ package org.apache.tinkerpop.gremlin.server;
import io.netty.channel.ChannelHandlerContext;
import nl.altindag.log.LogCaptor;
+import org.apache.tinkerpop.gremlin.server.handler.HttpGremlinEndpointHandler;
import org.apache.tinkerpop.gremlin.util.Tokens;
import org.apache.tinkerpop.gremlin.util.message.RequestMessage;
import org.apache.tinkerpop.gremlin.util.message.ResponseMessage;
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerHttpIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerHttpIntegrateTest.java
index e12859b82d..2321bed855 100644
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerHttpIntegrateTest.java
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerHttpIntegrateTest.java
@@ -21,11 +21,15 @@ package org.apache.tinkerpop.gremlin.server;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.buffer.UnpooledByteBufAllocator;
+import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.RandomStringUtils;
+import org.apache.http.Header;
import org.apache.http.HttpEntity;
import org.apache.http.HttpHeaders;
import org.apache.http.HttpVersion;
+import org.apache.http.conn.EofSensorInputStream;
import org.apache.http.entity.ByteArrayEntity;
+import org.apache.http.impl.io.ChunkedInputStream;
import org.apache.tinkerpop.gremlin.driver.simple.SimpleClient;
import org.apache.tinkerpop.gremlin.process.traversal.Bytecode;
import org.apache.tinkerpop.gremlin.structure.util.empty.EmptyGraph;
@@ -64,6 +68,9 @@ import org.junit.Test;
import javax.script.SimpleBindings;
import java.io.IOException;
+import java.io.InputStream;
+import java.lang.reflect.Field;
+import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.util.Base64;
import java.util.HashMap;
@@ -1131,14 +1138,91 @@ public class GremlinServerHttpIntegrateTest extends AbstractGremlinServerIntegra
final CloseableHttpClient httpclient = HttpClients.createDefault();
final HttpPost httppost = new HttpPost(TestClientFactory.createURLString());
httppost.addHeader("Content-Type", "application/json");
- httppost.setEntity(new StringEntity("{\"gremlin\":\"2-1\"}", Consts.UTF_8));
+ // default chunk size is 16, so should be 2 chunks
+ httppost.setEntity(new StringEntity("{\"gremlin\":\"g.inject(0,1,2,3,4,5,6,7,8,9,'ten',11,12,13,14,15,'new chunk')\"}", Consts.UTF_8));
+
+ try (final CloseableHttpResponse response = httpclient.execute(httppost)) {
+ assertEquals(200, response.getStatusLine().getStatusCode());
+ assertTrue(response.getEntity().isChunked());
+
+ String json = EntityUtils.toString(response.getEntity());
+ final JsonNode node = mapper.readTree(json);
+ assertEquals(8, node.get("result").get("data").get(GraphSONTokens.VALUEPROP).get(8).get(GraphSONTokens.VALUEPROP).intValue());
+ assertEquals("ten", node.get("result").get("data").get(GraphSONTokens.VALUEPROP).get(10).textValue());
+ assertEquals("new chunk", node.get("result").get("data").get(GraphSONTokens.VALUEPROP).get(16).textValue());
+
+ final Header[] footers = getTrailingHeaders(response);
+ assertEquals(2, footers.length);
+ assertEquals("code", footers[0].getName());
+ assertEquals("200", footers[0].getValue());
+ assertEquals("message", footers[1].getName());
+ assertEquals("OK", footers[1].getValue());
+ }
+ }
+
+ @Test
+ public void shouldHandleErrorsInFirstChunkPOSTWithChunkedResponse() throws Exception {
+ final CloseableHttpClient httpclient = HttpClients.createDefault();
+ final HttpPost httppost = new HttpPost(TestClientFactory.createURLString());
+ httppost.addHeader("Content-Type", "application/json");
+ // default chunk size is 16, so should be 2 chunks
+ httppost.setEntity(new StringEntity(
+ "{\"gremlin\":\"g.inject(0,1,2,3,4,5,6,7,8,9,'ten',11,12,13,14,15,16).coalesce(is(lt(0)),fail('some error'))\"}", Consts.UTF_8));
+
+ try (final CloseableHttpResponse response = httpclient.execute(httppost)) {
+ assertEquals(200, response.getStatusLine().getStatusCode());
+ assertTrue(response.getEntity().isChunked());
+
+ String json = EntityUtils.toString(response.getEntity());
+ final JsonNode node = mapper.readTree(json);
+ assertEquals("some error", node.get("status").get("message").textValue());
+ assertEquals(500, node.get("status").get("code").intValue());
+
+ final Header[] footers = getTrailingHeaders(response);
+ assertEquals(2, footers.length);
+ assertEquals("code", footers[0].getName());
+ assertEquals("500", footers[0].getValue());
+ assertEquals("message", footers[1].getName());
+ assertEquals("some error", footers[1].getValue());
+ }
+ }
+
+ @Test
+ public void shouldHandleErrorsNotInFirstChunkPOSTWithChunkedResponse() throws Exception {
+ final CloseableHttpClient httpclient = HttpClients.createDefault();
+ final HttpPost httppost = new HttpPost(TestClientFactory.createURLString());
+ httppost.addHeader("Content-Type", "application/json");
+ // default chunk size is 16, so should be 2 chunks
+ httppost.setEntity(new StringEntity(
+ "{\"gremlin\":\"g.inject(0,1,2,3,4,5,6,7,8,9,'ten',11,12,13,14,15,16).coalesce(is(lt(15)),fail('some error'))\"}", Consts.UTF_8));
try (final CloseableHttpResponse response = httpclient.execute(httppost)) {
assertEquals(200, response.getStatusLine().getStatusCode());
assertTrue(response.getEntity().isChunked());
+
+ String json = EntityUtils.toString(response.getEntity());
+ final JsonNode node = mapper.readTree(json);
+ assertEquals(0, node.get("result").get("data").get(GraphSONTokens.VALUEPROP).get(0).get(GraphSONTokens.VALUEPROP).intValue());
+ assertEquals("some error", node.get("status").get("message").textValue());
+ assertEquals(500, node.get("status").get("code").intValue());
+
+ final Header[] footers = getTrailingHeaders(response);
+ assertEquals(2, footers.length);
+ assertEquals("code", footers[0].getName());
+ assertEquals("500", footers[0].getValue());
+ assertEquals("message", footers[1].getName());
+ assertEquals("some error", footers[1].getValue());
}
}
+ private Header[] getTrailingHeaders(final CloseableHttpResponse response) throws IOException, NoSuchFieldException, IllegalAccessException {
+ final EofSensorInputStream content = (EofSensorInputStream) response.getEntity().getContent();
+ final Field field = content.getClass().getDeclaredField("wrappedStream");
+ field.setAccessible(true);
+ final ChunkedInputStream stream = (ChunkedInputStream) field.get(content);
+ return stream.getFooters();
+ }
+
@Test
public void should200OnPOSTWithValidGraphSONBytecodeRequest() throws Exception {
final Bytecode bytecodeQuery = EmptyGraph.instance().traversal().V().asAdmin().getBytecode();
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 b0dbf0b5f0..aa9f2bdf7d 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
@@ -67,6 +67,30 @@ public final class ResponseMessage {
'}';
}
+ public static class ResponseMessageHeader {
+ private final ResponseMessage responseMessage;
+
+ public ResponseMessageHeader(final ResponseMessage responseMessage) {
+ this.responseMessage = responseMessage;
+ }
+
+ public ResponseMessage getResponseMessage() {
+ return responseMessage;
+ }
+ }
+
+ public static class ResponseMessageFooter {
+ private final ResponseMessage responseMessage;
+
+ public ResponseMessageFooter(final ResponseMessage responseMessage) {
+ this.responseMessage = responseMessage;
+ }
+
+ public ResponseMessage getResponseMessage() {
+ return responseMessage;
+ }
+ }
+
public static Builder build(final RequestMessage requestMessage) {
return new Builder(requestMessage);
}
diff --git a/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/AbstractGraphSONMessageSerializerV2.java b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/AbstractGraphSONMessageSerializerV2.java
index 87f82304eb..123ed0daa3 100644
--- a/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/AbstractGraphSONMessageSerializerV2.java
+++ b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/AbstractGraphSONMessageSerializerV2.java
@@ -145,15 +145,15 @@ public abstract class AbstractGraphSONMessageSerializerV2 extends AbstractMessag
}
private GraphSONMapper.Builder applyMaxTokenLimits(final GraphSONMapper.Builder builder, final Map<String, Object> config) {
- if(config != null) {
- if(config.containsKey("maxNumberLength")) {
- builder.maxNumberLength((int)config.get("maxNumberLength"));
+ if (config != null) {
+ if (config.containsKey("maxNumberLength")) {
+ builder.maxNumberLength((int) config.get("maxNumberLength"));
}
- if(config.containsKey("maxStringLength")) {
- builder.maxStringLength((int)config.get("maxStringLength"));
+ if (config.containsKey("maxStringLength")) {
+ builder.maxStringLength((int) config.get("maxStringLength"));
}
- if(config.containsKey("maxNestingDepth")) {
- builder.maxNestingDepth((int)config.get("maxNestingDepth"));
+ if (config.containsKey("maxNestingDepth")) {
+ builder.maxNestingDepth((int) config.get("maxNestingDepth"));
}
}
return builder;
@@ -270,7 +270,7 @@ public abstract class AbstractGraphSONMessageSerializerV2 extends AbstractMessag
GraphSONUtil.writeStartObject(responseMessage, jsonGenerator, typeSerializer);
- if (null == responseMessage.getResult().getData()){
+ if (null == responseMessage.getResult().getData()) {
jsonGenerator.writeNullField(SerTokens.TOKEN_DATA);
} else {
jsonGenerator.writeFieldName(SerTokens.TOKEN_DATA);
@@ -284,12 +284,12 @@ public abstract class AbstractGraphSONMessageSerializerV2 extends AbstractMessag
GraphSONUtil.writeEndObject(responseMessage, jsonGenerator, typeSerializer);
}
}
-
+
public final static class ResponseMessageDeserializer extends AbstractObjectDeserializer<ResponseMessage> {
protected ResponseMessageDeserializer() {
super(ResponseMessage.class);
}
-
+
@Override
public ResponseMessage createObject(final Map<String, Object> data) {
final Map<String, Object> status = (Map<String, Object>) data.get(SerTokens.TOKEN_STATUS);
diff --git a/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/GraphSONMessageSerializerV4.java b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/GraphSONMessageSerializerV4.java
new file mode 100644
index 0000000000..ca981986b2
--- /dev/null
+++ b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/GraphSONMessageSerializerV4.java
@@ -0,0 +1,340 @@
+/*
+ * 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.util.ser;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.util.ReferenceCountUtil;
+import org.apache.tinkerpop.gremlin.structure.io.graphson.GraphSONMapper;
+import org.apache.tinkerpop.gremlin.structure.io.graphson.GraphSONUtil;
+import org.apache.tinkerpop.gremlin.structure.io.graphson.GraphSONVersion;
+import org.apache.tinkerpop.gremlin.structure.io.graphson.GraphSONXModuleV3;
+import org.apache.tinkerpop.gremlin.util.message.RequestMessage;
+import org.apache.tinkerpop.gremlin.util.message.ResponseMessage;
+import org.apache.tinkerpop.shaded.jackson.core.JsonGenerator;
+import org.apache.tinkerpop.shaded.jackson.core.JsonProcessingException;
+import org.apache.tinkerpop.shaded.jackson.databind.ObjectMapper;
+import org.apache.tinkerpop.shaded.jackson.databind.SerializerProvider;
+import org.apache.tinkerpop.shaded.jackson.databind.jsontype.TypeSerializer;
+import org.apache.tinkerpop.shaded.jackson.databind.module.SimpleModule;
+import org.apache.tinkerpop.shaded.jackson.databind.ser.std.StdSerializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+
+/**
+ * Serialize results to JSON with version 3.0.x schema and the extended module.
+ *
+ * @author Stephen Mallette (http://stephen.genoprime.com)
+ */
+public final class GraphSONMessageSerializerV4 extends AbstractGraphSONMessageSerializerV2
+ implements MessageTextSerializer<ObjectMapper>, MessageChunkSerializer<ObjectMapper> {
+
+ private static final Logger logger = LoggerFactory.getLogger(GraphSONMessageSerializerV4.class);
+ private static final String MIME_TYPE = SerTokens.MIME_GRAPHSON_V4;
+
+ private static byte[] header;
+
+ static {
+ final ByteBuffer buffer = ByteBuffer.allocate(MIME_TYPE.length() + 1);
+ buffer.put((byte) MIME_TYPE.length());
+ buffer.put(MIME_TYPE.getBytes());
+ header = buffer.array();
+ }
+
+ /**
+ * Creates a default GraphSONMessageSerializer.
+ * <p>
+ * By default this will internally instantiate a {@link GraphSONMapper} and register
+ * a {@link GremlinServerModule} and {@link GraphSONXModuleV3} to the mapper.
+ *
+ * @see #GraphSONMessageSerializerV4(GraphSONMapper.Builder)
+ */
+ public GraphSONMessageSerializerV4() {
+ super();
+ }
+
+ /**
+ * Create a GraphSONMessageSerializer from a {@link GraphSONMapper}. Deprecated, use
+ * {@link #GraphSONMessageSerializerV4(GraphSONMapper.Builder)} instead.
+ */
+ @Deprecated
+ public GraphSONMessageSerializerV4(final GraphSONMapper mapper) {
+ super(mapper);
+ }
+
+ /**
+ * Create a GraphSONMessageSerializer with a provided {@link GraphSONMapper.Builder}.
+ *
+ * Note that to make this mapper usable in the context of request messages and responses,
+ * this method will automatically register a {@link GremlinServerModule} to the provided
+ * mapper.
+ */
+ public GraphSONMessageSerializerV4(final GraphSONMapper.Builder mapperBuilder) {
+ super(mapperBuilder);
+ }
+
+ @Override
+ public String[] mimeTypesSupported() {
+ return new String[]{MIME_TYPE, SerTokens.MIME_JSON};
+ }
+
+ @Override
+ GraphSONMapper.Builder configureBuilder(final GraphSONMapper.Builder builder) {
+ // override the 2.0 in AbstractGraphSONMessageSerializerV2
+ return builder.version(GraphSONVersion.V4_0).addCustomModule(new GremlinServerModuleV4());
+ }
+
+ @Override
+ byte[] obtainHeader() {
+ return header;
+ }
+
+ @Override
+ public ResponseMessage deserializeResponse(final String msg) throws SerializationException {
+ try {
+ return mapper.readValue(msg, ResponseMessage.class);
+ } catch (Exception ex) {
+ logger.warn(String.format("Response [%s] could not be deserialized by %s.", msg, GraphSONMessageSerializerV4.class.getName()), ex);
+ throw new SerializationException(ex);
+ }
+ }
+
+ @Override
+ public String serializeResponseAsString(final ResponseMessage responseMessage, final ByteBufAllocator allocator) throws SerializationException {
+ try {
+ return mapper.writeValueAsString(responseMessage);
+ } catch (Exception ex) {
+ logger.warn(String.format("Response [%s] could not be serialized by %s.", responseMessage.toString(), GraphSONMessageSerializerV4.class.getName()), ex);
+ throw new SerializationException(ex);
+ }
+ }
+
+ @Override
+ public RequestMessage deserializeRequest(final String msg) throws SerializationException {
+ try {
+ return mapper.readValue(msg, RequestMessage.class);
+ } catch (Exception ex) {
+ logger.warn(String.format("Request [%s] could not be deserialized by %s.", msg, GraphSONMessageSerializerV4.class.getName()), ex);
+ throw new SerializationException(ex);
+ }
+ }
+
+ @Override
+ public String serializeRequestAsString(final RequestMessage requestMessage, final ByteBufAllocator allocator) throws SerializationException {
+ try {
+ return mapper.writeValueAsString(requestMessage);
+ } catch (Exception ex) {
+ logger.warn(String.format("Request [%s] could not be serialized by %s.", requestMessage.toString(), GraphSONMessageSerializerV4.class.getName()), ex);
+ throw new SerializationException(ex);
+ }
+ }
+
+
+ // !!!
+ @Override
+ public ByteBuf writeResponseHeader(final ResponseMessage responseMessage, final ByteBufAllocator allocator) throws SerializationException {
+ ByteBuf encodedMessage = null;
+ try {
+ final byte[] header = mapper.writeValueAsBytes(new ResponseMessage.ResponseMessageHeader(responseMessage));
+ final byte[] result = getChunk(true, responseMessage.getResult().getData());
+ // skip closing }}
+ encodedMessage = allocator.buffer(header.length - 2 + result.length);
+ encodedMessage.writeBytes(header, 0, header.length - 2);
+ encodedMessage.writeBytes(result);
+
+ return encodedMessage;
+ } catch (Exception ex) {
+ if (encodedMessage != null) ReferenceCountUtil.release(encodedMessage);
+
+ logger.warn(String.format("Response [%s] could not be serialized by %s.", responseMessage, AbstractGraphSONMessageSerializerV2.class.getName()), ex);
+ throw new SerializationException(ex);
+ }
+ }
+
+ private byte[] getChunk(final boolean isFirst, final Object aggregate) throws JsonProcessingException {
+ // todo: cleanup
+ List asList = (List)aggregate;
+ Object[] array = new Object[asList.size()];
+ asList.toArray(array);
+ String str = mapper.writeValueAsString(array);
+ str = str.substring(1, str.length() - 1);
+ if (!isFirst) {
+ str = "," + str;
+ }
+ return str.getBytes();
+ }
+
+ @Override
+ public ByteBuf writeResponseChunk(final Object aggregate, final ByteBufAllocator allocator) throws SerializationException {
+ ByteBuf encodedMessage = null;
+ try {
+ final byte[] payload = getChunk(false, aggregate);
+ encodedMessage = allocator.buffer(payload.length);
+ encodedMessage.writeBytes(payload);
+
+ return encodedMessage;
+ } catch (Exception ex) {
+ if (encodedMessage != null) ReferenceCountUtil.release(encodedMessage);
+
+ logger.warn(String.format("Response [%s] could not be serialized by %s.", aggregate, GraphSONMessageSerializerV4.class.getName()), ex);
+ throw new SerializationException(ex);
+ }
+ }
+
+ @Override
+ public ByteBuf writeResponseFooter(final ResponseMessage responseMessage, final ByteBufAllocator allocator) throws SerializationException {
+ ByteBuf encodedMessage = null;
+ try {
+ final byte[] footer = mapper.writeValueAsBytes(new ResponseMessage.ResponseMessageFooter(responseMessage));
+ final byte[] result = getChunk(false, responseMessage.getResult().getData());
+ // skip opening {
+ encodedMessage = allocator.buffer(footer.length - 1 + result.length);
+ encodedMessage.writeBytes(result);
+ encodedMessage.writeBytes(footer, 1, footer.length - 1);
+
+ return encodedMessage;
+ } catch (Exception ex) {
+ if (encodedMessage != null) ReferenceCountUtil.release(encodedMessage);
+
+ logger.warn(String.format("Response [%s] could not be serialized by %s.", responseMessage, GraphSONMessageSerializerV4.class.getName()), ex);
+ throw new SerializationException(ex);
+ }
+ }
+
+ @Override
+ public ByteBuf writeError(final ResponseMessage responseMessage, final ByteBufAllocator allocator) throws SerializationException {
+ ByteBuf encodedMessage = null;
+ try {
+ final byte[] footer = mapper.writeValueAsBytes(new ResponseMessage.ResponseMessageFooter(responseMessage));
+ // skip opening {
+ encodedMessage = allocator.buffer(footer.length - 1);
+ encodedMessage.writeBytes(footer, 1, footer.length - 1);
+
+ return encodedMessage;
+ } catch (Exception ex) {
+ if (encodedMessage != null) ReferenceCountUtil.release(encodedMessage);
+
+ logger.warn(String.format("Response [%s] could not be serialized by %s.", responseMessage, GraphSONMessageSerializerV4.class.getName()), ex);
+ throw new SerializationException(ex);
+ }
+ }
+
+ public final static class ResponseMessageHeaderSerializer extends StdSerializer<ResponseMessage.ResponseMessageHeader> {
+ public ResponseMessageHeaderSerializer() {
+ super(ResponseMessage.ResponseMessageHeader.class);
+ }
+
+ @Override
+ public void serialize(final ResponseMessage.ResponseMessageHeader responseMessage, final JsonGenerator jsonGenerator,
+ final SerializerProvider serializerProvider) throws IOException {
+ ser(responseMessage, jsonGenerator, serializerProvider, null);
+ }
+
+ @Override
+ public void serializeWithType(final ResponseMessage.ResponseMessageHeader responseMessage, final JsonGenerator jsonGenerator,
+ final SerializerProvider serializerProvider,
+ final TypeSerializer typeSerializer) throws IOException {
+ ser(responseMessage, jsonGenerator, serializerProvider, typeSerializer);
+ }
+
+ public void ser(final ResponseMessage.ResponseMessageHeader responseMessageHeader, final JsonGenerator jsonGenerator,
+ final SerializerProvider serializerProvider,
+ final TypeSerializer typeSerializer) throws IOException {
+ final ResponseMessage responseMessage = responseMessageHeader.getResponseMessage();
+
+ GraphSONUtil.writeStartObject(responseMessage, jsonGenerator, typeSerializer);
+
+ jsonGenerator.writeStringField(SerTokens.TOKEN_REQUEST, responseMessage.getRequestId() != null ? responseMessage.getRequestId().toString() : null);
+
+ jsonGenerator.writeFieldName(SerTokens.TOKEN_RESULT);
+
+ GraphSONUtil.writeStartObject(responseMessage, jsonGenerator, typeSerializer);
+ jsonGenerator.writeFieldName(SerTokens.TOKEN_DATA);
+ jsonGenerator.writeRaw(":{\"@type\":\"g:List\",\"@value\":[");
+
+ // jsonGenerator will add 2 closing }
+ // jsonGenerator.configure(JsonGenerator.Feature.AUTO_CLOSE_JSON_CONTENT, false);
+ }
+ }
+
+ public final static class ResponseMessageFooterSerializer extends StdSerializer<ResponseMessage.ResponseMessageFooter> {
+ public ResponseMessageFooterSerializer() {
+ super(ResponseMessage.ResponseMessageFooter.class);
+ }
+
+ @Override
+ public void serialize(final ResponseMessage.ResponseMessageFooter responseMessage, final JsonGenerator jsonGenerator,
+ final SerializerProvider serializerProvider) throws IOException {
+ ser(responseMessage, jsonGenerator, serializerProvider, null);
+ }
+
+ @Override
+ public void serializeWithType(final ResponseMessage.ResponseMessageFooter responseMessage, final JsonGenerator jsonGenerator,
+ final SerializerProvider serializerProvider,
+ final TypeSerializer typeSerializer) throws IOException {
+ ser(responseMessage, jsonGenerator, serializerProvider, typeSerializer);
+ }
+
+ public void ser(final ResponseMessage.ResponseMessageFooter responseMessageFooter, final JsonGenerator jsonGenerator,
+ final SerializerProvider serializerProvider,
+ final TypeSerializer typeSerializer) throws IOException {
+ final ResponseMessage responseMessage = responseMessageFooter.getResponseMessage();
+
+ // todo: find a way to get rid off
+ GraphSONUtil.writeStartObject(responseMessage, jsonGenerator, typeSerializer);
+
+ // close result field
+ jsonGenerator.writeRaw("]},");
+
+ jsonGenerator.writeObjectField(SerTokens.TOKEN_META, responseMessage.getResult().getMeta());
+
+ jsonGenerator.writeRaw("}");
+
+ jsonGenerator.writeFieldName(SerTokens.TOKEN_STATUS);
+ GraphSONUtil.writeStartObject(responseMessage, jsonGenerator, typeSerializer);
+ jsonGenerator.writeStringField(SerTokens.TOKEN_MESSAGE, responseMessage.getStatus().getMessage());
+ jsonGenerator.writeNumberField(SerTokens.TOKEN_CODE, responseMessage.getStatus().getCode().getValue());
+ jsonGenerator.writeObjectField(SerTokens.TOKEN_ATTRIBUTES, responseMessage.getStatus().getAttributes());
+ GraphSONUtil.writeEndObject(responseMessage, jsonGenerator, typeSerializer);
+
+ GraphSONUtil.writeEndObject(responseMessage, jsonGenerator, typeSerializer);
+ }
+ }
+
+ public final static class GremlinServerModuleV4 extends SimpleModule {
+ public GremlinServerModuleV4() {
+ super("graphson-gremlin-server");
+
+ // SERIALIZERS
+ addSerializer(ResponseMessage.class, new ResponseMessageSerializer());
+ addSerializer(ResponseMessage.ResponseMessageHeader.class, new ResponseMessageHeaderSerializer());
+ addSerializer(ResponseMessage.ResponseMessageFooter.class, new ResponseMessageFooterSerializer());
+ addSerializer(RequestMessage.class, new RequestMessageSerializer());
+
+ //DESERIALIZERS
+ addDeserializer(ResponseMessage.class, new ResponseMessageDeserializer());
+ addDeserializer(RequestMessage.class, new RequestMessageDeserializer());
+ }
+ }
+
+}
diff --git a/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/GraphSONUntypedMessageSerializerV3.java b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/GraphSONUntypedMessageSerializerV3.java
index 9dffa216c3..f10828ed2a 100644
--- a/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/GraphSONUntypedMessageSerializerV3.java
+++ b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/GraphSONUntypedMessageSerializerV3.java
@@ -18,6 +18,7 @@
*/
package org.apache.tinkerpop.gremlin.util.ser;
+import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import org.apache.tinkerpop.gremlin.structure.io.graphson.GraphSONMapper;
import org.apache.tinkerpop.gremlin.structure.io.graphson.GraphSONXModuleV2;
diff --git a/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/MessageChunkSerializer.java b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/MessageChunkSerializer.java
new file mode 100644
index 0000000000..3167b185cf
--- /dev/null
+++ b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/MessageChunkSerializer.java
@@ -0,0 +1,34 @@
+/*
+ * 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.util.ser;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import org.apache.tinkerpop.gremlin.util.MessageSerializer;
+import org.apache.tinkerpop.gremlin.util.message.ResponseMessage;
+
+public interface MessageChunkSerializer<M> extends MessageSerializer<M> {
+ public ByteBuf writeResponseHeader(final ResponseMessage responseMessage, final ByteBufAllocator allocator) throws SerializationException;
+
+ public ByteBuf writeResponseChunk(final Object aggregate, final ByteBufAllocator allocator) throws SerializationException;
+
+ public ByteBuf writeResponseFooter(final ResponseMessage responseMessage, final ByteBufAllocator allocator) throws SerializationException;
+
+ public ByteBuf writeError(final ResponseMessage responseMessage, final ByteBufAllocator allocator) throws SerializationException;
+}
diff --git a/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/SerTokens.java b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/SerTokens.java
index 8be6d74848..8e20f531af 100644
--- a/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/SerTokens.java
+++ b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/SerTokens.java
@@ -43,5 +43,6 @@ public final class SerTokens {
public static final String MIME_GRAPHSON_V2_UNTYPED = "application/vnd.gremlin-v2.0+json;types=false";
public static final String MIME_GRAPHSON_V3 = "application/vnd.gremlin-v3.0+json";
public static final String MIME_GRAPHSON_V3_UNTYPED = "application/vnd.gremlin-v3.0+json;types=false";
+ public static final String MIME_GRAPHSON_V4 = "application/vnd.gremlin-v4.0+json";
public static final String MIME_GRAPHBINARY_V1 = "application/vnd.graphbinary-v1.0";
}
diff --git a/gremlin-util/src/test/java/org/apache/tinkerpop/gremlin/util/ser/GraphSONMessageSerializerV3Test.java b/gremlin-util/src/test/java/org/apache/tinkerpop/gremlin/util/ser/GraphSONMessageSerializerV3Test.java
index 584af0649d..24cf8391bf 100644
--- a/gremlin-util/src/test/java/org/apache/tinkerpop/gremlin/util/ser/GraphSONMessageSerializerV3Test.java
+++ b/gremlin-util/src/test/java/org/apache/tinkerpop/gremlin/util/ser/GraphSONMessageSerializerV3Test.java
@@ -69,9 +69,9 @@ import static org.junit.Assert.fail;
@SuppressWarnings("unchecked")
public class GraphSONMessageSerializerV3Test {
- private final UUID requestId = UUID.fromString("6457272A-4018-4538-B9AE-08DD5DDC0AA1");
- private final ResponseMessage.Builder responseMessageBuilder = ResponseMessage.build(requestId);
- private final static ByteBufAllocator allocator = UnpooledByteBufAllocator.DEFAULT;
+ protected final UUID requestId = UUID.fromString("6457272A-4018-4538-B9AE-08DD5DDC0AA1");
+ protected final ResponseMessage.Builder responseMessageBuilder = ResponseMessage.build(requestId);
+ protected final static ByteBufAllocator allocator = UnpooledByteBufAllocator.DEFAULT;
public final GraphSONMessageSerializerV3 serializer = new GraphSONMessageSerializerV3();
diff --git a/gremlin-util/src/test/java/org/apache/tinkerpop/gremlin/util/ser/GraphSONMessageSerializerV4Test.java b/gremlin-util/src/test/java/org/apache/tinkerpop/gremlin/util/ser/GraphSONMessageSerializerV4Test.java
new file mode 100644
index 0000000000..67db315a20
--- /dev/null
+++ b/gremlin-util/src/test/java/org/apache/tinkerpop/gremlin/util/ser/GraphSONMessageSerializerV4Test.java
@@ -0,0 +1,141 @@
+/*
+ * 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.util.ser;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.UnpooledByteBufAllocator;
+import io.netty.util.CharsetUtil;
+import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
+import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
+import org.apache.tinkerpop.gremlin.process.traversal.step.util.Tree;
+import org.apache.tinkerpop.gremlin.structure.Edge;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.Property;
+import org.apache.tinkerpop.gremlin.structure.Vertex;
+import org.apache.tinkerpop.gremlin.structure.VertexProperty;
+import org.apache.tinkerpop.gremlin.structure.io.graphson.GraphSONMapper;
+import org.apache.tinkerpop.gremlin.structure.io.graphson.GraphSONXModuleV3;
+import org.apache.tinkerpop.gremlin.structure.util.empty.EmptyGraph;
+import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerFactory;
+import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerGraph;
+import org.apache.tinkerpop.gremlin.util.MessageSerializer;
+import org.apache.tinkerpop.gremlin.util.Tokens;
+import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
+import org.apache.tinkerpop.gremlin.util.message.RequestMessage;
+import org.apache.tinkerpop.gremlin.util.message.ResponseMessage;
+import org.apache.tinkerpop.gremlin.util.message.ResponseStatusCode;
+import org.apache.tinkerpop.shaded.jackson.databind.JsonMappingException;
+import org.junit.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.apache.tinkerpop.gremlin.util.MockitoHamcrestMatcherAdapter.reflectionEquals;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasItemInArray;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/**
+ * Serializer tests that cover non-lossy serialization/deserialization methods.
+ *
+ * @author Stephen Mallette (http://stephen.genoprime.com)
+ */
+@SuppressWarnings("unchecked")
+public class GraphSONMessageSerializerV4Test extends GraphSONMessageSerializerV3Test {
+
+ public final GraphSONMessageSerializerV4 serializer = new GraphSONMessageSerializerV4();
+
+ @Test
+ public void shouldDeserializeChunkedResponseMessage() throws SerializationException {
+ final UUID id = UUID.randomUUID();
+ final ResponseMessage response = ResponseMessage.build(id)
+ .code(ResponseStatusCode.SUCCESS)
+ .result(Arrays.asList("start/end", 0))
+ .statusMessage("OK")
+ .create();
+
+ final ByteBuf bb0 = serializer.writeResponseHeader(response, allocator);
+ final ByteBuf bb1 = serializer.writeResponseChunk(Arrays.asList("chunk", 1), allocator);
+ final ByteBuf bb2 = serializer.writeResponseChunk(Arrays.asList("chunk", 2), allocator);
+ final ByteBuf bb3 = serializer.writeResponseFooter(response, allocator);
+
+ final ByteBuf bbCombined = allocator.buffer(bb0.readableBytes() + bb1.readableBytes() +
+ bb2.readableBytes() + bb3.readableBytes());
+
+ bbCombined.writeBytes(bb0);
+ bbCombined.writeBytes(bb1);
+ bbCombined.writeBytes(bb2);
+ bbCombined.writeBytes(bb3);
+
+ final ResponseMessage deserialized = serializer.deserializeResponse(bbCombined);
+
+ assertEquals(8, ((List)deserialized.getResult().getData()).size());
+ assertEquals(200, deserialized.getStatus().getCode().getValue());
+ assertEquals("OK", deserialized.getStatus().getMessage());
+ }
+
+ @Test
+ public void shouldDeserializeChunkedResponseMessageWithError() throws SerializationException {
+ final UUID id = UUID.randomUUID();
+ final ResponseMessage response = ResponseMessage.build(id)
+ .code(ResponseStatusCode.SUCCESS)
+ .result(Arrays.asList("start/end", 0))
+ .statusMessage("OK")
+ .create();
+
+ final ByteBuf bb0 = serializer.writeResponseHeader(response, allocator);
+ final ByteBuf bb1 = serializer.writeResponseChunk(Arrays.asList("chunk", 1), allocator);
+ final ByteBuf bb2 = serializer.writeResponseChunk(Arrays.asList("chunk", 2), allocator);
+ final ByteBuf bb3 = serializer.writeError(response, allocator);
+
+ final ByteBuf bbCombined = allocator.buffer(bb0.readableBytes() + bb1.readableBytes() +
+ bb2.readableBytes() + bb3.readableBytes());
+
+ bbCombined.writeBytes(bb0);
+ bbCombined.writeBytes(bb1);
+ bbCombined.writeBytes(bb2);
+ bbCombined.writeBytes(bb3);
+
+ final ResponseMessage deserialized = serializer.deserializeResponse(bbCombined);
+
+ // 3 chunks without errors, 2 items in each
+ assertEquals(6, ((List)deserialized.getResult().getData()).size());
+ // error description is in trailing headers
+ assertEquals(200, deserialized.getStatus().getCode().getValue());
+ assertEquals("OK", deserialized.getStatus().getMessage());
+ }
+
+ private static String getString(final ByteBuf bb) {
+ final String result = bb.readCharSequence(bb.readableBytes(), CharsetUtil.UTF_8).toString();
+ bb.resetReaderIndex();
+ return result;
+ }
+}