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/09 21:03:52 UTC
(tinkerpop) branch valentyn/http-java-driver updated: cleanup and tests
This is an automated email from the ASF dual-hosted git repository.
valentyn pushed a commit to branch valentyn/http-java-driver
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
The following commit(s) were added to refs/heads/valentyn/http-java-driver by this push:
new 6c2098536c cleanup and tests
6c2098536c is described below
commit 6c2098536c52dac0bdb95fff93a72c4f3944c723
Author: Valentyn Kahamlyk <Va...@improving.com>
AuthorDate: Tue Apr 9 14:03:42 2024 -0700
cleanup and tests
---
.../src/main/java/examples/Connections.java | 2 +-
.../tinkerpop/gremlin/driver/Channelizer.java | 135 +-----
.../driver/handler/GremlinResponseHandler.java | 1 +
.../driver/handler/HttpGremlinResponseDecoder.java | 63 ---
.../handler/HttpGremlinResponseStreamDecoder.java | 1 -
.../driver/handler/WebSocketClientHandler.java | 161 -------
.../handler/WebSocketGremlinRequestEncoder.java | 65 ---
.../handler/WebSocketGremlinResponseDecoder.java | 56 ---
.../gremlin/driver/simple/AbstractClient.java | 8 +-
.../gremlin/driver/simple/SimpleHttpClient.java | 15 +-
.../gremlin/driver/simple/WebSocketClient.java | 140 ------
.../gremlin/driver/util/ProfilingApplication.java | 4 +-
.../gremlin/server/GremlinDriverIntegrateTest.java | 6 +-
.../gremlin/server/GremlinServerIntegrateTest.java | 12 +-
.../gremlin/server/HttpDriverIntegrateTest.java | 525 ++++++++++++++++++---
.../util/ser/GraphBinaryMessageSerializerV4.java | 2 +-
16 files changed, 498 insertions(+), 698 deletions(-)
diff --git a/gremlin-driver/src/main/java/examples/Connections.java b/gremlin-driver/src/main/java/examples/Connections.java
index 458d5c8fd5..f018533f0a 100644
--- a/gremlin-driver/src/main/java/examples/Connections.java
+++ b/gremlin-driver/src/main/java/examples/Connections.java
@@ -76,7 +76,7 @@ public class Connections {
// See reference/#gremlin-java-configuration for full list of configurations
private static void withCluster() throws Exception {
Cluster cluster = Cluster.build("localhost").
- channelizer(Channelizer.WebSocketChannelizer.class).
+ channelizer(Channelizer.HttpChannelizer.class).
keepAliveInterval(180000).
maxConnectionPoolSize(8).
path("/gremlin").
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java
index fae760178d..3598086428 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Channelizer.java
@@ -23,34 +23,17 @@ import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
-import io.netty.handler.codec.http.DefaultHttpHeaders;
import io.netty.handler.codec.http.HttpClientCodec;
-import io.netty.handler.codec.http.HttpHeaders;
-import io.netty.handler.codec.http.HttpObjectAggregator;
-import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
-import io.netty.handler.codec.http.websocketx.WebSocketVersion;
-import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketClientCompressionHandler;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslHandler;
-import io.netty.handler.timeout.IdleStateHandler;
-import org.apache.tinkerpop.gremlin.driver.exception.ConnectionException;
import org.apache.tinkerpop.gremlin.driver.handler.GremlinResponseHandler;
import org.apache.tinkerpop.gremlin.driver.handler.HttpGremlinRequestEncoder;
-import org.apache.tinkerpop.gremlin.driver.handler.WebSocketClientHandler;
-import org.apache.tinkerpop.gremlin.driver.handler.WebSocketGremlinRequestEncoder;
-import org.apache.tinkerpop.gremlin.driver.handler.WebSocketGremlinResponseDecoder;
import org.apache.tinkerpop.gremlin.driver.handler.HttpGremlinResponseStreamDecoder;
import org.apache.tinkerpop.gremlin.util.ser.MessageChunkSerializer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-
-import static java.lang.Math.toIntExact;
/**
* Client-side channel initializer interface. It is responsible for constructing the Netty {@code ChannelPipeline}
@@ -69,7 +52,7 @@ public interface Channelizer extends ChannelHandler {
* Called on {@link Connection#closeAsync()} to perform an {@code Channelizer} specific functions. Note that the
* {@link Connection} already calls {@code Channel.close()} so there is no need to call that method here.
* An implementation will typically use this method to send a {@code Channelizer} specific message to the
- * server to notify of shutdown coming from the client side (e.g. a "close" websocket frame).
+ * server to notify of shutdown coming from the client side.
*/
public void close(final Channel channel);
@@ -95,10 +78,13 @@ public interface Channelizer extends ChannelHandler {
protected Cluster cluster;
private ConcurrentMap<UUID, ResultQueue> pending;
- protected static final String PIPELINE_GREMLIN_SASL_HANDLER = "gremlin-sasl-handler";
protected static final String PIPELINE_GREMLIN_HANDLER = "gremlin-handler";
public static final String PIPELINE_SSL_HANDLER = "gremlin-ssl-handler";
+ protected static final String PIPELINE_HTTP_CODEC = "http-codec";
+ protected static final String PIPELINE_HTTP_ENCODER = "gremlin-encoder";
+ protected static final String PIPELINE_HTTP_DECODER = "gremlin-decoder";
+
public boolean supportsSsl() {
return cluster.connectionPoolSettings().enableSsl;
}
@@ -149,104 +135,6 @@ public interface Channelizer extends ChannelHandler {
}
}
- /**
- * WebSocket {@link Channelizer} implementation.
- */
- public final class WebSocketChannelizer extends AbstractChannelizer {
- private static final Logger logger = LoggerFactory.getLogger(WebSocketChannelizer.class);
- private WebSocketClientHandler handler;
-
- private WebSocketGremlinRequestEncoder gremlinRequestEncoder;
- private WebSocketGremlinResponseDecoder gremlinResponseDecoder;
-
- @Override
- public void init(final Connection connection) {
- super.init(connection);
- gremlinRequestEncoder = new WebSocketGremlinRequestEncoder(true, cluster.getSerializer());
- gremlinResponseDecoder = new WebSocketGremlinResponseDecoder(cluster.getSerializer());
- }
-
- /**
- * Sends a {@code CloseWebSocketFrame} to the server for the specified channel.
- */
- @Override
- public void close(final Channel channel) {
- if (channel.isOpen()) {
- if (logger.isDebugEnabled()) {
- logger.debug("Sending CloseWS frame to server from channel={}", channel.id().asShortText());
- }
- channel.writeAndFlush(new CloseWebSocketFrame());
- }
- }
-
- @Override
- public boolean supportsSsl() {
- final String scheme = connection.getUri().getScheme();
- return "wss".equalsIgnoreCase(scheme);
- }
-
- @Override
- public void configure(final ChannelPipeline pipeline) {
- final String scheme = connection.getUri().getScheme();
- if (!"ws".equalsIgnoreCase(scheme) && !"wss".equalsIgnoreCase(scheme))
- throw new IllegalStateException("Unsupported scheme (only ws: or wss: supported): " + scheme);
-
- if (!supportsSsl() && "wss".equalsIgnoreCase(scheme))
- throw new IllegalStateException("To use wss scheme ensure that enableSsl is set to true in configuration");
-
- final int maxContentLength = cluster.connectionPoolSettings().maxContentLength;
- final HttpHeaders httpHeaders = new DefaultHttpHeaders();
- if(connection.getCluster().isUserAgentOnConnectEnabled()) {
- httpHeaders.set(UserAgent.USER_AGENT_HEADER_NAME, UserAgent.USER_AGENT);
- }
- handler = new WebSocketClientHandler(
- new WebSocketClientHandler.InterceptedWebSocketClientHandshaker13(
- connection.getUri(), WebSocketVersion.V13, null, true,
- httpHeaders, maxContentLength, true, false, -1,
- cluster.getRequestInterceptor()), cluster.getConnectionSetupTimeout(), supportsSsl());
-
- final int keepAliveInterval = toIntExact(TimeUnit.SECONDS.convert(
- cluster.connectionPoolSettings().keepAliveInterval, TimeUnit.MILLISECONDS));
-
- pipeline.addLast("http-codec", new HttpClientCodec());
- pipeline.addLast("aggregator", new HttpObjectAggregator(maxContentLength));
- // Add compression extension for WebSocket defined in https://tools.ietf.org/html/rfc7692
- pipeline.addLast(WebSocketClientCompressionHandler.INSTANCE);
- pipeline.addLast("idle-state-Handler", new IdleStateHandler(0, keepAliveInterval, 0));
- pipeline.addLast("ws-handler", handler);
- pipeline.addLast("gremlin-encoder", gremlinRequestEncoder);
- pipeline.addLast("gremlin-decoder", gremlinResponseDecoder);
- }
-
- @Override
- public void connected() {
- try {
- // Block until the handshake is complete either successfully or with an error. The handshake future
- // will complete with a timeout exception after some time so it is guaranteed that this future will
- // complete.
- // If future completed with an exception more than likely, SSL is enabled on the server, but the client
- // forgot to enable it or perhaps the server is not configured for websockets.
- handler.handshakeFuture().sync();
- } catch (Exception ex) {
- String errMsg = "";
- if (ex instanceof TimeoutException) {
- errMsg = "Timed out while waiting to complete the connection setup. Consider increasing the " +
- "WebSocket handshake timeout duration.";
- } else {
- errMsg = "Could not complete connection setup to the server. Ensure that SSL is correctly " +
- "configured at both the client and the server. Ensure that client WebSocket handshake " +
- "protocol matches the server. Ensure that the server is still reachable.";
- }
- throw new ConnectionException(connection.getUri(), errMsg, ex);
- }
- }
-
- @Override
- public String getScheme(boolean sslEnabled) {
- return sslEnabled ? "wss" : "ws";
- }
- }
-
/**
* Sends requests over the HTTP endpoint. Client functionality is governed by the limitations of the HTTP endpoint,
* meaning that sessions are not available and as such {@code tx()} (i.e. transactions) are not available over this
@@ -254,10 +142,7 @@ public interface Channelizer extends ChannelHandler {
*/
public final class HttpChannelizer extends AbstractChannelizer {
- private HttpClientCodec handler;
-
private HttpGremlinRequestEncoder gremlinRequestEncoder;
- // private HttpGremlinResponseDecoder gremlinResponseDecoder;
private HttpGremlinResponseStreamDecoder gremlinResponseDecoder;
@Override
@@ -292,13 +177,11 @@ public interface Channelizer extends ChannelHandler {
if (!supportsSsl() && "https".equalsIgnoreCase(scheme))
throw new IllegalStateException("To use https scheme ensure that enableSsl is set to true in configuration");
- final int maxContentLength = cluster.connectionPoolSettings().maxContentLength;
- handler = new HttpClientCodec();
+ final HttpClientCodec handler = new HttpClientCodec();
- pipeline.addLast("http-codec", handler);
- // pipeline.addLast("aggregator", new HttpObjectAggregator(maxContentLength));
- pipeline.addLast("gremlin-encoder", gremlinRequestEncoder);
- pipeline.addLast("gremlin-decoder", gremlinResponseDecoder);
+ pipeline.addLast(PIPELINE_HTTP_CODEC, handler);
+ pipeline.addLast(PIPELINE_HTTP_ENCODER, gremlinRequestEncoder);
+ pipeline.addLast(PIPELINE_HTTP_DECODER, gremlinResponseDecoder);
}
}
}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/GremlinResponseHandler.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/GremlinResponseHandler.java
index daa9384b1c..3794ff1f47 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/GremlinResponseHandler.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/GremlinResponseHandler.java
@@ -124,6 +124,7 @@ public class GremlinResponseHandler extends SimpleChannelInboundHandler<Response
if (ctx.channel().isActive()) ctx.close();
}
+ // todo: solution is not decided
private Map<String, Object> cleanStatusAttributes(final Map<String, Object> statusAttributes) {
final Map<String, Object> m = new HashMap<>();
statusAttributes.forEach((k, v) -> {
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinResponseDecoder.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinResponseDecoder.java
deleted file mode 100644
index 4d7d452239..0000000000
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinResponseDecoder.java
+++ /dev/null
@@ -1,63 +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.driver.handler;
-
-import io.netty.buffer.ByteBufInputStream;
-import io.netty.channel.ChannelHandler;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.handler.codec.MessageToMessageDecoder;
-import io.netty.handler.codec.http.FullHttpResponse;
-import io.netty.handler.codec.http.HttpResponseStatus;
-import org.apache.tinkerpop.gremlin.util.MessageSerializer;
-import org.apache.tinkerpop.gremlin.util.Tokens;
-import org.apache.tinkerpop.gremlin.util.message.ResponseMessage;
-import org.apache.tinkerpop.gremlin.util.message.ResponseStatusCode;
-import org.apache.tinkerpop.gremlin.util.ser.SerTokens;
-import org.apache.tinkerpop.shaded.jackson.databind.JsonNode;
-import org.apache.tinkerpop.shaded.jackson.databind.ObjectMapper;
-
-import java.util.List;
-import java.util.UUID;
-
-/**
- * Converts {@code HttpResponse} to a {@link ResponseMessage}.
- * Can be used for GraphSON
- */
-@ChannelHandler.Sharable
-public final class HttpGremlinResponseDecoder extends MessageToMessageDecoder<FullHttpResponse> {
- private final MessageSerializer<?> serializer;
- private final ObjectMapper mapper = new ObjectMapper();
-
- public HttpGremlinResponseDecoder(final MessageSerializer<?> serializer) {
- this.serializer = serializer;
- }
-
- @Override
- protected void decode(final ChannelHandlerContext channelHandlerContext, final FullHttpResponse httpResponse, final List<Object> objects) throws Exception {
- if (httpResponse.status() == HttpResponseStatus.OK) {
- objects.add(serializer.deserializeResponse(httpResponse.content()));
- } else {
- final JsonNode root = mapper.readTree(new ByteBufInputStream(httpResponse.content()));
- objects.add(ResponseMessage.build(UUID.fromString(root.get(Tokens.REQUEST_ID).asText()))
- .code(ResponseStatusCode.SERVER_ERROR)
- .statusMessage(root.get(SerTokens.TOKEN_MESSAGE).asText())
- .create());
- }
- }
-}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinResponseStreamDecoder.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinResponseStreamDecoder.java
index 1306f7fff0..6a19cdbc91 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinResponseStreamDecoder.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/HttpGremlinResponseStreamDecoder.java
@@ -43,7 +43,6 @@ public class HttpGremlinResponseStreamDecoder extends MessageToMessageDecoder<De
private final MessageChunkSerializer<?> serializer;
public HttpGremlinResponseStreamDecoder(MessageChunkSerializer<?> serializer) {
-
this.serializer = serializer;
}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/WebSocketClientHandler.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/WebSocketClientHandler.java
deleted file mode 100644
index 55df38f989..0000000000
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/WebSocketClientHandler.java
+++ /dev/null
@@ -1,161 +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.driver.handler;
-
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelPromise;
-import io.netty.handler.codec.http.FullHttpRequest;
-import io.netty.handler.codec.http.HttpHeaders;
-import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
-import io.netty.handler.codec.http.websocketx.WebSocketClientHandshaker13;
-import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolHandler;
-import io.netty.handler.codec.http.websocketx.WebSocketClientHandshaker;
-import io.netty.handler.codec.http.websocketx.WebSocketVersion;
-import io.netty.handler.ssl.SslHandler;
-import io.netty.handler.ssl.SslHandshakeCompletionEvent;
-import io.netty.handler.timeout.IdleState;
-import io.netty.handler.timeout.IdleStateEvent;
-import io.netty.util.concurrent.Promise;
-
-import java.net.URI;
-import java.util.concurrent.TimeoutException;
-import java.util.function.UnaryOperator;
-
-import javax.net.ssl.SSLHandshakeException;
-
-import org.apache.tinkerpop.gremlin.driver.Channelizer;
-import org.apache.tinkerpop.gremlin.driver.Cluster;
-import org.apache.tinkerpop.gremlin.driver.HandshakeInterceptor;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Wrapper over {@link WebSocketClientProtocolHandler}. This wrapper provides a future which represents the termination
- * of a WS handshake (both success and failure).
- */
-public final class WebSocketClientHandler extends WebSocketClientProtocolHandler {
- private static final Logger logger = LoggerFactory.getLogger(WebSocketClientHandler.class);
-
- private final long connectionSetupTimeoutMillis;
- private ChannelPromise handshakeFuture;
- private boolean sslHandshakeCompleted;
- private boolean useSsl;
-
- public WebSocketClientHandler(final WebSocketClientHandshaker handshaker, final long timeoutMillis, final boolean useSsl) {
- super(handshaker, /*handleCloseFrames*/true, /*dropPongFrames*/true, timeoutMillis);
- this.connectionSetupTimeoutMillis = timeoutMillis;
- this.useSsl = useSsl;
- }
-
- public ChannelFuture handshakeFuture() {
- return handshakeFuture;
- }
-
- @Override
- public void handlerAdded(final ChannelHandlerContext ctx) {
- super.handlerAdded(ctx);
- handshakeFuture = ctx.newPromise();
- }
-
- @Override
- public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cause) throws Exception {
- if (!handshakeFuture.isDone()) {
- handshakeFuture.setFailure(cause);
- }
-
- // let the GremlinResponseHandler take care of exception logging, channel closing, and cleanup
- ctx.fireExceptionCaught(cause);
- }
-
- @Override
- public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
- if (!handshakeFuture.isDone()) {
- // channel was closed before the handshake could be completed.
- handshakeFuture.setFailure(
- new RuntimeException(String.format("WebSocket channel=[%s] closed before the handshake could complete." +
- " Server logs could contain the reason for abrupt connection disconnect or the " +
- "server might not be reachable from the client anymore.",
- ctx.channel().id().asShortText())));
- }
-
- super.channelInactive(ctx);
- }
-
- @Override
- public void userEventTriggered(final ChannelHandlerContext ctx, final Object event) throws Exception {
- if (event instanceof IdleStateEvent) {
- IdleStateEvent e = (IdleStateEvent) event;
- if (e.state() == IdleState.READER_IDLE) {
- logger.warn("WebSocket connection {} has been idle for too long.", ctx.channel());
- } else if (e.state() == IdleState.WRITER_IDLE) {
- logger.debug("Sending ping frame to the server");
- ctx.writeAndFlush(new PingWebSocketFrame());
- }
- } else if (ClientHandshakeStateEvent.HANDSHAKE_COMPLETE.equals(event)) {
- if (!handshakeFuture.isDone()) {
- handshakeFuture.setSuccess();
- }
- } else if (ClientHandshakeStateEvent.HANDSHAKE_TIMEOUT.equals(event)) {
- if (!handshakeFuture.isDone()) {
- TimeoutException te = new TimeoutException(
- String.format((useSsl && !sslHandshakeCompleted) ?
- "SSL handshake not completed in stipulated time=[%s]ms" :
- "WebSocket handshake not completed in stipulated time=[%s]ms",
- connectionSetupTimeoutMillis));
- handshakeFuture.setFailure(te);
- logger.error(te.getMessage());
- }
-
- if (useSsl && !sslHandshakeCompleted) {
- SslHandler handler = ((SslHandler) ctx.pipeline().get(Channelizer.AbstractChannelizer.PIPELINE_SSL_HANDLER));
- ((Promise<Channel>) handler.handshakeFuture()).tryFailure(new SSLHandshakeException("SSL handshake timed out."));
- }
- } else if (event instanceof SslHandshakeCompletionEvent) {
- sslHandshakeCompleted = true;
- } else {
- super.userEventTriggered(ctx, event);
- }
- }
-
- /**
- * Extension to the Netty implementation that allows for the {@link #newHandshakeRequest()} to be modified by way
- * of a {@link HandshakeInterceptor} that is supplied to the {@link Cluster} when it is created.
- */
- public static class InterceptedWebSocketClientHandshaker13 extends WebSocketClientHandshaker13 {
-
- private final UnaryOperator<FullHttpRequest> interceptor;
-
- public InterceptedWebSocketClientHandshaker13(final URI webSocketURL, final WebSocketVersion version,
- final String subprotocol, final boolean allowExtensions,
- final HttpHeaders customHeaders, final int maxFramePayloadLength,
- final boolean performMasking, final boolean allowMaskMismatch,
- final long forceCloseTimeoutMillis, final UnaryOperator<FullHttpRequest> interceptor) {
- super(webSocketURL, version, subprotocol, allowExtensions, customHeaders, maxFramePayloadLength,
- performMasking, allowMaskMismatch, forceCloseTimeoutMillis);
- this.interceptor = interceptor;
- }
-
- @Override
- protected FullHttpRequest newHandshakeRequest() {
- return this.interceptor.apply(super.newHandshakeRequest());
- }
- }
-}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/WebSocketGremlinRequestEncoder.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/WebSocketGremlinRequestEncoder.java
deleted file mode 100644
index 05226f7d8e..0000000000
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/WebSocketGremlinRequestEncoder.java
+++ /dev/null
@@ -1,65 +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.driver.handler;
-
-import org.apache.tinkerpop.gremlin.util.MessageSerializer;
-import org.apache.tinkerpop.gremlin.driver.exception.ResponseException;
-import org.apache.tinkerpop.gremlin.util.message.RequestMessage;
-import org.apache.tinkerpop.gremlin.util.message.ResponseStatusCode;
-import org.apache.tinkerpop.gremlin.util.ser.MessageTextSerializer;
-import io.netty.buffer.ByteBuf;
-import io.netty.channel.ChannelHandler;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.handler.codec.MessageToMessageEncoder;
-import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
-import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
-
-import java.util.List;
-
-/**
- * @author Stephen Mallette (http://stephen.genoprime.com)
- */
-@ChannelHandler.Sharable
-public final class WebSocketGremlinRequestEncoder extends MessageToMessageEncoder<RequestMessage> {
- private final boolean binaryEncoding;
-
- private final MessageSerializer<?> serializer;
-
- public WebSocketGremlinRequestEncoder(final boolean binaryEncoding, final MessageSerializer<?> serializer) {
- this.binaryEncoding = binaryEncoding;
- this.serializer = serializer;
- }
-
- @Override
- protected void encode(final ChannelHandlerContext channelHandlerContext, final RequestMessage requestMessage, final List<Object> objects) throws Exception {
- try {
- if (binaryEncoding) {
- final ByteBuf encodedMessage = serializer.serializeRequestAsBinary(requestMessage, channelHandlerContext.alloc());
- objects.add(new BinaryWebSocketFrame(encodedMessage));
- } else {
- final MessageTextSerializer<?> textSerializer = (MessageTextSerializer<?>) serializer;
- objects.add(new TextWebSocketFrame(textSerializer.serializeRequestAsString(requestMessage, channelHandlerContext.alloc())));
- }
- } catch (Exception ex) {
- throw new ResponseException(ResponseStatusCode.REQUEST_ERROR_SERIALIZATION, String.format(
- "An error occurred during serialization of this request [%s] - it could not be sent to the server - Reason: %s",
- requestMessage, ex));
- }
- }
-}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/WebSocketGremlinResponseDecoder.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/WebSocketGremlinResponseDecoder.java
deleted file mode 100644
index 3ef1eed66b..0000000000
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/WebSocketGremlinResponseDecoder.java
+++ /dev/null
@@ -1,56 +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.driver.handler;
-
-import org.apache.tinkerpop.gremlin.util.MessageSerializer;
-import org.apache.tinkerpop.gremlin.util.ser.MessageTextSerializer;
-import io.netty.channel.ChannelHandler;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.handler.codec.MessageToMessageDecoder;
-import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
-import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
-import io.netty.handler.codec.http.websocketx.WebSocketFrame;
-
-import java.util.List;
-
-/**
- * @author Stephen Mallette (http://stephen.genoprime.com)
- */
-@ChannelHandler.Sharable
-public final class WebSocketGremlinResponseDecoder extends MessageToMessageDecoder<WebSocketFrame> {
- private final MessageSerializer<?> serializer;
-
- public WebSocketGremlinResponseDecoder(final MessageSerializer<?> serializer) {
- this.serializer = serializer;
- }
-
- @Override
- protected void decode(final ChannelHandlerContext channelHandlerContext, final WebSocketFrame webSocketFrame, final List<Object> objects) throws Exception {
- if (webSocketFrame instanceof BinaryWebSocketFrame) {
- final BinaryWebSocketFrame tf = (BinaryWebSocketFrame) webSocketFrame;
- objects.add(serializer.deserializeResponse(tf.content()));
- } else if (webSocketFrame instanceof TextWebSocketFrame) {
- final TextWebSocketFrame tf = (TextWebSocketFrame) webSocketFrame;
- final MessageTextSerializer<?> textSerializer = (MessageTextSerializer<?>) serializer;
- objects.add(textSerializer.deserializeResponse(tf.text()));
- } else {
- throw new RuntimeException(String.format("WebSocket channel does not handle this type of message: %s", webSocketFrame.getClass().getName()));
- }
- }
-}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/simple/AbstractClient.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/simple/AbstractClient.java
index 3574cd2c91..b79abed7b1 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/simple/AbstractClient.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/simple/AbstractClient.java
@@ -25,7 +25,6 @@ import io.netty.channel.nio.NioEventLoopGroup;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
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 java.util.ArrayList;
import java.util.List;
@@ -33,6 +32,8 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
+import static org.apache.tinkerpop.gremlin.util.message.ResponseStatusCode.NO_CONTENT;
+
/**
* @author Stephen Mallette (http://stephen.genoprime.com)
*/
@@ -66,13 +67,14 @@ public abstract class AbstractClient implements SimpleClient {
final List<ResponseMessage> results = new ArrayList<>();
final CompletableFuture<List<ResponseMessage>> f = new CompletableFuture<>();
callbackResponseHandler.callback = response -> {
- if (f.isDone())
+ // message with trailers
+ if (f.isDone() && response.getStatus().getCode() != NO_CONTENT)
throw new RuntimeException("A terminating message was already encountered - no more messages should have been received");
results.add(response);
// check if the current message is terminating - if it is then we can mark complete
- if (response.getStatus().getCode().isFinalResponse()) {
+ if (response.getStatus() != null && response.getStatus().getCode().isFinalResponse()) {
f.complete(results);
}
};
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/simple/SimpleHttpClient.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/simple/SimpleHttpClient.java
index 8a37dfacb5..d5eed778bc 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/simple/SimpleHttpClient.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/simple/SimpleHttpClient.java
@@ -46,9 +46,7 @@ import java.net.URI;
import java.util.concurrent.TimeUnit;
/**
- * A simple, non-thread safe Gremlin Server client using websockets. Typical use is for testing and demonstration.
- *
- * @author Stephen Mallette (http://stephen.genoprime.com)
+ * A simple, non-thread safe Gremlin Server client using HTTP. Typical use is for testing and demonstration.
*/
public class SimpleHttpClient extends AbstractClient {
private static final Logger logger = LoggerFactory.getLogger(SimpleHttpClient.class);
@@ -90,8 +88,6 @@ public class SimpleHttpClient extends AbstractClient {
sslCtx = null;
}
-// final WebSocketClientHandler wsHandler = new WebSocketClientHandler(WebSocketClientHandshakerFactory.newHandshaker(
-// uri, WebSocketVersion.V13, null, true, EmptyHttpHeaders.INSTANCE, 65536), 10000, false);
final MessageChunkSerializer<GraphBinaryMapper> serializer = new GraphBinaryMessageSerializerV4();
b.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@@ -104,21 +100,14 @@ public class SimpleHttpClient extends AbstractClient {
p.addLast(
new HttpClientCodec(),
new HttpGremlinResponseStreamDecoder(serializer),
-// new HttpObjectAggregator(65536),
-
-// wsHandler,
-// new WebSocketGremlinRequestEncoder(true, serializer),
-// new WebSocketGremlinResponseDecoder(serializer),
new HttpGremlinRequestEncoder(serializer, HandshakeInterceptor.NO_OP, false),
- // new HttpGremlinResponseDecoder(serializer),
- new HttpGremlinResponseDebugStreamDecoder(),
+ // new HttpGremlinResponseDebugStreamDecoder(),
callbackResponseHandler);
}
});
channel = b.connect(uri.getHost(), uri.getPort()).sync().channel();
-// wsHandler.handshakeFuture().get(30, TimeUnit.SECONDS);
} catch (Exception ex) {
throw new RuntimeException(ex);
}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/simple/WebSocketClient.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/simple/WebSocketClient.java
deleted file mode 100644
index 76cc274efb..0000000000
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/simple/WebSocketClient.java
+++ /dev/null
@@ -1,140 +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.driver.simple;
-
-import io.netty.buffer.PooledByteBufAllocator;
-import io.netty.channel.ChannelOption;
-import io.netty.handler.codec.http.EmptyHttpHeaders;
-import io.netty.handler.ssl.SslContext;
-import io.netty.handler.ssl.SslContextBuilder;
-import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
-import org.apache.tinkerpop.gremlin.util.MessageSerializer;
-import org.apache.tinkerpop.gremlin.driver.handler.WebSocketClientHandler;
-import org.apache.tinkerpop.gremlin.driver.handler.WebSocketGremlinRequestEncoder;
-import org.apache.tinkerpop.gremlin.driver.handler.WebSocketGremlinResponseDecoder;
-import org.apache.tinkerpop.gremlin.util.message.RequestMessage;
-import io.netty.bootstrap.Bootstrap;
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelInitializer;
-import io.netty.channel.ChannelPipeline;
-import io.netty.channel.socket.SocketChannel;
-import io.netty.channel.socket.nio.NioSocketChannel;
-import io.netty.handler.codec.http.HttpClientCodec;
-import io.netty.handler.codec.http.HttpObjectAggregator;
-import io.netty.handler.codec.http.websocketx.WebSocketClientHandshakerFactory;
-import io.netty.handler.codec.http.websocketx.WebSocketVersion;
-import org.apache.tinkerpop.gremlin.util.ser.GraphBinaryMessageSerializerV1;
-import org.apache.tinkerpop.gremlin.structure.io.binary.GraphBinaryMapper;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.net.URI;
-import java.util.concurrent.TimeUnit;
-
-/**
- * A simple, non-thread safe Gremlin Server client using websockets. Typical use is for testing and demonstration.
- *
- * @author Stephen Mallette (http://stephen.genoprime.com)
- */
-//public class WebSocketClient extends AbstractClient {
-// private static final Logger logger = LoggerFactory.getLogger(WebSocketClient.class);
-// private final Channel channel;
-//
-// public WebSocketClient() {
-// this(URI.create("ws://localhost:8182/gremlin"));
-// }
-//
-// public WebSocketClient(final URI uri) {
-// super("ws-client-%d");
-// final Bootstrap b = new Bootstrap().group(group);
-// b.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
-//
-// final String protocol = uri.getScheme();
-// if (!"ws".equalsIgnoreCase(protocol) && !"wss".equalsIgnoreCase(protocol))
-// throw new IllegalArgumentException("Unsupported protocol: " + protocol);
-// final String host = uri.getHost();
-// final int port;
-// if (uri.getPort() == -1) {
-// if ("ws".equalsIgnoreCase(protocol)) {
-// port = 80;
-// } else if ("wss".equalsIgnoreCase(protocol)) {
-// port = 443;
-// } else {
-// port = -1;
-// }
-// } else {
-// port = uri.getPort();
-// }
-//
-// try {
-// final boolean ssl = "wss".equalsIgnoreCase(protocol);
-// final SslContext sslCtx;
-// if (ssl) {
-// sslCtx = SslContextBuilder.forClient()
-// .trustManager(InsecureTrustManagerFactory.INSTANCE).build();
-// } else {
-// sslCtx = null;
-// }
-// final WebSocketClientHandler wsHandler = new WebSocketClientHandler(WebSocketClientHandshakerFactory.newHandshaker(
-// uri, WebSocketVersion.V13, null, true, EmptyHttpHeaders.INSTANCE, 65536), 10000, false);
-// final MessageSerializer<GraphBinaryMapper> serializer = new GraphBinaryMessageSerializerV1();
-// b.channel(NioSocketChannel.class)
-// .handler(new ChannelInitializer<SocketChannel>() {
-// @Override
-// protected void initChannel(final SocketChannel ch) {
-// final ChannelPipeline p = ch.pipeline();
-// if (sslCtx != null) {
-// p.addLast(sslCtx.newHandler(ch.alloc(), host, port));
-// }
-// p.addLast(
-// new HttpClientCodec(),
-// new HttpObjectAggregator(65536),
-// wsHandler,
-// new WebSocketGremlinRequestEncoder(true, serializer),
-// new WebSocketGremlinResponseDecoder(serializer),
-// callbackResponseHandler);
-// }
-// });
-//
-// channel = b.connect(uri.getHost(), uri.getPort()).sync().channel();
-// wsHandler.handshakeFuture().get(30, TimeUnit.SECONDS);
-// } catch (Exception ex) {
-// throw new RuntimeException(ex);
-// }
-// }
-//
-// @Override
-// public void writeAndFlush(final RequestMessage requestMessage) throws Exception {
-// channel.writeAndFlush(requestMessage);
-// }
-//
-// @Override
-// public void close() throws IOException {
-// try {
-// channel.close().get(30, TimeUnit.SECONDS);
-// } catch (Exception ex) {
-// logger.error("Failure closing simple WebSocketClient", ex);
-// } finally {
-// if (!group.shutdownGracefully().awaitUninterruptibly(30, TimeUnit.SECONDS)) {
-// logger.error("Could not cleanly shutdown thread pool on WebSocketClient");
-// }
-// }
-// }
-//}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/util/ProfilingApplication.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/util/ProfilingApplication.java
index 4259e2545f..b66f838fcf 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/util/ProfilingApplication.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/util/ProfilingApplication.java
@@ -157,8 +157,8 @@ public class ProfilingApplication {
final int maxWaitForConnection = Integer.parseInt(options.getOrDefault("maxWaitForConnection", "3000").toString());
final int workerPoolSize = Integer.parseInt(options.getOrDefault("workerPoolSize", "2").toString());
final int tooSlowThreshold = Integer.parseInt(options.getOrDefault("tooSlowThreshold", "125").toString());
- final String channelizer = options.getOrDefault("channelizer", Channelizer.WebSocketChannelizer.class.getName()).toString();
- final String serializer = options.getOrDefault("serializer", Serializers.GRAPHBINARY_V1.name()).toString();
+ final String channelizer = options.getOrDefault("channelizer", Channelizer.HttpChannelizer.class.getName()).toString();
+ final String serializer = options.getOrDefault("serializer", Serializers.GRAPHBINARY_V4.name()).toString();
final boolean exercise = Boolean.parseBoolean(options.getOrDefault("exercise", "false").toString());
final String script = options.getOrDefault("script", "1+1").toString();
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinDriverIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinDriverIntegrateTest.java
index 1426453107..f57d8f20e4 100644
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinDriverIntegrateTest.java
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinDriverIntegrateTest.java
@@ -22,6 +22,7 @@ import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.Logger;
import nl.altindag.log.LogCaptor;
import org.apache.tinkerpop.gremlin.driver.Channelizer;
+import org.apache.tinkerpop.gremlin.driver.handler.GremlinResponseHandler;
import org.apache.tinkerpop.gremlin.server.channel.HttpChannelizer;
import org.apache.tinkerpop.gremlin.util.ExceptionHelper;
import org.apache.tinkerpop.gremlin.TestHelper;
@@ -34,7 +35,6 @@ import org.apache.tinkerpop.gremlin.util.Tokens;
import org.apache.tinkerpop.gremlin.driver.exception.ConnectionException;
import org.apache.tinkerpop.gremlin.driver.exception.NoHostAvailableException;
import org.apache.tinkerpop.gremlin.driver.exception.ResponseException;
-import org.apache.tinkerpop.gremlin.driver.handler.WebSocketClientHandler;
import org.apache.tinkerpop.gremlin.util.message.RequestMessageV4;
import org.apache.tinkerpop.gremlin.util.message.ResponseStatusCode;
import org.apache.tinkerpop.gremlin.driver.remote.DriverRemoteConnection;
@@ -126,7 +126,7 @@ public class GremlinDriverIntegrateTest extends AbstractGremlinServerIntegration
public void setupForEachTest() {
if (name.getMethodName().equals("shouldKeepAliveForWebSockets") ||
name.getMethodName().equals("shouldKeepAliveForWebSocketsWithNoInFlightRequests")) {
- final Logger webSocketClientHandlerLogger = (Logger) LoggerFactory.getLogger(WebSocketClientHandler.class);
+ final Logger webSocketClientHandlerLogger = (Logger) LoggerFactory.getLogger(GremlinResponseHandler.class);
previousLogLevel = webSocketClientHandlerLogger.getLevel();
webSocketClientHandlerLogger.setLevel(Level.DEBUG);
}
@@ -138,7 +138,7 @@ public class GremlinDriverIntegrateTest extends AbstractGremlinServerIntegration
public void afterEachTest() {
if (name.getMethodName().equals("shouldKeepAliveForWebSockets") ||
name.getMethodName().equals("shouldKeepAliveForWebSocketsWithNoInFlightRequests")) {
- final Logger webSocketClientHandlerLogger = (Logger) LoggerFactory.getLogger(WebSocketClientHandler.class);
+ final Logger webSocketClientHandlerLogger = (Logger) LoggerFactory.getLogger(GremlinResponseHandler.class);
webSocketClientHandlerLogger.setLevel(previousLogLevel);
}
}
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java
index 761e3bc4f0..9074cec276 100644
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java
@@ -936,11 +936,13 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration
final RequestMessageV4 request = RequestMessageV4.build("new String[100]").create();
final List<ResponseMessage> responses = client.submit(request);
assertThat(responses.size(), Matchers.greaterThan(1));
- for (Iterator<ResponseMessage> it = responses.iterator(); it.hasNext(); ) {
- final ResponseMessage msg = it.next();
- final ResponseStatusCode expected = it.hasNext() ? ResponseStatusCode.PARTIAL_CONTENT : ResponseStatusCode.SUCCESS;
- assertEquals(expected, msg.getStatus().getCode());
- }
+
+ // first message have no status
+ assertNull(responses.get(0).getStatus());
+ // second one contains last piece of data
+ assertEquals(ResponseStatusCode.SUCCESS, responses.get(1).getStatus().getCode());
+ // last message with no data, but with trailing headers
+ assertEquals(ResponseStatusCode.NO_CONTENT, responses.get(2).getStatus().getCode());
}
}
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/HttpDriverIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/HttpDriverIntegrateTest.java
index 6c036da7e5..23d76e0c40 100644
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/HttpDriverIntegrateTest.java
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/HttpDriverIntegrateTest.java
@@ -18,25 +18,41 @@
*/
package org.apache.tinkerpop.gremlin.server;
-import io.netty.handler.codec.EncoderException;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.tinkerpop.gremlin.driver.Channelizer;
import org.apache.tinkerpop.gremlin.driver.Client;
import org.apache.tinkerpop.gremlin.driver.Cluster;
+import org.apache.tinkerpop.gremlin.driver.RequestOptions;
+import org.apache.tinkerpop.gremlin.driver.Result;
+import org.apache.tinkerpop.gremlin.driver.ResultSet;
+import org.apache.tinkerpop.gremlin.driver.exception.ResponseException;
import org.apache.tinkerpop.gremlin.driver.remote.DriverRemoteConnection;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
import org.apache.tinkerpop.gremlin.server.channel.HttpChannelizer;
import org.apache.tinkerpop.gremlin.structure.Transaction;
+import org.apache.tinkerpop.gremlin.util.ExceptionHelper;
+import org.apache.tinkerpop.gremlin.util.message.ResponseStatusCode;
import org.apache.tinkerpop.gremlin.util.ser.Serializers;
import org.junit.Test;
+import java.awt.*;
+import java.util.HashMap;
+import java.util.Iterator;
import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.tinkerpop.gremlin.process.traversal.AnonymousTraversalSource.traversal;
+import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.core.AllOf.allOf;
-import static org.hamcrest.core.StringContains.containsString;
+import static org.hamcrest.core.IsInstanceOf.instanceOf;
+import static org.hamcrest.core.StringEndsWith.endsWith;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public class HttpDriverIntegrateTest extends AbstractGremlinServerIntegrationTest {
@@ -47,21 +63,21 @@ public class HttpDriverIntegrateTest extends AbstractGremlinServerIntegrationTes
return settings;
}
- @Test
- public void shouldSubmitScriptWithGraphSON() throws Exception {
- final Cluster cluster = TestClientFactory.build()
- .channelizer(Channelizer.HttpChannelizer.class)
- .serializer(Serializers.GRAPHSON_V4)
- .create();
- try {
- final Client client = cluster.connect();
- assertEquals(2, client.submit("1+1").all().get().get(0).getInt());
- } catch (Exception ex) {
- throw ex;
- } finally {
- cluster.close();
- }
- }
+// @Test
+// public void shouldSubmitScriptWithGraphSON() throws Exception {
+// final Cluster cluster = TestClientFactory.build()
+// .channelizer(Channelizer.HttpChannelizer.class)
+// .serializer(Serializers.GRAPHSON_V4)
+// .create();
+// try {
+// final Client client = cluster.connect();
+// assertEquals(2, client.submit("1+1").all().get().get(0).getInt());
+// } catch (Exception ex) {
+// throw ex;
+// } finally {
+// cluster.close();
+// }
+// }
@Test
public void shouldSubmitScriptWithGraphBinary() throws Exception {
@@ -80,40 +96,40 @@ public class HttpDriverIntegrateTest extends AbstractGremlinServerIntegrationTes
}
}
- @Test
- public void shouldSubmitBytecodeWithGraphSON() throws Exception {
- final Cluster cluster = TestClientFactory.build()
- .channelizer(Channelizer.HttpChannelizer.class)
- .serializer(Serializers.GRAPHSON_V4)
- .create();
- try {
- final GraphTraversalSource g = traversal().withRemote(DriverRemoteConnection.using(cluster));
- final String result = g.inject("2").toList().get(0);
- assertEquals("2", result);
- } catch (Exception ex) {
- throw ex;
- } finally {
- cluster.close();
- }
- }
+// @Test
+// public void shouldSubmitBytecodeWithGraphSON() throws Exception {
+// final Cluster cluster = TestClientFactory.build()
+// .channelizer(Channelizer.HttpChannelizer.class)
+// .serializer(Serializers.GRAPHSON_V4)
+// .create();
+// try {
+// final GraphTraversalSource g = traversal().withRemote(DriverRemoteConnection.using(cluster));
+// final String result = g.inject("2").toList().get(0);
+// assertEquals("2", result);
+// } catch (Exception ex) {
+// throw ex;
+// } finally {
+// cluster.close();
+// }
+// }
- @Test
- public void shouldGetErrorForBytecodeWithUntypedGraphSON() throws Exception {
- final Cluster cluster = TestClientFactory.build()
- .channelizer(Channelizer.HttpChannelizer.class)
- .serializer(Serializers.GRAPHSON_V4_UNTYPED)
- .create();
- try {
- final GraphTraversalSource g = traversal().withRemote(DriverRemoteConnection.using(cluster));
- g.inject("2").toList();
- fail("Exception expected");
- } catch (EncoderException ex) {
- assertThat(ex.getMessage(), allOf(containsString("An error occurred during serialization of this request"),
- containsString("it could not be sent to the server - Reason: only GraphSON3 and GraphBinary recommended for serialization of Bytecode requests, but used org.apache.tinkerpop.gremlin.")));
- } finally {
- cluster.close();
- }
- }
+// @Test
+// public void shouldGetErrorForBytecodeWithUntypedGraphSON() throws Exception {
+// final Cluster cluster = TestClientFactory.build()
+// .channelizer(Channelizer.HttpChannelizer.class)
+// .serializer(Serializers.GRAPHSON_V4_UNTYPED)
+// .create();
+// try {
+// final GraphTraversalSource g = traversal().withRemote(DriverRemoteConnection.using(cluster));
+// g.inject("2").toList();
+// fail("Exception expected");
+// } catch (EncoderException ex) {
+// assertThat(ex.getMessage(), allOf(containsString("An error occurred during serialization of this request"),
+// containsString("it could not be sent to the server - Reason: only GraphSON3 and GraphBinary recommended for serialization of Bytecode requests, but used org.apache.tinkerpop.gremlin.")));
+// } finally {
+// cluster.close();
+// }
+// }
@Test
public void shouldSubmitBytecodeWithGraphBinary() throws Exception {
@@ -188,19 +204,412 @@ public class HttpDriverIntegrateTest extends AbstractGremlinServerIntegrationTes
}
}
- // !!!
+// @Test
+// public void shouldDeserializeErrorWithGraphSON() throws Exception {
+// final Cluster cluster = TestClientFactory.build()
+// .channelizer(Channelizer.HttpChannelizer.class)
+// .serializer(Serializers.GRAPHSON_V4)
+// .create();
+// try {
+// final GraphTraversalSource g = traversal().withRemote(DriverRemoteConnection.using(cluster, "doesNotExist"));
+// g.V().next();
+// fail("Expected exception to be thrown.");
+// } catch (Exception ex) {
+// assert ex.getMessage().contains("Could not rebind");
+// } finally {
+// cluster.close();
+// }
+// }
+
@Test
- public void shouldDeserializeErrorWithGraphSON() throws Exception {
+ public void shouldReportErrorWhenRequestCantBeSerialized() throws Exception {
final Cluster cluster = TestClientFactory.build()
.channelizer(Channelizer.HttpChannelizer.class)
- .serializer(Serializers.GRAPHSON_V4)
+ .serializer(Serializers.GRAPHBINARY_V4)
.create();
try {
- final GraphTraversalSource g = traversal().withRemote(DriverRemoteConnection.using(cluster, "doesNotExist"));
- g.V().next();
- fail("Expected exception to be thrown.");
+ final Client client = cluster.connect();
+
+ try {
+ final Map<String, Object> params = new HashMap<>();
+ params.put("r", Color.RED);
+ client.submit("r", params).all().get();
+ fail("Should have thrown exception over bad serialization");
+ } catch (Exception ex) {
+ final Throwable inner = ExceptionHelper.getRootCause(ex);
+ assertThat(inner, instanceOf(ResponseException.class));
+ assertEquals(ResponseStatusCode.REQUEST_ERROR_SERIALIZATION, ((ResponseException) inner).getResponseStatusCode());
+ assertTrue(ex.getMessage().contains("An error occurred during serialization of this request"));
+ }
+
+ // should not die completely just because we had a bad serialization error. that kind of stuff happens
+ // from time to time, especially in the console if you're just exploring.
+ assertEquals(2, client.submit("1+1").all().get().get(0).getInt());
+ } finally {
+ cluster.close();
+ }
+ }
+
+ // not supported on server side now
+ @Test
+ public void shouldProcessTraversalInterruption() throws Exception {
+ final Cluster cluster = TestClientFactory.build()
+ .channelizer(Channelizer.HttpChannelizer.class)
+ .serializer(Serializers.GRAPHBINARY_V4)
+ .create();
+
+ try {
+ final Client client = cluster.connect();
+ client.submit("g.inject(1).sideEffect{Thread.sleep(5000)}").all().get();
+ fail("Should have timed out");
+ } catch (Exception ex) {
+ final ResponseException re = (ResponseException) ex.getCause();
+ assertEquals(ResponseStatusCode.SERVER_ERROR_TIMEOUT, re.getResponseStatusCode());
+ } finally {
+ cluster.close();
+ }
+ }
+
+ // not supported on server side now
+ @Test
+ public void shouldProcessEvalInterruption() throws Exception {
+ final Cluster cluster = TestClientFactory.build()
+ .channelizer(Channelizer.HttpChannelizer.class)
+ .serializer(Serializers.GRAPHBINARY_V4)
+ .create();
+
+ try {
+ final Client client = cluster.connect();
+ client.submit("Thread.sleep(5000);'done'").all().get();
+ fail("Should have timed out");
} catch (Exception ex) {
- assert ex.getMessage().contains("Could not rebind");
+ final ResponseException re = (ResponseException) ex.getCause();
+ assertEquals(ResponseStatusCode.SERVER_ERROR_TIMEOUT, re.getResponseStatusCode());
+ } finally {
+ cluster.close();
+ }
+ }
+
+ @Test
+ public void shouldFailWithBadClientSideSerialization() throws Exception {
+ final Cluster cluster = TestClientFactory.build()
+ .channelizer(Channelizer.HttpChannelizer.class)
+ .serializer(Serializers.GRAPHBINARY_V4)
+ .create();
+ final Client client = cluster.connect();
+
+ try {
+
+ final ResultSet results = client.submit("java.awt.Color.RED");
+
+ try {
+ results.all().join();
+ fail("Should have thrown exception over bad serialization");
+ } catch (Exception ex) {
+ final Throwable inner = ExceptionHelper.getRootCause(ex);
+ assertThat(inner, instanceOf(ResponseException.class));
+ assertEquals(ResponseStatusCode.SERVER_ERROR_SERIALIZATION, ((ResponseException) inner).getResponseStatusCode());
+ }
+
+ // should not die completely just because we had a bad serialization error. that kind of stuff happens
+ // from time to time, especially in the console if you're just exploring.
+ assertEquals(2, client.submit("1+1").all().get().get(0).getInt());
+ } finally {
+ cluster.close();
+ }
+ }
+
+ @Test
+ public void shouldFailWithScriptExecutionException() throws Exception {
+ final Cluster cluster = TestClientFactory.build()
+ .channelizer(Channelizer.HttpChannelizer.class)
+ .serializer(Serializers.GRAPHBINARY_V4)
+ .create();
+ final Client client = cluster.connect();
+
+ try {
+ try {
+ final ResultSet results = client.submit("1/0");
+ results.all().join();
+ fail("Should have thrown exception over division by zero");
+ } catch (Exception ex) {
+ final Throwable inner = ExceptionHelper.getRootCause(ex);
+ assertTrue(inner instanceof ResponseException);
+ assertThat(inner.getMessage(), endsWith("Division by zero"));
+
+ final ResponseException rex = (ResponseException) inner;
+ // todo: not implemented
+// assertEquals("java.lang.ArithmeticException", rex.getRemoteExceptionHierarchy().get().get(0));
+// assertEquals(1, rex.getRemoteExceptionHierarchy().get().size());
+// assertThat(rex.getRemoteStackTrace().get(), containsString("Division by zero"));
+ }
+
+ // should not die completely just because we had a bad serialization error. that kind of stuff happens
+ // from time to time, especially in the console if you're just exploring.
+ assertEquals(2, client.submit("1+1").all().get().get(0).getInt());
+ } finally {
+ cluster.close();
+ }
+ }
+
+ @Test
+ public void shouldProcessRequestsOutOfOrder() throws Exception {
+ final Cluster cluster = TestClientFactory.build()
+ .channelizer(Channelizer.HttpChannelizer.class)
+ .serializer(Serializers.GRAPHBINARY_V4)
+ .create();
+ try {
+ final Client client = cluster.connect();
+
+ final ResultSet rsFive = client.submit("Thread.sleep(5000);'five'");
+ final ResultSet rsZero = client.submit("'zero'");
+
+ final CompletableFuture<List<Result>> futureFive = rsFive.all();
+ final CompletableFuture<List<Result>> futureZero = rsZero.all();
+
+ assertFalse(futureFive.isDone());
+ assertEquals("zero", futureZero.get().get(0).getString());
+
+ assertFalse(futureFive.isDone());
+ assertEquals("five", futureFive.get(10, TimeUnit.SECONDS).get(0).getString());
+ } finally {
+ cluster.close();
+ }
+ }
+
+ @Test
+ public void shouldWaitForAllResultsToArrive() throws Exception {
+ final Cluster cluster = TestClientFactory.build()
+ .channelizer(Channelizer.HttpChannelizer.class)
+ .serializer(Serializers.GRAPHBINARY_V4)
+ .create();
+ try {
+ final Client client = cluster.connect();
+
+ final AtomicInteger checked = new AtomicInteger(0);
+ final ResultSet results = client.submit("[1,2,3,4,5,6,7,8,9]");
+ while (!results.allItemsAvailable()) {
+ assertTrue(results.getAvailableItemCount() < 10);
+ checked.incrementAndGet();
+ Thread.sleep(100);
+ }
+
+ assertTrue(checked.get() > 0);
+ assertEquals(9, results.getAvailableItemCount());
+ } finally {
+ cluster.close();
+ }
+ }
+
+ @Test
+ public void shouldStream() throws Exception {
+ final Cluster cluster = TestClientFactory.build()
+ .channelizer(Channelizer.HttpChannelizer.class)
+ .serializer(Serializers.GRAPHBINARY_V4)
+ .create();
+ final Client client = cluster.connect();
+
+ try {
+ final ResultSet results = client.submit("[1,2,3,4,5,6,7,8,9]");
+ final AtomicInteger counter = new AtomicInteger(0);
+ results.stream().map(i -> i.get(Integer.class) * 2).forEach(i -> assertEquals(counter.incrementAndGet() * 2, Integer.parseInt(i.toString())));
+ assertEquals(9, counter.get());
+ assertThat(results.allItemsAvailable(), is(true));
+
+ // cant stream it again
+ assertThat(results.stream().iterator().hasNext(), is(false));
+ } finally {
+ cluster.close();
+ }
+ }
+
+ @Test
+ public void shouldIterate() throws Exception {
+ final Cluster cluster = TestClientFactory.build()
+ .channelizer(Channelizer.HttpChannelizer.class)
+ .serializer(Serializers.GRAPHBINARY_V4)
+ .create();
+ final Client client = cluster.connect();
+
+ try {
+ final ResultSet results = client.submit("[1,2,3,4,5,6,7,8,9]");
+ final Iterator<Result> itty = results.iterator();
+ final AtomicInteger counter = new AtomicInteger(0);
+ while (itty.hasNext()) {
+ counter.incrementAndGet();
+ assertEquals(counter.get(), itty.next().getInt());
+ }
+
+ assertEquals(9, counter.get());
+ assertThat(results.allItemsAvailable(), is(true));
+
+ // can't stream it again
+ assertThat(results.iterator().hasNext(), is(false));
+ } finally {
+ cluster.close();
+ }
+ }
+
+ @Test
+ public void shouldGetSomeThenSomeMore() throws Exception {
+ final Cluster cluster = TestClientFactory.build()
+ .channelizer(Channelizer.HttpChannelizer.class)
+ .serializer(Serializers.GRAPHBINARY_V4)
+ .create();
+ final Client client = cluster.connect();
+
+ try {
+ final ResultSet results = client.submit("[1,2,3,4,5,6,7,8,9]");
+ final CompletableFuture<List<Result>> batch1 = results.some(5);
+ final CompletableFuture<List<Result>> batch2 = results.some(5);
+ final CompletableFuture<List<Result>> batchNothingLeft = results.some(5);
+
+ assertEquals(5, batch1.get().size());
+ assertEquals(1, batch1.get().get(0).getInt());
+ assertEquals(2, batch1.get().get(1).getInt());
+ assertEquals(3, batch1.get().get(2).getInt());
+ assertEquals(4, batch1.get().get(3).getInt());
+ assertEquals(5, batch1.get().get(4).getInt());
+
+ assertEquals(4, batch2.get().size());
+ assertEquals(6, batch2.get().get(0).getInt());
+ assertEquals(7, batch2.get().get(1).getInt());
+ assertEquals(8, batch2.get().get(2).getInt());
+ assertEquals(9, batch2.get().get(3).getInt());
+
+ assertEquals(0, batchNothingLeft.get().size());
+ } finally {
+ cluster.close();
+ }
+ }
+
+ @Test
+ public void shouldGetOneThenSomeThenSomeMore() throws Exception {
+ final Cluster cluster = TestClientFactory.build()
+ .channelizer(Channelizer.HttpChannelizer.class)
+ .serializer(Serializers.GRAPHBINARY_V4)
+ .create();
+ final Client client = cluster.connect();
+
+ try {
+ final ResultSet results = client.submit("[1,2,3,4,5,6,7,8,9]");
+ final Result one = results.one();
+ final CompletableFuture<List<Result>> batch1 = results.some(4);
+ final CompletableFuture<List<Result>> batch2 = results.some(5);
+ final CompletableFuture<List<Result>> batchNothingLeft = results.some(5);
+
+ assertEquals(1, one.getInt());
+
+ assertEquals(4, batch1.get().size());
+ assertEquals(2, batch1.get().get(0).getInt());
+ assertEquals(3, batch1.get().get(1).getInt());
+ assertEquals(4, batch1.get().get(2).getInt());
+ assertEquals(5, batch1.get().get(3).getInt());
+
+ assertEquals(4, batch2.get().size());
+ assertEquals(6, batch2.get().get(0).getInt());
+ assertEquals(7, batch2.get().get(1).getInt());
+ assertEquals(8, batch2.get().get(2).getInt());
+ assertEquals(9, batch2.get().get(3).getInt());
+
+ assertEquals(0, batchNothingLeft.get().size());
+ } finally {
+ cluster.close();
+ }
+ }
+
+ @Test
+ public void shouldCloseWithServerDown() throws Exception {
+ final Cluster cluster = TestClientFactory.build()
+ .channelizer(Channelizer.HttpChannelizer.class)
+ .serializer(Serializers.GRAPHBINARY_V4)
+ .create();
+ try {
+ cluster.connect().init();
+
+ stopServer();
+ } finally {
+ cluster.close();
+ }
+ }
+
+ @Test
+ public void shouldFailWithBadServerSideSerialization() throws Exception {
+ final Cluster cluster = TestClientFactory.build()
+ .channelizer(Channelizer.HttpChannelizer.class)
+ .serializer(Serializers.GRAPHBINARY_V4)
+ .create();
+ final Client client = cluster.connect();
+ try {
+
+ final ResultSet results = client.submit("TinkerGraph.open().variables()");
+
+ try {
+ results.all().join();
+ fail();
+ } catch (Exception ex) {
+ final Throwable inner = ExceptionHelper.getRootCause(ex);
+ assertTrue(inner instanceof ResponseException);
+ assertEquals(ResponseStatusCode.SERVER_ERROR_SERIALIZATION, ((ResponseException) inner).getResponseStatusCode());
+ }
+
+ // should not die completely just because we had a bad serialization error. that kind of stuff happens
+ // from time to time, especially in the console if you're just exploring.
+ assertEquals(2, client.submit("1+1").all().get().get(0).getInt());
+ } finally {
+ cluster.close();
+ }
+ }
+
+ @Test
+ public void shouldNotThrowNoSuchElementException() {
+ final Cluster cluster = TestClientFactory.build()
+ .channelizer(Channelizer.HttpChannelizer.class)
+ .serializer(Serializers.GRAPHBINARY_V4)
+ .create();
+ final Client client = cluster.connect();
+
+ try {
+ // this should return "nothing" - there should be no exception
+ assertNull(client.submit("g.V().has('name','kadfjaldjfla')").one());
+ } finally {
+ cluster.close();
+ }
+ }
+
+ @Test
+ public void shouldEvalInGremlinLang() {
+ final Cluster cluster = TestClientFactory.build()
+ .channelizer(Channelizer.HttpChannelizer.class)
+ .serializer(Serializers.GRAPHBINARY_V4)
+ .create();
+ final Client client = cluster.connect();
+
+ try {
+ final RequestOptions ro = RequestOptions.build().language("gremlin-lang").create();
+ assertEquals(111, client.submit("g.inject(111)", ro).one().getInt());
+ } finally {
+ cluster.close();
+ }
+ }
+
+ @Test
+ public void shouldEvalInGremlinLangWithParameters() throws Exception {
+ final Cluster cluster = TestClientFactory.build()
+ .channelizer(Channelizer.HttpChannelizer.class)
+ .serializer(Serializers.GRAPHBINARY_V4)
+ .create();
+ final Client client = cluster.connect();
+
+ try {
+ final RequestOptions ro = RequestOptions.build().language("gremlin-lang").
+ addParameter("x", 100).
+ addParameter("y", "test").
+ addParameter("z", true).create();
+ final List<Result> l = client.submit("g.inject(x, y, z)", ro).all().get();
+ assertEquals(100, l.get(0).getInt());
+ assertEquals("test", l.get(1).getString());
+ assertTrue(l.get(2).getBoolean());
} finally {
cluster.close();
}
diff --git a/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/GraphBinaryMessageSerializerV4.java b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/GraphBinaryMessageSerializerV4.java
index ac268b4a32..46bde7e353 100644
--- a/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/GraphBinaryMessageSerializerV4.java
+++ b/gremlin-util/src/main/java/org/apache/tinkerpop/gremlin/util/ser/GraphBinaryMessageSerializerV4.java
@@ -174,7 +174,7 @@ public class GraphBinaryMessageSerializerV4 extends AbstractGraphBinaryMessageSe
final List<Object> result = new ArrayList<>();
while (buffer.readableBytes() != 0) {
final Object obj = reader.read(buffer);
- if (obj.equals(Marker.END_OF_STREAM)) {
+ if (Marker.END_OF_STREAM.equals(obj)) {
break;
}
result.add(obj);