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/04/04 18:58:08 UTC
(tinkerpop) branch master-http updated: HttpGremlinEndpointHandler cleanup (#2540)
This is an automated email from the ASF dual-hosted git repository.
valentyn pushed a commit to branch master-http
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
The following commit(s) were added to refs/heads/master-http by this push:
new a2c483aaca HttpGremlinEndpointHandler cleanup (#2540)
a2c483aaca is described below
commit a2c483aacaa2528ee18a49df5a1d7a12d4839750
Author: Valentyn Kahamlyk <vk...@users.noreply.github.com>
AuthorDate: Thu Apr 4 11:58:00 2024 -0700
HttpGremlinEndpointHandler cleanup (#2540)
---
.../apache/tinkerpop/gremlin/server/Context.java | 103 +---------
.../tinkerpop/gremlin/server/handler/Frame.java | 48 -----
.../server/handler/HttpGremlinEndpointHandler.java | 220 +++++++++------------
.../gremlin/server/handler/HttpHandlerUtil.java | 19 +-
.../tinkerpop/gremlin/server/ContextTest.java | 129 ------------
.../server/GremlinServerHttpIntegrateTest.java | 4 +-
6 files changed, 103 insertions(+), 420 deletions(-)
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 34559756d0..0cca811a83 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
@@ -18,22 +18,17 @@
*/
package org.apache.tinkerpop.gremlin.server;
+import io.netty.channel.ChannelHandlerContext;
+import org.apache.tinkerpop.gremlin.groovy.engine.GremlinExecutor;
+import org.apache.tinkerpop.gremlin.jsr223.GremlinScriptChecker;
+import org.apache.tinkerpop.gremlin.process.traversal.Bytecode;
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.Graph;
import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceFactory;
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.groovy.engine.GremlinExecutor;
-import io.netty.channel.ChannelHandlerContext;
-import org.apache.tinkerpop.gremlin.jsr223.GremlinScriptChecker;
-import org.apache.tinkerpop.gremlin.process.traversal.Bytecode;
-import org.apache.tinkerpop.gremlin.server.handler.Frame;
-import org.apache.tinkerpop.gremlin.structure.Graph;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import java.util.List;
import java.util.Map;
@@ -48,14 +43,12 @@ import java.util.concurrent.atomic.AtomicBoolean;
* @author Stephen Mallette (http://stephen.genoprime.com)
*/
public class Context {
- private static final Logger logger = LoggerFactory.getLogger(Context.class);
private final RequestMessage requestMessage;
private final ChannelHandlerContext channelHandlerContext;
private final Settings settings;
private final GraphManager graphManager;
private final GremlinExecutor gremlinExecutor;
private final ScheduledExecutorService scheduledExecutorService;
- private final AtomicBoolean finalResponseWritten = new AtomicBoolean();
private final long requestTimeout;
private final String materializeProperties;
private final RequestContentType requestContentType;
@@ -145,18 +138,6 @@ public class Context {
return materializeProperties;
}
- public boolean isFinalResponseWritten() {
- return this.finalResponseWritten.get();
- }
-
- public RequestContentType getRequestContentType() {
- return requestContentType;
- }
-
- public Object getGremlinArgument() {
- return gremlinArgument;
- }
-
public ScheduledExecutorService getScheduledExecutorService() {
return scheduledExecutorService;
}
@@ -207,80 +188,6 @@ public class Context {
*/
public void setStartedResponse() { startedResponse.set(true); }
- /**
- * Writes a default timeout error response message to the underlying channel.
- */
- public void sendTimeoutResponse() {
- sendTimeoutResponse(String.format("A timeout occurred during traversal evaluation of [%s] - consider increasing the limit given to evaluationTimeout", requestMessage));
- }
-
- /**
- * Writes a specific timeout error response message to the underlying channel.
- */
- public void sendTimeoutResponse(final String message) {
- logger.warn(message);
- writeAndFlush(ResponseMessage.build(requestMessage)
- .code(ResponseStatusCode.SERVER_ERROR_TIMEOUT)
- .statusMessage(message)
- .statusAttributeException(new InterruptedException()).create());
- }
-
- /**
- * Writes a response message to the underlying channel while ensuring that at most one
- * {@link ResponseStatusCode#isFinalResponse() final} response is written.
- * <p>Note: this method should be used instead of writing to the channel directly when multiple threads
- * are expected to produce response messages concurrently.</p>
- * <p>Attempts to write more than one final response message will be ignored.</p>
- * @see #writeAndFlush(ResponseStatusCode, Object)
- */
- public void writeAndFlush(final ResponseMessage message) {
- writeAndFlush(message.getStatus().getCode(), message);
- }
-
- /**
- * Writes a response message to the underlying channel while ensuring that at most one
- * {@link ResponseStatusCode#isFinalResponse() final} response is written.
- * <p>The caller must make sure that the provided response status code matches the content of the message.</p>
- * <p>Note: this method should be used instead of writing to the channel directly when multiple threads
- * are expected to produce response messages concurrently.</p>
- * <p>Attempts to write more than one final response message will be ignored.</p>
- * @see #writeAndFlush(ResponseMessage)
- */
- public void writeAndFlush(final ResponseStatusCode code, final Object responseMessage) {
- writeAndMaybeFlush(code, responseMessage, true);
- }
-
- public void write(final ResponseMessage message) {
- this.write(message.getStatus().getCode(), message);
- }
-
- public void write(final ResponseStatusCode code, final Object responseMessage) {
- writeAndMaybeFlush(code, responseMessage, false);
- }
-
- /**
- * Flushes messages to the underlying transport.
- */
- public void flush() {
- this.getChannelHandlerContext().flush();
- }
-
- private void writeAndMaybeFlush(final ResponseStatusCode code, final Object responseMessage, final boolean flush) {
- final boolean messageIsFinal = code.isFinalResponse();
- if (finalResponseWritten.compareAndSet(false, messageIsFinal)) {
- this.getChannelHandlerContext().write(responseMessage);
- if (flush) this.getChannelHandlerContext().flush();
- } else {
- if (responseMessage instanceof Frame) {
- ((Frame) responseMessage).tryRelease();
- }
-
- final String logMessage = String.format("Another final response message was already written for request %s, ignoring response code: %s",
- this.getRequestMessage().getRequestId(), code);
- logger.warn(logMessage);
- }
- }
-
private RequestContentType determineRequestContents() {
if (gremlinArgument instanceof Bytecode)
return RequestContentType.BYTECODE;
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/Frame.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/Frame.java
deleted file mode 100644
index 76d772e1a0..0000000000
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/Frame.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * 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.server.handler;
-
-import io.netty.util.ReferenceCounted;
-
-/**
- * A holder for a {@code String} or {@code ByteBuf} that represents a message to be written back to the requesting
- * client.
- *
- * @author Stephen Mallette (http://stephen.genoprime.com)
- */
-public class Frame {
- private final Object msg;
-
- public Frame(final Object msg) {
- this.msg = msg;
- }
-
- public Object getMsg() {
- return msg;
- }
-
- /**
- * If the object contained in the frame is {@code ReferenceCounted} then it may need to be released or else
- * Netty will generate warnings that counted resources are leaking.
- */
- public void tryRelease() {
- if (msg instanceof ReferenceCounted)
- ((ReferenceCounted) msg).release();
- }
-}
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 eac946e578..2b2f5342a4 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
@@ -68,7 +68,6 @@ 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;
import org.javatuples.Pair;
@@ -82,7 +81,6 @@ import java.lang.reflect.UndeclaredThrowableException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
-import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
@@ -114,9 +112,10 @@ import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
import static org.apache.tinkerpop.gremlin.server.handler.HttpGremlinEndpointHandler.RequestState.CHUNKING_NOT_SUPPORTED;
import static org.apache.tinkerpop.gremlin.server.handler.HttpGremlinEndpointHandler.RequestState.FINISHED;
import static org.apache.tinkerpop.gremlin.server.handler.HttpGremlinEndpointHandler.RequestState.FINISHING;
+import static org.apache.tinkerpop.gremlin.server.handler.HttpGremlinEndpointHandler.RequestState.NOT_STARTED;
import static org.apache.tinkerpop.gremlin.server.handler.HttpGremlinEndpointHandler.RequestState.STREAMING;
import static org.apache.tinkerpop.gremlin.server.handler.HttpHandlerUtil.sendTrailingHeaders;
-import static org.apache.tinkerpop.gremlin.server.handler.HttpHandlerUtil.writeErrorFrame;
+import static org.apache.tinkerpop.gremlin.server.handler.HttpHandlerUtil.writeError;
/**
* Handler that processes HTTP requests to the HTTP Gremlin endpoint.
@@ -209,7 +208,7 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter {
final RequestMessage requestMessage;
try {
requestMessage = HttpHandlerUtil.getRequestMessageFromHttpRequest(req, serializers);
- } catch (IllegalArgumentException|SerializationException|NullPointerException ex) {
+ } catch (IllegalArgumentException | SerializationException | NullPointerException ex) {
HttpHandlerUtil.sendError(ctx, BAD_REQUEST, ex.getMessage(), keepAlive);
ReferenceCountUtil.release(msg);
return;
@@ -231,7 +230,7 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter {
ReferenceCountUtil.release(msg);
final RequestState requestState = serializer.getValue1() instanceof MessageChunkSerializer
- ? RequestState.NOT_STARTED
+ ? NOT_STARTED
: CHUNKING_NOT_SUPPORTED;
final Context requestCtx = new Context(requestMessage, ctx, settings, graphManager, gremlinExecutor,
@@ -302,21 +301,21 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter {
}
} catch (ProcessingException ope) {
logger.warn(ope.getMessage(), ope);
- writeErrorFrame(ctx, requestCtx, ope.getResponseMessage(), serializer.getValue1());
+ writeError(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, requestCtx,
+ writeError(requestCtx,
ResponseMessage.build(requestId)
.code(ResponseStatusCode.SERVER_ERROR)
.statusMessage(t.getMessage() + " - increase the maxContentLength")
.create(),
serializer.getValue1());
} else {
- writeErrorFrame(ctx, requestCtx,
+ writeError(requestCtx,
ResponseMessage.build(requestId)
.code(ResponseStatusCode.SERVER_ERROR)
.statusMessage((t != null) ? t.getMessage() : ex.getMessage())
@@ -345,13 +344,19 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter {
requestCtx.setTimeoutExecutor(requestCtx.getScheduledExecutorService().schedule(() -> {
executionFuture.cancel(true);
if (!requestCtx.getStartedResponse()) {
- requestCtx.sendTimeoutResponse();
+ final String errorMessage = String.format("A timeout occurred during traversal evaluation of [%s] - consider increasing the limit given to evaluationTimeout", requestMessage);
+ writeError(requestCtx,
+ ResponseMessage.build(requestMessage)
+ .code(ResponseStatusCode.SERVER_ERROR_TIMEOUT)
+ .statusMessage(errorMessage)
+ .create(),
+ serializer.getValue1());
}
}, seto, TimeUnit.MILLISECONDS));
}
} catch (RejectedExecutionException ree) {
- writeErrorFrame(requestCtx.getChannelHandlerContext(), requestCtx,
- ResponseMessage.build(requestMessage).code(ResponseStatusCode.TOO_MANY_REQUESTS) .statusMessage("Rate limiting").create(),
+ writeError(requestCtx,
+ ResponseMessage.build(requestMessage).code(ResponseStatusCode.TOO_MANY_REQUESTS).statusMessage("Rate limiting").create(),
serializer.getValue1());
}
}
@@ -465,7 +470,7 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter {
final RequestMessage msg = ctx.getRequestMessage();
if (msg.getArgs().containsKey(Tokens.ARGS_ALIASES)) {
final Map<String, String> aliases = (Map<String, String>) msg.getArgs().get(Tokens.ARGS_ALIASES);
- for (Map.Entry<String,String> aliasKv : aliases.entrySet()) {
+ for (Map.Entry<String, String> aliasKv : aliases.entrySet()) {
boolean found = false;
// first check if the alias refers to a Graph instance
@@ -553,24 +558,22 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter {
specialResponseMsg.code(ResponseStatusCode.SERVER_ERROR_FAIL_STEP).
statusAttribute(Tokens.STATUS_ATTRIBUTE_FAIL_STEP_MESSAGE, failure.format());
}
- writeErrorFrame(context.getChannelHandlerContext(), context, specialResponseMsg.create(), serializer);
+ writeError(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,
+ writeError(context,
ResponseMessage.build(msg)
- .code(ResponseStatusCode.SERVER_ERROR_TIMEOUT)
- .statusMessage(errorMessage)
- .statusAttributeException(ex)
- .create(),
+ .code(ResponseStatusCode.SERVER_ERROR_TIMEOUT)
+ .statusMessage(errorMessage)
+ .statusAttributeException(ex)
+ .create(),
serializer);
} else if (t instanceof TimedInterruptTimeoutException) {
// occurs when the TimedInterruptCustomizerProvider is in play
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,
+ writeError(context,
ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_TIMEOUT)
.statusMessage("Timeout during script evaluation triggered by TimedInterruptCustomizerProvider")
.statusAttributeException(t).create(),
@@ -578,8 +581,7 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter {
} else if (t instanceof TimeoutException) {
final String errorMessage = String.format("Script evaluation exceeded the configured threshold for request [%s]", msg);
logger.warn(errorMessage, t);
- writeErrorFrame(context.getChannelHandlerContext(),
- context,
+ writeError(context,
ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_TIMEOUT)
.statusMessage(t.getMessage())
.statusAttributeException(t).create(),
@@ -588,16 +590,14 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter {
((MultipleCompilationErrorsException) t).getErrorCollector().getErrorCount() == 1) {
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,
+ writeError(context,
ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_EVALUATION)
.statusMessage(errorMessage)
.statusAttributeException(t).create(),
serializer);
} else {
logger.warn(String.format("Exception processing a Traversal on iteration for request [%s].", msg.getRequestId()), ex);
- writeErrorFrame(context.getChannelHandlerContext(),
- context,
+ writeError(context,
ResponseMessage.build(msg)
.code(ResponseStatusCode.SERVER_ERROR)
.statusMessage(ex.getMessage())
@@ -608,8 +608,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,
+ writeError(context,
ResponseMessage.build(msg)
.code(ResponseStatusCode.SERVER_ERROR)
.statusMessage(t.getMessage())
@@ -631,9 +630,16 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter {
// we have an empty iterator - happens on stuff like: g.V().iterate()
if (!itty.hasNext()) {
- // as there is nothing left to iterate if we are transaction managed then we should execute a
- // commit here before we send back a NO_CONTENT which implies success
- sendTrailingHeaders(nettyContext, 206, "NO_CONTENT");
+ ByteBuf chunk = null;
+ try {
+ chunk = makeChunk(context, msg, serializer, new ArrayList<>(), false);
+ nettyContext.writeAndFlush(new DefaultHttpContent(chunk));
+ } catch (Exception ex) {
+ // Bytebuf is a countable release - if it does not get written downstream
+ // it needs to be released here
+ if (chunk != null) chunk.release();
+ }
+ sendTrailingHeaders(nettyContext, ResponseStatusCode.SUCCESS, "OK");
return;
}
@@ -681,21 +687,13 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter {
// results till the timeout. checking for isActive() should help prevent that.
if (nettyContext.channel().isActive() && nettyContext.channel().isWritable()) {
if (aggregate.size() == resultIterationBatchSize || !itty.hasNext()) {
- final ResponseStatusCode code = itty.hasNext() ? ResponseStatusCode.PARTIAL_CONTENT : ResponseStatusCode.SUCCESS;
-
- // serialize here because in sessionless requests the serialization must occur in the same
- // thread as the eval. as eval occurs in the GremlinExecutor there's no way to get back to the
- // thread that processed the eval of the script so, we have to push serialization down into that
- final Map<String, Object> metadata = Collections.emptyMap();
- final Map<String, Object> statusAttrb = generateStatusAttributes(nettyContext, msg, code, itty, settings);
- Frame frame = null;
+ ByteBuf chunk = null;
try {
- frame = makeFrame(context, msg, serializer, true, aggregate, code,
- metadata, statusAttrb);
+ chunk = makeChunk(context, msg, serializer, aggregate, itty.hasNext());
} catch (Exception ex) {
- // a frame may use a Bytebuf which is a countable release - if it does not get written
- // downstream it needs to be released here
- if (frame != null) frame.tryRelease();
+ // Bytebuf is a countable release - if it does not get written downstream
+ // it needs to be released here
+ if (chunk != null) chunk.release();
// exception is handled in makeFrame() - serialization error gets written back to driver
// at that point
@@ -713,20 +711,16 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter {
aggregate = new ArrayList<>(resultIterationBatchSize);
}
} catch (Exception ex) {
- // a frame may use a Bytebuf which is a countable release - if it does not get written
- // downstream it needs to be released here
- if (frame != null) frame.tryRelease();
+ // Bytebuf is a countable release - if it does not get written downstream
+ // it needs to be released here
+ if (chunk != null) chunk.release();
throw ex;
}
- // the flush is called after the commit has potentially occurred. in this way, if a commit was
- // required then it will be 100% complete before the client receives it. the "frame" at this point
- // should have completely detached objects from the transaction (i.e. serialization has occurred)
- // so a new one should not be opened on the flush down the netty pipeline
- context.writeAndFlush(code, new DefaultHttpContent((ByteBuf) frame.getMsg()));
+ nettyContext.writeAndFlush(new DefaultHttpContent(chunk));
if (!hasMore) {
- sendTrailingHeaders(nettyContext, 200, "OK");
+ sendTrailingHeaders(nettyContext, ResponseStatusCode.SUCCESS, "OK");
}
}
} else {
@@ -752,100 +746,66 @@ public class HttpGremlinEndpointHandler extends ChannelInboundHandlerAdapter {
filter(i -> i instanceof TemporaryException || i instanceof Failure).findFirst();
}
- /**
- * Generates response status meta-data to put on a {@link ResponseMessage}.
- *
- * @param itty a reference to the current {@link Iterator} of results - it is not meant to be forwarded in
- * this method
- */
- private Map<String, Object> generateStatusAttributes(final ChannelHandlerContext ctx, final RequestMessage msg,
- final ResponseStatusCode code, final Iterator itty,
- final Settings settings) {
- // only return server metadata on the last message
- if (itty.hasNext()) return Collections.emptyMap();
-
- final Map<String, Object> metaData = new HashMap<>();
- metaData.put(Tokens.ARGS_HOST, String.valueOf(ctx.channel().remoteAddress()));
-
- return metaData;
- }
-
- private static Frame makeFrame(final Context ctx, final RequestMessage msg,
- final MessageSerializer<?> serializer, final boolean useBinary, final List<Object> aggregate,
- final ResponseStatusCode code, final Map<String,Object> responseMetaData,
- final Map<String,Object> statusAttributes) throws Exception {
+ private static ByteBuf makeChunk(final Context ctx, final RequestMessage msg,
+ final MessageSerializer<?> serializer, final List<Object> aggregate,
+ final boolean hasMore) throws Exception {
try {
final ChannelHandlerContext nettyContext = ctx.getChannelHandlerContext();
ctx.handleDetachment(aggregate);
- if (useBinary) {
- if (code == ResponseStatusCode.SUCCESS && ctx.getRequestState() == STREAMING) {
- ctx.setRequestState(FINISHING);
- }
-
- ResponseMessage responseMessage = null;
-
- // for this state no need to build full ResponseMessage
- if (ctx.getRequestState() != STREAMING) {
- final ResponseMessage.Builder builder = ResponseMessage.buildV4(msg.getRequestId())
- .statusAttributes(statusAttributes)
- .responseMetaData(responseMetaData)
- .result(aggregate);
+ if (!hasMore && ctx.getRequestState() == STREAMING) {
+ ctx.setRequestState(FINISHING);
+ }
- // need to put status in last message
- if (ctx.getRequestState() == FINISHING || ctx.getRequestState() == CHUNKING_NOT_SUPPORTED) {
- builder.code(ResponseStatusCode.SUCCESS).statusMessage("OK");
- }
+ ResponseMessage responseMessage = null;
- responseMessage = builder.create();
- }
+ // for this state no need to build full ResponseMessage
+ if (ctx.getRequestState() != STREAMING) {
+ final ResponseMessage.Builder builder = ResponseMessage.buildV4(msg.getRequestId()).result(aggregate);
- if (ctx.getRequestState() == CHUNKING_NOT_SUPPORTED) {
- return new Frame(serializer.serializeResponseAsBinary(responseMessage, nettyContext.alloc()));
+ // need to put status in last message
+ if (ctx.getRequestState() == FINISHING || ctx.getRequestState() == CHUNKING_NOT_SUPPORTED) {
+ builder.code(ResponseStatusCode.SUCCESS).statusMessage("OK");
}
- final MessageChunkSerializer<?> chunkSerializer = (MessageChunkSerializer) serializer;
+ responseMessage = builder.create();
+ }
- switch (ctx.getRequestState()) {
- case NOT_STARTED:
- if (code == ResponseStatusCode.SUCCESS) {
- ctx.setRequestState(FINISHED);
+ if (ctx.getRequestState() == CHUNKING_NOT_SUPPORTED) {
+ return serializer.serializeResponseAsBinary(responseMessage, nettyContext.alloc());
+ }
- return new Frame(serializer.serializeResponseAsBinary(ResponseMessage.buildV4(msg.getRequestId())
- .statusAttributes(statusAttributes)
- .responseMetaData(responseMetaData)
- .result(aggregate)
- .code(ResponseStatusCode.SUCCESS)
- .statusMessage("OK")
- .create(), nettyContext.alloc()));
- }
+ final MessageChunkSerializer<?> chunkSerializer = (MessageChunkSerializer) serializer;
+ switch (ctx.getRequestState()) {
+ case NOT_STARTED:
+ if (hasMore) {
ctx.setRequestState(STREAMING);
- return new Frame(chunkSerializer.writeHeader(responseMessage, nettyContext.alloc()));
- case STREAMING:
- return new Frame(chunkSerializer.writeChunk(aggregate, nettyContext.alloc()));
- case FINISHING:
- ctx.setRequestState(FINISHED);
- return new Frame(chunkSerializer.writeFooter(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.
- final MessageTextSerializer<?> textSerializer = (MessageTextSerializer<?>) serializer;
- return new Frame(textSerializer.serializeResponseAsString(ResponseMessage.build(msg)
- .code(code)
- .statusAttributes(statusAttributes)
- .responseMetaData(responseMetaData)
- .result(aggregate).create(), nettyContext.alloc()));
+ return chunkSerializer.writeHeader(responseMessage, nettyContext.alloc());
+ }
+ ctx.setRequestState(FINISHED);
+
+ return serializer.serializeResponseAsBinary(ResponseMessage.buildV4(msg.getRequestId())
+ .result(aggregate)
+ .code(ResponseStatusCode.SUCCESS)
+ .statusMessage("OK")
+ .create(), nettyContext.alloc());
+
+ case STREAMING:
+ return chunkSerializer.writeChunk(aggregate, nettyContext.alloc());
+ case FINISHING:
+ ctx.setRequestState(FINISHED);
+ return chunkSerializer.writeFooter(responseMessage, nettyContext.alloc());
}
+
+ // todo: just throw?
+ return serializer.serializeResponseAsBinary(responseMessage, nettyContext.alloc());
+
} catch (Exception ex) {
logger.warn("The result [{}] in the request {} could not be serialized and returned.", aggregate, msg.getRequestId(), ex);
final String errorMessage = String.format("Error during serialization: %s", ExceptionHelper.getMessageFromExceptionOrCause(ex));
- writeErrorFrame(ctx.getChannelHandlerContext(),
+ writeError(ctx,
ResponseMessage.build(msg.getRequestId())
.statusMessage(errorMessage)
.statusAttributeException(ex)
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 39fb04e9f8..6bd68669ad 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
@@ -42,6 +42,7 @@ import org.apache.tinkerpop.gremlin.util.Tokens;
import org.apache.tinkerpop.gremlin.util.message.RequestMessage;
import org.apache.tinkerpop.gremlin.util.message.RequestMessageV4;
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.MessageTextSerializerV4;
import org.apache.tinkerpop.gremlin.util.ser.SerTokens;
@@ -264,17 +265,9 @@ public class HttpHandlerUtil {
}
}
- static void writeErrorFrame(final ChannelHandlerContext ctx, final ResponseMessage responseMessage, final MessageSerializer<?> serializer) {
- try {
- 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) {
+ static void writeError(final Context context, ResponseMessage responseMessage, final MessageSerializer<?> serializer) {
try {
+ final ChannelHandlerContext ctx = context.getChannelHandlerContext();
final ByteBuf ByteBuf = context.getRequestState() == HttpGremlinEndpointHandler.RequestState.STREAMING
? ((MessageChunkSerializer) serializer).writeErrorFooter(responseMessage, ctx.alloc())
: serializer.serializeResponseAsBinary(responseMessage, ctx.alloc());
@@ -282,15 +275,15 @@ public class HttpHandlerUtil {
context.setRequestState(HttpGremlinEndpointHandler.RequestState.ERROR);
ctx.writeAndFlush(new DefaultHttpContent(ByteBuf));
- sendTrailingHeaders(ctx, responseMessage.getStatus().getCode().getValue(), responseMessage.getStatus().getMessage());
+ sendTrailingHeaders(ctx, responseMessage.getStatus().getCode(), 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) {
+ static void sendTrailingHeaders(final ChannelHandlerContext ctx, final ResponseStatusCode statusCode, final String message) {
final DefaultLastHttpContent defaultLastHttpContent = new DefaultLastHttpContent();
- defaultLastHttpContent.trailingHeaders().add(SerTokens.TOKEN_CODE, statusCode);
+ defaultLastHttpContent.trailingHeaders().add(SerTokens.TOKEN_CODE, statusCode.getValue());
defaultLastHttpContent.trailingHeaders().add(SerTokens.TOKEN_MESSAGE, message);
ctx.writeAndFlush(defaultLastHttpContent);
}
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..dad908779b 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
@@ -22,26 +22,14 @@ import io.netty.channel.ChannelHandlerContext;
import nl.altindag.log.LogCaptor;
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.server.handler.Frame;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
import org.mockito.Mockito;
-import java.util.Arrays;
-import java.util.UUID;
-import java.util.function.BiFunction;
-
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.core.Is.is;
import static org.junit.Assert.assertEquals;
-@RunWith(Parameterized.class)
public class ContextTest {
private static LogCaptor logCaptor;
@@ -61,123 +49,6 @@ public class ContextTest {
logCaptor.clearLogs();
}
- @Parameterized.Parameter(value = 0)
- public BiFunction<Context, ResponseStatusCode, Void> writeInvoker;
-
- private final ChannelHandlerContext ctx = Mockito.mock(ChannelHandlerContext.class);
- private final RequestMessage request = RequestMessage.build("test").create();
- private final Settings settings = new Settings();
- private final Context context = new Context(request, ctx, settings, null, null, null);
-
- @Parameterized.Parameters(name = "{0}")
- public static Iterable<Object[]> data() {
- return Arrays.asList(new Object[][] {
- {
- new BiFunction<Context, ResponseStatusCode, Void>() {
- @Override
- public Void apply(final Context context, final ResponseStatusCode code) {
- context.writeAndFlush(code, "testMessage");
- return null;
- }
-
- @Override
- public String toString() {
- return "writeAndFlush(ResponseStatusCode, Object)";
- }
- }
- }, {
- new BiFunction<Context, ResponseStatusCode, Void>() {
- @Override
- public Void apply(final Context context, final ResponseStatusCode code) {
- context.writeAndFlush(ResponseMessage.build(UUID.randomUUID()).code(code).create());
- return null;
- }
-
- @Override
- public String toString() {
- return "writeAndFlush(ResponseMessage)";
- }
- }
- },
- });
- }
-
- @Test
- public void shouldAllowMultipleNonFinalResponses() {
- writeInvoker.apply(context, ResponseStatusCode.AUTHENTICATE);
- Mockito.verify(ctx, Mockito.times(1)).write(Mockito.any());
- Mockito.verify(ctx, Mockito.times(1)).flush();
-
- writeInvoker.apply(context, ResponseStatusCode.PARTIAL_CONTENT);
- Mockito.verify(ctx, Mockito.times(2)).write(Mockito.any());
- Mockito.verify(ctx, Mockito.times(2)).flush();
-
- writeInvoker.apply(context, ResponseStatusCode.PARTIAL_CONTENT);
- Mockito.verify(ctx, Mockito.times(3)).write(Mockito.any());
- Mockito.verify(ctx, Mockito.times(3)).flush();
- }
-
- @Test
- public void shouldAllowAtMostOneFinalResponse() {
- writeInvoker.apply(context, ResponseStatusCode.AUTHENTICATE);
- Mockito.verify(ctx, Mockito.times(1)).write(Mockito.any());
- Mockito.verify(ctx, Mockito.times(1)).flush();
-
- writeInvoker.apply(context, ResponseStatusCode.SUCCESS);
- Mockito.verify(ctx, Mockito.times(2)).write(Mockito.any());
- Mockito.verify(ctx, Mockito.times(2)).flush();
-
- writeInvoker.apply(context, ResponseStatusCode.SERVER_ERROR_TIMEOUT);
- assertThat(logCaptor.getLogs().stream().anyMatch(m ->
- m.matches(".*" + request.getRequestId() + ".*")), is(true));
- assertThat(logCaptor.getLogs().stream().anyMatch(m ->
- m.matches(".*" + ResponseStatusCode.SERVER_ERROR_TIMEOUT + "$")), is(true));
-
- // ensure there were no other writes to the channel
- Mockito.verify(ctx, Mockito.times(2)).write(Mockito.any());
- Mockito.verify(ctx, Mockito.times(2)).flush();
- }
-
- @Test
- public void shouldNotAllowNonFinalMessagesAfterFinalResponse() {
- writeInvoker.apply(context, ResponseStatusCode.SERVER_ERROR_TIMEOUT);
- Mockito.verify(ctx, Mockito.times(1)).write(Mockito.any());
- Mockito.verify(ctx, Mockito.times(1)).flush();
-
- writeInvoker.apply(context, ResponseStatusCode.PARTIAL_CONTENT);
- assertThat(logCaptor.getLogs().stream().anyMatch(m ->
- m.matches(".*" + request.getRequestId() + ".*")), is(true));
- assertThat(logCaptor.getLogs().stream().anyMatch(m ->
- m.matches(".*" + ResponseStatusCode.PARTIAL_CONTENT + "$")), is(true));
-
- // ensure there were no other writes to the channel
- Mockito.verify(ctx, Mockito.times(1)).write(Mockito.any());
- Mockito.verify(ctx, Mockito.times(1)).flush();
- }
-
- @Test
- public void shouldReleaseIgnoredFrames() {
- writeInvoker.apply(context, ResponseStatusCode.SERVER_ERROR_TIMEOUT);
- Mockito.verify(ctx, Mockito.times(1)).write(Mockito.any());
- Mockito.verify(ctx, Mockito.times(1)).flush();
-
- Frame frame = Mockito.mock(Frame.class);
- context.writeAndFlush(ResponseStatusCode.SUCCESS, frame);
-
- assertThat(logCaptor.getLogs().stream().anyMatch(m ->
- m.matches(".*" + request.getRequestId() + ".*")), is(true));
- assertThat(logCaptor.getLogs().stream().anyMatch(m ->
- m.matches(".*" + ResponseStatusCode.SUCCESS + "$")), is(true));
-
- // ensure there were no other writes to the channel
- Mockito.verify(ctx, Mockito.times(1)).write(Mockito.any());
- Mockito.verify(ctx, Mockito.times(1)).flush();
-
- // ensure the frame was released
- Mockito.verify(frame, Mockito.times(1)).tryRelease();
- Mockito.verify(ctx, Mockito.times(1)).flush();
- }
-
@Test
public void shouldParseParametersFromScriptRequest()
{
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 e7070bf44b..b64d050f4c 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
@@ -1133,9 +1133,9 @@ public class GremlinServerHttpIntegrateTest extends AbstractGremlinServerIntegra
final Header[] footers = getTrailingHeaders(response);
assertEquals(2, footers.length);
assertEquals("code", footers[0].getName());
- assertEquals("206", footers[0].getValue());
+ assertEquals("200", footers[0].getValue());
assertEquals("message", footers[1].getName());
- assertEquals("NO_CONTENT", footers[1].getValue());
+ assertEquals("OK", footers[1].getValue());
}
}