You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by sp...@apache.org on 2021/03/19 00:14:32 UTC
[tinkerpop] 01/01: TINKERPOP-2537 Enable g.tx() for remotes and
enable bytecode based sessions
This is an automated email from the ASF dual-hosted git repository.
spmallette pushed a commit to branch TINKERPOP-2537
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit 10bbe8d06acadac5af4ab1b3dd1cf8bc305e2fa8
Author: Stephen Mallette <st...@amazon.com>
AuthorDate: Thu Mar 18 20:13:15 2021 -0400
TINKERPOP-2537 Enable g.tx() for remotes and enable bytecode based sessions
---
.../tinkerpop/gremlin/jsr223/JavaTranslator.java | 6 +-
.../gremlin/process/remote/RemoteConnection.java | 8 +
.../gremlin/process/traversal/Bytecode.java | 12 +
.../traversal/dsl/graph/GraphTraversalSource.java | 15 +-
.../process/traversal/util/BytecodeHelper.java | 7 +
.../apache/tinkerpop/gremlin/structure/Graph.java | 12 +-
.../tinkerpop/gremlin/structure/Transaction.java | 98 ++++-
.../structure/util/AbstractTransaction.java | 13 +-
.../process/traversal/util/BytecodeHelperTest.java | 10 +-
.../apache/tinkerpop/gremlin/driver/Client.java | 5 +-
.../apache/tinkerpop/gremlin/driver/Tokens.java | 6 +-
.../driver/remote/DriverRemoteConnection.java | 33 +-
.../driver/remote/DriverRemoteTransaction.java | 170 +++++++++
.../gremlin/server/op/AbstractEvalOpProcessor.java | 4 +-
.../tinkerpop/gremlin/server/op/OpLoader.java | 2 +-
.../server/op/session/SessionOpProcessor.java | 417 ++++++++++++++++++++-
.../server/op/standard/StandardOpProcessor.java | 2 +-
.../server/GremlinSessionTxIntegrateTest.java | 241 ++++++++++++
18 files changed, 1022 insertions(+), 39 deletions(-)
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/jsr223/JavaTranslator.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/jsr223/JavaTranslator.java
index bf90367..7adf87f 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/jsr223/JavaTranslator.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/jsr223/JavaTranslator.java
@@ -28,6 +28,7 @@ import org.apache.tinkerpop.gremlin.process.traversal.TraversalSource;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.BulkSet;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.Tree;
import org.apache.tinkerpop.gremlin.process.traversal.strategy.TraversalStrategyProxy;
+import org.apache.tinkerpop.gremlin.process.traversal.util.BytecodeHelper;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import java.lang.reflect.Array;
@@ -35,7 +36,6 @@ import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Parameter;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
@@ -43,7 +43,6 @@ import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.StringJoiner;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -76,6 +75,9 @@ public final class JavaTranslator<S extends TraversalSource, T extends Traversal
@Override
public T translate(final Bytecode bytecode) {
+ if (BytecodeHelper.isGraphOperation(bytecode))
+ throw new IllegalArgumentException("JavaTranslator cannot translate traversal operations");
+
TraversalSource dynamicSource = this.traversalSource;
Traversal.Admin<?, ?> traversal = null;
for (final Bytecode.Instruction instruction : bytecode.getSourceInstructions()) {
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/remote/RemoteConnection.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/remote/RemoteConnection.java
index c0daaa7..639e286 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/remote/RemoteConnection.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/remote/RemoteConnection.java
@@ -22,6 +22,7 @@ import org.apache.commons.configuration2.Configuration;
import org.apache.tinkerpop.gremlin.process.remote.traversal.RemoteTraversal;
import org.apache.tinkerpop.gremlin.process.traversal.Bytecode;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
+import org.apache.tinkerpop.gremlin.structure.Transaction;
import java.lang.reflect.Constructor;
import java.util.Iterator;
@@ -41,6 +42,13 @@ public interface RemoteConnection extends AutoCloseable {
public static final String GREMLIN_REMOTE_CONNECTION_CLASS = GREMLIN_REMOTE + "remoteConnectionClass";
/**
+ * Creates a {@link Transaction} object designed to work with remote semantics.
+ */
+ public default Transaction tx() {
+ throw new UnsupportedOperationException("This implementation does not support remote transactions");
+ }
+
+ /**
* Submits {@link Traversal} {@link Bytecode} to a server and returns a promise of a {@link RemoteTraversal}.
* The {@link RemoteTraversal} is an abstraction over two types of results that can be returned as part of the
* response from the server: the results of the {@link Traversal} itself and the side-effects that it produced.
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/Bytecode.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/Bytecode.java
index 186a87c..810b2af 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/Bytecode.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/Bytecode.java
@@ -28,6 +28,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
@@ -54,6 +55,17 @@ public final class Bytecode implements Cloneable, Serializable {
private List<Instruction> sourceInstructions = new ArrayList<>();
private List<Instruction> stepInstructions = new ArrayList<>();
+ public static final Bytecode TX_COMMIT = new Bytecode("tx", "commit");
+ public static final Bytecode TX_ROLLBACK = new Bytecode("tx", "rollback");
+
+ public static final Set<Bytecode> GRAPH_OPERATIONS = new HashSet<>(Arrays.asList(TX_COMMIT, TX_ROLLBACK));
+
+ public Bytecode() {}
+
+ private Bytecode(final String sourceName, final Object... arguments) {
+ this.sourceInstructions.add(new Instruction(sourceName, flattenArguments(arguments)));
+ }
+
/**
* Add a {@link TraversalSource} instruction to the bytecode.
*
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/dsl/graph/GraphTraversalSource.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/dsl/graph/GraphTraversalSource.java
index b08ec4f..64fff5d 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/dsl/graph/GraphTraversalSource.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/dsl/graph/GraphTraversalSource.java
@@ -402,10 +402,21 @@ public class GraphTraversalSource implements TraversalSource {
}
/**
- * Proxies calls through to the underlying {@link Graph#tx()}.
+ * Proxies calls through to the underlying {@link Graph#tx()} or to the {@link RemoteConnection#tx()}.
*/
public Transaction tx() {
- return this.graph.tx();
+ if (null == this.connection)
+ return this.graph.tx();
+ else {
+ // prevent child transactions and let the current Transaction object be bound to the
+ // TraversalSource that spawned it
+ final Transaction tx = this.connection.tx();
+ if (tx == Transaction.NO_OP && this.connection instanceof Transaction)
+ return (Transaction) this.connection;
+ else
+ return tx;
+ }
+
}
/**
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/util/BytecodeHelper.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/util/BytecodeHelper.java
index 77df45b..4ffaee5 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/util/BytecodeHelper.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/util/BytecodeHelper.java
@@ -67,6 +67,13 @@ public final class BytecodeHelper {
return clone;
}
+ /**
+ * Checks if the bytecode is one of the standard {@link Bytecode#GRAPH_OPERATIONS}.
+ */
+ public static boolean isGraphOperation(final Bytecode bytecode) {
+ return Bytecode.GRAPH_OPERATIONS.contains(bytecode);
+ }
+
public static Optional<String> getLambdaLanguage(final Bytecode bytecode) {
for (final Bytecode.Instruction instruction : bytecode.getInstructions()) {
for (Object object : instruction.getArguments()) {
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/Graph.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/Graph.java
index b165cbd..da1c998 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/Graph.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/Graph.java
@@ -271,13 +271,19 @@ public interface Graph extends AutoCloseable, Host {
public Iterator<Edge> edges(final Object... edgeIds);
/**
- * Configure and control the transactions for those graphs that support this feature. Note that this method does
- * not indicate the creation of a "transaction" object. A {@link Transaction} in the TinkerPop context is a
- * transaction "factory" or "controller" that helps manage transactions owned by the underlying graph database.
+ * Configure and control the transactions for those graphs that support this feature.
*/
public Transaction tx();
/**
+ * Configure and control the transactions for those graphs that support this feature. Graphs that support multiple
+ * transaction models can use this method expose different sorts of {@link Transaction} implementations.
+ */
+ public default <Tx extends Transaction> Tx tx(final Class<Tx> txClass) {
+ throw new UnsupportedOperationException("This Graph does not support multiple transaction types - use tx() instead");
+ }
+
+ /**
* Closing a {@code Graph} is equivalent to "shutdown" and implies that no further operations can be executed on
* the instance. Users should consult the documentation of the underlying graph database implementation for what
* this "shutdown" will mean in general and, if supported, how open transactions are handled. It will typically
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/Transaction.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/Transaction.java
index 64a58c8..a81ed15 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/Transaction.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/Transaction.java
@@ -18,27 +18,17 @@
*/
package org.apache.tinkerpop.gremlin.structure;
+import org.apache.tinkerpop.gremlin.process.traversal.TraversalSource;
+import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
import org.apache.tinkerpop.gremlin.structure.util.AbstractTransaction;
import org.apache.tinkerpop.gremlin.structure.util.TransactionException;
-import java.util.Collections;
-import java.util.Set;
-import java.util.function.BiFunction;
import java.util.function.Consumer;
-import java.util.function.Function;
/**
* A set of methods that allow for control of transactional behavior of a {@link Graph} instance. Providers may
* consider using {@link AbstractTransaction} as a base implementation that provides default features for most of
* these methods.
- * <p/>
- * It is expected that this interface be implemented by providers in a {@link ThreadLocal} fashion. In other words
- * transactions are bound to the current thread, which means that any graph operation executed by the thread occurs
- * in the context of that transaction and that there may only be one thread executing in a single transaction.
- * <p/>
- * It is important to realize that this class is not a "transaction object". It is a class that holds transaction
- * related methods thus hiding them from the {@link Graph} interface. This object is not meant to be passed around
- * as a transactional context.
*
* @author Marko A. Rodriguez (http://markorodriguez.com)
* @author Stephen Mallette (http://stephen.genoprime.com)
@@ -71,8 +61,28 @@ public interface Transaction extends AutoCloseable {
* A threaded transaction is a {@link Graph} instance that has a transaction context that enables multiple
* threads to collaborate on the same transaction. A standard transactional context tied to a {@link Graph}
* that supports transactions will typically bind a transaction to a single thread via {@link ThreadLocal}.
+ *
+ * @deprecated As of release 3.5.0, replaced by {@link Graph#tx(Class)} ()} in which an implementation of
+ * {@code Transaction} should provide its own methods for exposing a "threaded transaction".
*/
- public <G extends Graph> G createThreadedTx();
+ @Deprecated
+ public default <G extends Graph> G createThreadedTx() {
+ throw Transaction.Exceptions.threadedTransactionsNotSupported();
+ }
+
+ /**
+ * Starts a transaction in the context of a {@link GraphTraversalSource} instance. It is up to the
+ * {@link Transaction} implementation to decide what this means and up to users to be aware of that meaning.
+ */
+ public default <T extends TraversalSource> T begin() {
+ return (T) begin(GraphTraversalSource.class);
+ }
+
+ /**
+ * Starts a transaction in the context of a particular {@link TraversalSource} instance. It is up to the
+ * {@link Transaction} implementation to decide what this means and up to users to be aware of that meaning.
+ */
+ public <T extends TraversalSource> T begin(final Class<T> traversalSourceClass);
/**
* Determines if a transaction is currently open.
@@ -218,4 +228,66 @@ public interface Transaction extends AutoCloseable {
}
}
}
+
+ public static final Transaction NO_OP = new Transaction() {
+ @Override
+ public void open() {
+ throw new UnsupportedOperationException("This Transaction implementation is a no-op for all methods");
+ }
+
+ @Override
+ public void commit() {
+ throw new UnsupportedOperationException("This Transaction implementation is a no-op for all methods");
+ }
+
+ @Override
+ public void rollback() {
+ throw new UnsupportedOperationException("This Transaction implementation is a no-op for all methods");
+ }
+
+ @Override
+ public <C extends TraversalSource> C begin(final Class<C> traversalSourceClass) {
+ throw new UnsupportedOperationException("This Transaction implementation is a no-op for all methods");
+ }
+
+ @Override
+ public boolean isOpen() {
+ throw new UnsupportedOperationException("This Transaction implementation is a no-op for all methods");
+ }
+
+ @Override
+ public void readWrite() {
+ throw new UnsupportedOperationException("This Transaction implementation is a no-op for all methods");
+ }
+
+ @Override
+ public void close() {
+ throw new UnsupportedOperationException("This Transaction implementation is a no-op for all methods");
+ }
+
+ @Override
+ public Transaction onReadWrite(final Consumer<Transaction> consumer) {
+ throw new UnsupportedOperationException("This Transaction implementation is a no-op for all methods");
+ }
+
+ @Override
+ public Transaction onClose(final Consumer<Transaction> consumer) {
+ throw new UnsupportedOperationException("This Transaction implementation is a no-op for all methods");
+ }
+
+ @Override
+ public void addTransactionListener(final Consumer<Status> listener) {
+ throw new UnsupportedOperationException("This Transaction implementation is a no-op for all methods");
+ }
+
+ @Override
+ public void removeTransactionListener(final Consumer<Status> listener) {
+ throw new UnsupportedOperationException("This Transaction implementation is a no-op for all methods");
+ }
+
+ @Override
+ public void clearTransactionListeners() {
+ throw new UnsupportedOperationException("This Transaction implementation is a no-op for all methods");
+ }
+ };
}
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/AbstractTransaction.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/AbstractTransaction.java
index bbc2bc6..ece32dc 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/AbstractTransaction.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/AbstractTransaction.java
@@ -18,11 +18,11 @@
*/
package org.apache.tinkerpop.gremlin.structure.util;
+import org.apache.tinkerpop.gremlin.process.traversal.TraversalSource;
import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.Transaction;
import java.util.function.Consumer;
-import java.util.function.Function;
/**
* A simple base class for {@link Transaction} that provides some common functionality and default behavior.
@@ -34,10 +34,10 @@ import java.util.function.Function;
* @author Stephen Mallette (http://stephen.genoprime.com)
*/
public abstract class AbstractTransaction implements Transaction {
- private Graph g;
+ private Graph graph;
- public AbstractTransaction(final Graph g) {
- this.g = g;
+ public AbstractTransaction(final Graph graph) {
+ this.graph = graph;
}
/**
@@ -123,6 +123,11 @@ public abstract class AbstractTransaction implements Transaction {
throw Transaction.Exceptions.threadedTransactionsNotSupported();
}
+ @Override
+ public <T extends TraversalSource> T begin(final Class<T> traversalSourceClass) {
+ return graph.traversal(traversalSourceClass);
+ }
+
/**
* {@inheritDoc}
*/
diff --git a/gremlin-core/src/test/java/org/apache/tinkerpop/gremlin/process/traversal/util/BytecodeHelperTest.java b/gremlin-core/src/test/java/org/apache/tinkerpop/gremlin/process/traversal/util/BytecodeHelperTest.java
index 24a94aa..510b353 100644
--- a/gremlin-core/src/test/java/org/apache/tinkerpop/gremlin/process/traversal/util/BytecodeHelperTest.java
+++ b/gremlin-core/src/test/java/org/apache/tinkerpop/gremlin/process/traversal/util/BytecodeHelperTest.java
@@ -18,15 +18,12 @@
*/
package org.apache.tinkerpop.gremlin.process.traversal.util;
-import org.apache.tinkerpop.gremlin.jsr223.JavaTranslator;
import org.apache.tinkerpop.gremlin.process.traversal.Bindings;
import org.apache.tinkerpop.gremlin.process.traversal.Bytecode;
-import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__;
import org.apache.tinkerpop.gremlin.process.traversal.strategy.decoration.OptionsStrategy;
import org.apache.tinkerpop.gremlin.process.traversal.strategy.verification.ReadOnlyStrategy;
-import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.util.empty.EmptyGraph;
import org.apache.tinkerpop.gremlin.util.function.Lambda;
import org.junit.Test;
@@ -94,4 +91,11 @@ public class BytecodeHelperTest {
bc, i -> Stream.of(i.getArguments()).anyMatch(o -> o instanceof Bytecode.Binding));
assertEquals(0, filteredAfterRemoved.getStepInstructions().size());
}
+
+ @Test
+ public void shouldDetermineOperation() {
+ assertThat(BytecodeHelper.isGraphOperation(Bytecode.TX_COMMIT), is(true));
+ assertThat(BytecodeHelper.isGraphOperation(Bytecode.TX_ROLLBACK), is(true));
+ assertThat(BytecodeHelper.isGraphOperation(g.V().out("knows").asAdmin().getBytecode()), is(false));
+ }
}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Client.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Client.java
index 9d22f13..ac14c85 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Client.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Client.java
@@ -699,7 +699,10 @@ public abstract class Client {
this.manageTransactions = settings.getSession().get().manageTransactions;
}
- String getSessionId() {
+ /**
+ * Returns the session identifier bound to this {@code Client}.
+ */
+ public String getSessionId() {
return sessionId;
}
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Tokens.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Tokens.java
index 68b5ebd..5617127 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Tokens.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Tokens.java
@@ -97,11 +97,11 @@ public final class Tokens {
* Implementations that set this key should consider using one of
* these two recommended value types:
* <ul>
- * <li>A {@link java.util.List} implementation containing
- * references for which {@link String#valueOf(Object)} produces
+ * <li>A {@code List} implementation containing
+ * references for which {@code String#valueOf(Object)} produces
* a meaningful return value. For example, a list of strings.</li>
* <li>Otherwise, any single non-list object for which
- * {@link String#valueOf(Object)} produces a meaningful return value.
+ * {@code String#valueOf(Object)} produces a meaningful return value.
* For example, a string.</li>
* </ul>
*/
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteConnection.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteConnection.java
index b542858..ac4eed6 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteConnection.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteConnection.java
@@ -29,6 +29,7 @@ import org.apache.tinkerpop.gremlin.process.traversal.Bytecode;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
import org.apache.tinkerpop.gremlin.process.traversal.strategy.decoration.OptionsStrategy;
import org.apache.tinkerpop.gremlin.process.traversal.util.BytecodeHelper;
+import org.apache.tinkerpop.gremlin.structure.Transaction;
import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
import java.util.Iterator;
@@ -55,7 +56,7 @@ public class DriverRemoteConnection implements RemoteConnection {
private static final String DEFAULT_TRAVERSAL_SOURCE = "g";
- private final Client client;
+ final Client client;
private final boolean tryCloseCluster;
private final boolean tryCloseClient;
private final String remoteTraversalSourceName;
@@ -113,11 +114,15 @@ public class DriverRemoteConnection implements RemoteConnection {
}
private DriverRemoteConnection(final Client client, final String remoteTraversalSourceName) {
+ this(client, remoteTraversalSourceName, false);
+ }
+
+ private DriverRemoteConnection(final Client client, final String remoteTraversalSourceName, final boolean tryCloseClient) {
this.client = client.alias(remoteTraversalSourceName);
this.remoteTraversalSourceName = remoteTraversalSourceName;
this.tryCloseCluster = false;
- attachElements = false;
- tryCloseClient = false;
+ this.attachElements = false;
+ this.tryCloseClient = tryCloseClient;
}
/**
@@ -227,6 +232,18 @@ public class DriverRemoteConnection implements RemoteConnection {
}
}
+ /**
+ * If the connection is bound to a session, then get the session identifier from it.
+ */
+ Optional<String> getSessionId() {
+ if (client instanceof Client.SessionedClient) {
+ Client.SessionedClient c = (Client.SessionedClient) client;
+ return Optional.of(c.getSessionId());
+ }
+
+ return Optional.empty();
+ }
+
protected static RequestOptions getRequestOptions(final Bytecode bytecode) {
final Iterator<OptionsStrategy> itty = BytecodeHelper.findStrategies(bytecode, OptionsStrategy.class);
final RequestOptions.Builder builder = RequestOptions.build();
@@ -258,6 +275,16 @@ public class DriverRemoteConnection implements RemoteConnection {
}
}
+ /**
+ * Constructs a new {@link DriverRemoteTransaction}.
+ */
+ @Override
+ public Transaction tx() {
+ final DriverRemoteConnection session = new DriverRemoteConnection(
+ client.getCluster().connect(UUID.randomUUID().toString()), remoteTraversalSourceName, true);
+ return new DriverRemoteTransaction(session);
+ }
+
@Override
public String toString() {
return "DriverServerConnection-" + client.getCluster() + " [graph=" + remoteTraversalSourceName + "]";
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTransaction.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTransaction.java
new file mode 100644
index 0000000..e76cff0
--- /dev/null
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTransaction.java
@@ -0,0 +1,170 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.tinkerpop.gremlin.driver.remote;
+
+import org.apache.tinkerpop.gremlin.process.remote.RemoteConnection;
+import org.apache.tinkerpop.gremlin.process.remote.RemoteConnectionException;
+import org.apache.tinkerpop.gremlin.process.remote.traversal.RemoteTraversal;
+import org.apache.tinkerpop.gremlin.process.traversal.Bytecode;
+import org.apache.tinkerpop.gremlin.process.traversal.TraversalSource;
+import org.apache.tinkerpop.gremlin.structure.Transaction;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
+
+/**
+ * A remote {@link Transaction} implementation that is implemented with the Java driver. It is also a proxy for a
+ * {@link RemoteConnection} that is bound to a session.
+ * <p/>
+ * For users, starting a transaction with {@link #begin()} will produce a {@link TraversalSource} that can be used
+ * across multiple threads sending the bytecode based requests to a remote session. It is worth noting that the session
+ * will process these requests in a serial fashion and not in parallel. Calling {@link #commit()} or
+ * {@link #rollback()} will also close the session and no additional traversal can be executed on the
+ * {@link TraversalSource}. A fresh call to {@link #begin()} will be required to open a fresh session to work with.
+ * The default behavior of {@link #close()} is to commit the transaction.
+ */
+public class DriverRemoteTransaction implements Transaction, RemoteConnection {
+
+ private final DriverRemoteConnection sessionBasedConnection;
+
+ protected Consumer<Transaction> closeConsumer = CLOSE_BEHAVIOR.COMMIT;
+
+ public DriverRemoteTransaction(final DriverRemoteConnection sessionBasedConnection) {
+ this.sessionBasedConnection = sessionBasedConnection;
+ }
+
+ @Override
+ public <T extends TraversalSource> T begin(final Class<T> traversalSourceClass) {
+ if (!isOpen())
+ throw new IllegalStateException("Transaction cannot begin as the session is already closed - create a new Transaction");
+
+ try {
+ return traversalSourceClass.getConstructor(RemoteConnection.class).newInstance(this);
+ } catch (final Exception e) {
+ throw new IllegalStateException(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * By virtue of creating a {@code DriverRemoteTransaction}, the transaction is considered open. There is no need
+ * to call this method. Calling it when the transaction is closed will result in exception.
+ */
+ @Override
+ public void open() {
+ // no need to issue a command to open the transaction, the server is already in such a state if the
+ if (!isOpen())
+ throw new IllegalStateException("Transaction cannot be opened as the session is already closed - create a new Transaction");
+ }
+
+ @Override
+ public void commit() {
+ closeRemoteTransaction(Bytecode.TX_COMMIT, "Transaction commit for %s failed");
+ }
+
+ @Override
+ public void rollback() {
+ closeRemoteTransaction(Bytecode.TX_ROLLBACK, "Transaction rollback for %s failed");
+ }
+
+ private void closeRemoteTransaction(final Bytecode closeTxWith, final String failureMsg) {
+ try {
+ // kinda weird but we hasNext() the graph command here to ensure that it runs to completion or
+ // else you don't guarantee that we have the returned NO_CONTENT message in hand before proceeding
+ // which could mean the transaction is still in the process of committing. not sure why iterate()
+ // doesn't quite work in this context.
+ this.sessionBasedConnection.submitAsync(closeTxWith).join().hasNext();
+ this.sessionBasedConnection.close();
+ } catch (Exception ex) {
+ throw new RuntimeException(String.format(failureMsg, sessionBasedConnection.getSessionId()), ex);
+ }
+ }
+
+ @Override
+ public boolean isOpen() {
+ // for tx purposes closing is a good enough check
+ return !sessionBasedConnection.client.isClosing();
+ }
+
+ /**
+ * The default close behavior for this {@link Transaction} implementation is to {@link #commit()}.
+ */
+ @Override
+ public void close() {
+ this.closeConsumer.accept(this);
+ }
+
+ /**
+ * This {@link Transaction} implementation is not auto-managed and therefore this method is not supported.
+ */
+ @Override
+ public void readWrite() {
+ throw new UnsupportedOperationException("Remote transaction behaviors are not auto-managed - they are always manually controlled");
+ }
+
+ /**
+ * This {@link Transaction} implementation is not auto-managed and therefore this method is not supported.
+ */
+ @Override
+ public Transaction onReadWrite(final Consumer<Transaction> consumer) {
+ throw new UnsupportedOperationException("Remote transaction behaviors are not configurable - they are always manually controlled");
+ }
+
+ @Override
+ public Transaction onClose(final Consumer<Transaction> consumer) {
+ this.closeConsumer = consumer;
+ return this;
+ }
+
+ /**
+ * There is no support for remote transaction listeners.
+ */
+ @Override
+ public void addTransactionListener(final Consumer<Status> listener) {
+ throw new UnsupportedOperationException("Remote transactions cannot have listeners attached");
+ }
+
+ /**
+ * There is no support for remote transaction listeners.
+ */
+ @Override
+ public void removeTransactionListener(final Consumer<Status> listener) {
+ throw new UnsupportedOperationException("Remote transactions cannot have listeners attached");
+ }
+
+ /**
+ * There is no support for remote transaction listeners.
+ */
+ @Override
+ public void clearTransactionListeners() {
+ throw new UnsupportedOperationException("Remote transactions cannot have listeners attached");
+ }
+
+ /**
+ * It is not possible to have child transactions, therefore this method always returns {@link Transaction#NO_OP}.
+ */
+ @Override
+ public Transaction tx() {
+ return Transaction.NO_OP;
+ }
+
+ @Override
+ public <E> CompletableFuture<RemoteTraversal<?, E>> submitAsync(final Bytecode bytecode) throws RemoteConnectionException {
+ return sessionBasedConnection.submitAsync(bytecode);
+ }
+}
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractEvalOpProcessor.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractEvalOpProcessor.java
index 49cc947..609d1f2 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractEvalOpProcessor.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/AbstractEvalOpProcessor.java
@@ -135,7 +135,7 @@ public abstract class AbstractEvalOpProcessor extends AbstractOpProcessor {
* handled will be passed to this method to see if the sub-class can service the requested op code.
* @return
*/
- public abstract Optional<ThrowingConsumer<Context>> selectOther(final RequestMessage requestMessage) throws OpProcessorException;
+ public abstract Optional<ThrowingConsumer<Context>> selectOther(final Context ctx) throws OpProcessorException;
@Override
public ThrowingConsumer<Context> select(final Context ctx) throws OpProcessorException {
@@ -151,7 +151,7 @@ public abstract class AbstractEvalOpProcessor extends AbstractOpProcessor {
final String msgInvalid = String.format("Message could not be parsed. Check the format of the request. [%s]", message);
throw new OpProcessorException(msgInvalid, ResponseMessage.build(message).code(ResponseStatusCode.REQUEST_ERROR_MALFORMED_REQUEST).statusMessage(msgInvalid).create());
default:
- op = selectOther(message).orElseThrow(() -> {
+ op = selectOther(ctx).orElseThrow(() -> {
final String msgDefault = String.format("Message with op code [%s] is not recognized.", message.getOp());
return new OpProcessorException(msgDefault, ResponseMessage.build(message).code(ResponseStatusCode.REQUEST_ERROR_MALFORMED_REQUEST).statusMessage(msgDefault).create());
});
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/OpLoader.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/OpLoader.java
index 5b69a64..b165436 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/OpLoader.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/OpLoader.java
@@ -65,7 +65,7 @@ public final class OpLoader {
}
/**
- * Gets an {@link OpProcessor} by its name. If it cannot be found an {@link Optional#EMPTY} is returned.
+ * Gets an {@link OpProcessor} by its name. If it cannot be found an {@link Optional#empty()} is returned.
*/
public static Optional<OpProcessor> getProcessor(final String name) {
return Optional.ofNullable(processors.get(name));
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/SessionOpProcessor.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/SessionOpProcessor.java
index b4d6d4e..19c373f 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/SessionOpProcessor.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/session/SessionOpProcessor.java
@@ -18,30 +18,54 @@
*/
package org.apache.tinkerpop.gremlin.server.op.session;
+import io.netty.channel.ChannelHandlerContext;
+import org.apache.tinkerpop.gremlin.driver.MessageSerializer;
import org.apache.tinkerpop.gremlin.driver.Tokens;
import org.apache.tinkerpop.gremlin.driver.message.RequestMessage;
import org.apache.tinkerpop.gremlin.driver.message.ResponseMessage;
import org.apache.tinkerpop.gremlin.driver.message.ResponseStatusCode;
import org.apache.tinkerpop.gremlin.groovy.jsr223.GroovyCompilerGremlinPlugin;
+import org.apache.tinkerpop.gremlin.jsr223.JavaTranslator;
+import org.apache.tinkerpop.gremlin.process.traversal.Bytecode;
+import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalSource;
+import org.apache.tinkerpop.gremlin.process.traversal.util.BytecodeHelper;
+import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalInterruptedException;
import org.apache.tinkerpop.gremlin.server.Context;
+import org.apache.tinkerpop.gremlin.server.GraphManager;
import org.apache.tinkerpop.gremlin.server.GremlinServer;
import org.apache.tinkerpop.gremlin.server.OpProcessor;
import org.apache.tinkerpop.gremlin.server.Settings;
+import org.apache.tinkerpop.gremlin.server.auth.AuthenticatedUser;
+import org.apache.tinkerpop.gremlin.server.handler.Frame;
import org.apache.tinkerpop.gremlin.server.handler.StateKey;
import org.apache.tinkerpop.gremlin.server.op.AbstractEvalOpProcessor;
import org.apache.tinkerpop.gremlin.server.op.OpProcessorException;
import org.apache.tinkerpop.gremlin.server.util.MetricManager;
+import org.apache.tinkerpop.gremlin.server.util.TraverserIterator;
import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.io.graphson.GraphSONMapper;
+import org.apache.tinkerpop.gremlin.structure.io.graphson.GraphSONVersion;
import org.apache.tinkerpop.gremlin.util.function.ThrowingConsumer;
+import org.apache.tinkerpop.shaded.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.script.Bindings;
+import javax.script.ScriptException;
+import javax.script.SimpleBindings;
+import java.lang.reflect.UndeclaredThrowableException;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.function.Supplier;
@@ -54,6 +78,10 @@ import static com.codahale.metrics.MetricRegistry.name;
* @author Stephen Mallette (http://stephen.genoprime.com)
*/
public class SessionOpProcessor extends AbstractEvalOpProcessor {
+ private static final ObjectMapper mapper = GraphSONMapper.build().version(GraphSONVersion.V2_0).create().createMapper();
+ private static final Logger auditLogger = LoggerFactory.getLogger(GremlinServer.AUDIT_LOGGER_NAME);
+ private static final Bindings EMPTY_BINDINGS = new SimpleBindings();
+
private static final Logger logger = LoggerFactory.getLogger(SessionOpProcessor.class);
public static final String OP_PROCESSOR_NAME = "session";
@@ -127,7 +155,9 @@ public class SessionOpProcessor extends AbstractEvalOpProcessor {
* this may be removed completely. Note that closing the channel kills the session now.
*/
@Override
- public Optional<ThrowingConsumer<Context>> selectOther(final RequestMessage requestMessage) throws OpProcessorException {
+ public Optional<ThrowingConsumer<Context>> selectOther(final Context ctx) throws OpProcessorException {
+ final RequestMessage requestMessage = ctx.getRequestMessage();
+
// deprecated the "close" message at 3.3.11 - left this check for the "close" token so that if older versions
// of the driver connect they won't get an error. basically just writes back a NO_CONTENT
// for the immediate term in 3.5.0 and then for some future version remove support for the message completely
@@ -145,11 +175,47 @@ public class SessionOpProcessor extends AbstractEvalOpProcessor {
.code(ResponseStatusCode.NO_CONTENT)
.create());
});
+ } else if (requestMessage.getOp().equals(Tokens.OPS_BYTECODE)) {
+ validateTraversalSourceAlias(ctx, requestMessage, validateTraversalRequest(requestMessage));
+ return Optional.of(this::iterateBytecodeTraversal);
} else {
return Optional.empty();
}
}
+ private static void validateTraversalSourceAlias(final Context ctx, final RequestMessage message, final Map<String, String> aliases) throws OpProcessorException {
+ final String traversalSourceBindingForAlias = aliases.values().iterator().next();
+ if (!ctx.getGraphManager().getTraversalSourceNames().contains(traversalSourceBindingForAlias)) {
+ final String msg = String.format("The traversal source [%s] for alias [%s] is not configured on the server.", traversalSourceBindingForAlias, Tokens.VAL_TRAVERSAL_SOURCE_ALIAS);
+ throw new OpProcessorException(msg, ResponseMessage.build(message).code(ResponseStatusCode.REQUEST_ERROR_INVALID_REQUEST_ARGUMENTS).statusMessage(msg).create());
+ }
+ }
+
+ private static Map<String, String> validateTraversalRequest(final RequestMessage message) throws OpProcessorException {
+ if (!message.optionalArgs(Tokens.ARGS_GREMLIN).isPresent()) {
+ final String msg = String.format("A message with [%s] op code requires a [%s] argument.", Tokens.OPS_BYTECODE, Tokens.ARGS_GREMLIN);
+ throw new OpProcessorException(msg, ResponseMessage.build(message).code(ResponseStatusCode.REQUEST_ERROR_INVALID_REQUEST_ARGUMENTS).statusMessage(msg).create());
+ }
+
+ return validatedAliases(message).get();
+ }
+
+ private static Optional<Map<String, String>> validatedAliases(final RequestMessage message) throws OpProcessorException {
+ final Optional<Map<String, String>> aliases = message.optionalArgs(Tokens.ARGS_ALIASES);
+ if (!aliases.isPresent()) {
+ final String msg = String.format("A message with [%s] op code requires a [%s] argument.", Tokens.OPS_BYTECODE, Tokens.ARGS_ALIASES);
+ throw new OpProcessorException(msg, ResponseMessage.build(message).code(ResponseStatusCode.REQUEST_ERROR_INVALID_REQUEST_ARGUMENTS).statusMessage(msg).create());
+ }
+
+ if (aliases.get().size() != 1 || !aliases.get().containsKey(Tokens.VAL_TRAVERSAL_SOURCE_ALIAS)) {
+ final String msg = String.format("A message with [%s] op code requires the [%s] argument to be a Map containing one alias assignment named '%s'.",
+ Tokens.OPS_BYTECODE, Tokens.ARGS_ALIASES, Tokens.VAL_TRAVERSAL_SOURCE_ALIAS);
+ throw new OpProcessorException(msg, ResponseMessage.build(message).code(ResponseStatusCode.REQUEST_ERROR_INVALID_REQUEST_ARGUMENTS).statusMessage(msg).create());
+ }
+
+ return aliases;
+ }
+
@Override
public ThrowingConsumer<Context> getEvalOp() {
return this::evalOp;
@@ -268,4 +334,353 @@ public class SessionOpProcessor extends AbstractEvalOpProcessor {
return bindings;
};
}
+
+ private void iterateBytecodeTraversal(final Context context) throws Exception {
+ final RequestMessage msg = context.getRequestMessage();
+ final Settings settings = context.getSettings();
+ logger.debug("Traversal request {} for in thread {}", msg.getRequestId(), Thread.currentThread().getName());
+
+ // right now the TraversalOpProcessor can take a direct GraphSON representation of Bytecode or directly take
+ // deserialized Bytecode object.
+ final Object bytecodeObj = msg.getArgs().get(Tokens.ARGS_GREMLIN);
+ final Bytecode bytecode = bytecodeObj instanceof Bytecode ? (Bytecode) bytecodeObj :
+ mapper.readValue(bytecodeObj.toString(), Bytecode.class);
+
+ // earlier validation in selection of this op method should free us to cast this without worry
+ final Map<String, String> aliases = (Map<String, String>) msg.optionalArgs(Tokens.ARGS_ALIASES).get();
+
+ // timeout override - handle both deprecated and newly named configuration. earlier logic should prevent
+ // both configurations from being submitted at the same time
+ final Map<String, Object> args = msg.getArgs();
+ final long seto = args.containsKey(Tokens.ARGS_EVAL_TIMEOUT) ?
+ ((Number) args.get(Tokens.ARGS_EVAL_TIMEOUT)).longValue() : context.getSettings().getEvaluationTimeout();
+
+ final GraphManager graphManager = context.getGraphManager();
+ final String traversalSourceName = aliases.entrySet().iterator().next().getValue();
+ final TraversalSource g = graphManager.getTraversalSource(traversalSourceName);
+
+ // todo: should session be grabbed here???
+ final Session session = getSession(context, msg);
+
+ // handle bytecode based graph operations like commit/rollback commands
+ if (BytecodeHelper.isGraphOperation(bytecode)) {
+ handleGraphOperation(bytecode, g.getGraph(), context);
+ return;
+ }
+
+ final Traversal.Admin<?, ?> traversal;
+ try {
+ final Optional<String> lambdaLanguage = BytecodeHelper.getLambdaLanguage(bytecode);
+ if (!lambdaLanguage.isPresent())
+ traversal = JavaTranslator.of(g).translate(bytecode);
+ else
+ traversal = session.getGremlinExecutor().eval(bytecode, EMPTY_BINDINGS, lambdaLanguage.get(), traversalSourceName);
+ } catch (ScriptException ex) {
+ logger.error("Traversal contains a lambda that cannot be compiled", ex);
+ throw new OpProcessorException("Traversal contains a lambda that cannot be compiled",
+ ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_EVALUATION)
+ .statusMessage(ex.getMessage())
+ .statusAttributeException(ex).create());
+ } catch (Exception ex) {
+ logger.error("Could not deserialize the Traversal instance", ex);
+ throw new OpProcessorException("Could not deserialize the Traversal instance",
+ ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_SERIALIZATION)
+ .statusMessage(ex.getMessage())
+ .statusAttributeException(ex).create());
+ }
+
+ if (settings.enableAuditLog) {
+ AuthenticatedUser user = context.getChannelHandlerContext().channel().attr(StateKey.AUTHENTICATED_USER).get();
+ if (null == user) { // This is expected when using the AllowAllAuthenticator
+ user = AuthenticatedUser.ANONYMOUS_USER;
+ }
+ String address = context.getChannelHandlerContext().channel().remoteAddress().toString();
+ if (address.startsWith("/") && address.length() > 1) address = address.substring(1);
+ auditLogger.info("User {} with address {} requested: {}", user.getName(), address, bytecode);
+ }
+ if (settings.authentication.enableAuditLog) {
+ String address = context.getChannelHandlerContext().channel().remoteAddress().toString();
+ if (address.startsWith("/") && address.length() > 1) address = address.substring(1);
+ auditLogger.info("User with address {} requested: {}", address, bytecode);
+ }
+
+ // todo: timer matter???
+ // final Timer.Context timerContext = traversalOpTimer.time();
+
+ final FutureTask<Void> evalFuture = new FutureTask<>(() -> {
+ final Graph graph = g.getGraph();
+
+ try {
+ beforeProcessing(graph, context);
+
+ try {
+ // compile the traversal - without it getEndStep() has nothing in it
+ traversal.applyStrategies();
+ handleIterator(context, new TraverserIterator(traversal), graph);
+ } catch (Exception ex) {
+ Throwable t = ex;
+ if (ex instanceof UndeclaredThrowableException)
+ t = t.getCause();
+
+ // if any exception in the chain is TemporaryException then we should respond with the right error
+ // code so that the client knows to retry
+ final Optional<Throwable> possibleTemporaryException = determineIfTemporaryException(ex);
+ if (possibleTemporaryException.isPresent()) {
+ context.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_TEMPORARY)
+ .statusMessage(possibleTemporaryException.get().getMessage())
+ .statusAttributeException(possibleTemporaryException.get()).create());
+ } 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);
+ context.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_TIMEOUT)
+ .statusMessage(errorMessage)
+ .statusAttributeException(ex).create());
+ } else {
+ logger.warn(String.format("Exception processing a Traversal on iteration for request [%s].", msg.getRequestId()), ex);
+ context.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR)
+ .statusMessage(ex.getMessage())
+ .statusAttributeException(ex).create());
+ }
+ onError(graph, context);
+ }
+ } catch (Exception ex) {
+ final Optional<Throwable> possibleTemporaryException = determineIfTemporaryException(ex);
+ if (possibleTemporaryException.isPresent()) {
+ context.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_TEMPORARY)
+ .statusMessage(possibleTemporaryException.get().getMessage())
+ .statusAttributeException(possibleTemporaryException.get()).create());
+ } else {
+ logger.warn(String.format("Exception processing a Traversal on request [%s].", msg.getRequestId()), ex);
+ context.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR)
+ .statusMessage(ex.getMessage())
+ .statusAttributeException(ex).create());
+ }
+ onError(graph, context);
+ } finally {
+ // todo: timer matter???
+ //timerContext.stop();
+ }
+
+ return null;
+ });
+
+ submitToGremlinExecutor(context, seto, session, evalFuture);
+ }
+
+ private static void submitToGremlinExecutor(final Context context, final long seto, final Session session,
+ final FutureTask<Void> evalFuture) {
+ final Future<?> executionFuture = session.getGremlinExecutor().getExecutorService().submit(evalFuture);
+ if (seto > 0) {
+ // Schedule a timeout in the thread pool for future execution
+ context.getScheduledExecutorService().schedule(() -> executionFuture.cancel(true), seto, TimeUnit.MILLISECONDS);
+ }
+ }
+
+ /**
+ * If {@link Bytecode} is detected to contain a "graph operation" then it gets processed by this method.
+ */
+ protected void handleGraphOperation(final Bytecode bytecode, final Graph graph, final Context context) {
+ final RequestMessage msg = context.getRequestMessage();
+ final Session session = getSession(context, msg);
+ if (graph.features().graph().supportsTransactions()) {
+ if (bytecode.equals(Bytecode.TX_COMMIT) || bytecode.equals(Bytecode.TX_ROLLBACK)) {
+ final boolean commit = bytecode.equals(Bytecode.TX_COMMIT);
+ submitToGremlinExecutor(context, 0, session, new FutureTask<>(() -> {
+ try {
+ if (graph.tx().isOpen()) {
+ if (commit)
+ graph.tx().commit();
+ else
+ graph.tx().rollback();
+ }
+
+ // write back a no-op for success
+ final Map<String, Object> attributes = generateStatusAttributes(
+ context.getChannelHandlerContext(), msg,
+ ResponseStatusCode.NO_CONTENT, Collections.emptyIterator(), context.getSettings());
+ context.writeAndFlush(ResponseMessage.build(msg)
+ .code(ResponseStatusCode.NO_CONTENT)
+ .statusAttributes(attributes)
+ .create());
+
+ } catch (Exception ex) {
+ final Optional<Throwable> possibleTemporaryException = determineIfTemporaryException(ex);
+ if (possibleTemporaryException.isPresent()) {
+ context.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_TEMPORARY)
+ .statusMessage(possibleTemporaryException.get().getMessage())
+ .statusAttributeException(possibleTemporaryException.get()).create());
+ } else {
+ logger.warn(String.format("Exception processing a Traversal on request [%s].", msg.getRequestId()), ex);
+ context.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR)
+ .statusMessage(ex.getMessage())
+ .statusAttributeException(ex).create());
+ }
+ onError(graph, context);
+ }
+
+ return null;
+ }));
+ } else {
+ throw new IllegalStateException(String.format(
+ "Bytecode in request is not a recognized graph operation: %s", bytecode.toString()));
+ }
+ }
+ }
+
+ protected void beforeProcessing(final Graph graph, final Context ctx) {
+ final boolean managedTransactionsForRequest = manageTransactions ?
+ true : (Boolean) ctx.getRequestMessage().getArgs().getOrDefault(Tokens.ARGS_MANAGE_TRANSACTION, false);
+ if (managedTransactionsForRequest && graph.features().graph().supportsTransactions() && graph.tx().isOpen()) graph.tx().rollback();
+ }
+
+ protected void onError(final Graph graph, final Context ctx) {
+ final boolean managedTransactionsForRequest = manageTransactions ?
+ true : (Boolean) ctx.getRequestMessage().getArgs().getOrDefault(Tokens.ARGS_MANAGE_TRANSACTION, false);
+ if (managedTransactionsForRequest && graph.features().graph().supportsTransactions() && graph.tx().isOpen()) graph.tx().rollback();
+ }
+
+ protected void onTraversalSuccess(final Graph graph, final Context ctx) {
+ final boolean managedTransactionsForRequest = manageTransactions ?
+ true : (Boolean) ctx.getRequestMessage().getArgs().getOrDefault(Tokens.ARGS_MANAGE_TRANSACTION, false);
+ if (managedTransactionsForRequest && graph.features().graph().supportsTransactions() && graph.tx().isOpen()) graph.tx().commit();
+ }
+
+ protected void handleIterator(final Context context, final Iterator itty, final Graph graph) throws InterruptedException {
+ final ChannelHandlerContext nettyContext = context.getChannelHandlerContext();
+ final RequestMessage msg = context.getRequestMessage();
+ final Settings settings = context.getSettings();
+ final MessageSerializer serializer = nettyContext.channel().attr(StateKey.SERIALIZER).get();
+ final boolean useBinary = nettyContext.channel().attr(StateKey.USE_BINARY).get();
+ boolean warnOnce = false;
+
+ // we have an empty iterator - happens on stuff like: g.V().iterate()
+ if (!itty.hasNext()) {
+ final Map<String, Object> attributes = generateStatusAttributes(nettyContext, msg, ResponseStatusCode.NO_CONTENT, itty, settings);
+
+ // 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
+ onTraversalSuccess(graph, context);
+ context.writeAndFlush(ResponseMessage.build(msg)
+ .code(ResponseStatusCode.NO_CONTENT)
+ .statusAttributes(attributes)
+ .create());
+ return;
+ }
+
+ // the batch size can be overridden by the request
+ final int resultIterationBatchSize = (Integer) msg.optionalArgs(Tokens.ARGS_BATCH_SIZE)
+ .orElse(settings.resultIterationBatchSize);
+ List<Object> aggregate = new ArrayList<>(resultIterationBatchSize);
+
+ // use an external control to manage the loop as opposed to just checking hasNext() in the while. this
+ // prevent situations where auto transactions create a new transaction after calls to commit() withing
+ // the loop on calls to hasNext().
+ boolean hasMore = itty.hasNext();
+
+ while (hasMore) {
+ if (Thread.interrupted()) throw new InterruptedException();
+
+ // check if an implementation needs to force flush the aggregated results before the iteration batch
+ // size is reached.
+ final boolean forceFlush = isForceFlushed(nettyContext, msg, itty);
+
+ // have to check the aggregate size because it is possible that the channel is not writeable (below)
+ // so iterating next() if the message is not written and flushed would bump the aggregate size beyond
+ // the expected resultIterationBatchSize. Total serialization time for the response remains in
+ // effect so if the client is "slow" it may simply timeout.
+ //
+ // there is a need to check hasNext() on the iterator because if the channel is not writeable the
+ // previous pass through the while loop will have next()'d the iterator and if it is "done" then a
+ // NoSuchElementException will raise its head. also need a check to ensure that this iteration doesn't
+ // require a forced flush which can be forced by sub-classes.
+ //
+ // this could be placed inside the isWriteable() portion of the if-then below but it seems better to
+ // allow iteration to continue into a batch if that is possible rather than just doing nothing at all
+ // while waiting for the client to catch up
+ if (aggregate.size() < resultIterationBatchSize && itty.hasNext() && !forceFlush) aggregate.add(itty.next());
+
+ // Don't keep executor busy if client has already given up; there is no way to catch up if the channel is
+ // not active, and hence we should break the loop.
+ if (!nettyContext.channel().isActive()) {
+ onError(graph, context);
+ break;
+ }
+
+ // send back a page of results if batch size is met or if it's the end of the results being iterated.
+ // also check writeability of the channel to prevent OOME for slow clients.
+ //
+ // clients might decide to close the Netty channel to the server with a CloseWebsocketFrame after errors
+ // like CorruptedFrameException. On the server, although the channel gets closed, there might be some
+ // executor threads waiting for watermark to clear which will not clear in these cases since client has
+ // already given up on these requests. This leads to these executors waiting for the client to consume
+ // results till the timeout. checking for isActive() should help prevent that.
+ if (nettyContext.channel().isActive() && nettyContext.channel().isWritable()) {
+ if (forceFlush || 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 = generateResultMetaData(nettyContext, msg, code, itty, settings);
+ final Map<String, Object> statusAttrb = generateStatusAttributes(nettyContext, msg, code, itty, settings);
+ Frame frame = null;
+ try {
+ frame = makeFrame(context, msg, serializer, useBinary, aggregate, code,
+ metadata, statusAttrb);
+ } 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();
+
+ // exception is handled in makeFrame() - serialization error gets written back to driver
+ // at that point
+ onError(graph, context);
+ break;
+ }
+
+ try {
+ // only need to reset the aggregation list if there's more stuff to write
+ if (itty.hasNext())
+ aggregate = new ArrayList<>(resultIterationBatchSize);
+ else {
+ // iteration and serialization are both complete which means this finished successfully. note that
+ // errors internal to script eval or timeout will rollback given GremlinServer's global configurations.
+ // local errors will get rolledback below because the exceptions aren't thrown in those cases to be
+ // caught by the GremlinExecutor for global rollback logic. this only needs to be committed if
+ // there are no more items to iterate and serialization is complete
+ onTraversalSuccess(graph, context);
+
+ // exit the result iteration loop as there are no more results left. using this external control
+ // because of the above commit. some graphs may open a new transaction on the call to
+ // hasNext()
+ hasMore = false;
+ }
+ } 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();
+ throw ex;
+ }
+
+ if (!itty.hasNext()) iterateComplete(nettyContext, msg, itty);
+
+ // 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, frame);
+ }
+ } else {
+ // don't keep triggering this warning over and over again for the same request
+ if (!warnOnce) {
+ logger.warn("Pausing response writing as writeBufferHighWaterMark exceeded on {} - writing will continue once client has caught up", msg);
+ warnOnce = true;
+ }
+
+ // since the client is lagging we can hold here for a period of time for the client to catch up.
+ // this isn't blocking the IO thread - just a worker.
+ TimeUnit.MILLISECONDS.sleep(10);
+ }
+ }
+ }
}
diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/standard/StandardOpProcessor.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/standard/StandardOpProcessor.java
index 746397a..81df30c 100644
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/standard/StandardOpProcessor.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/standard/StandardOpProcessor.java
@@ -84,7 +84,7 @@ public class StandardOpProcessor extends AbstractEvalOpProcessor {
}
@Override
- public Optional<ThrowingConsumer<Context>> selectOther(final RequestMessage requestMessage) throws OpProcessorException {
+ public Optional<ThrowingConsumer<Context>> selectOther(final Context ctx) throws OpProcessorException {
return Optional.empty();
}
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinSessionTxIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinSessionTxIntegrateTest.java
new file mode 100644
index 0000000..47b501c
--- /dev/null
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinSessionTxIntegrateTest.java
@@ -0,0 +1,241 @@
+/*
+ * 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;
+
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.tinkerpop.gremlin.driver.Cluster;
+import org.apache.tinkerpop.gremlin.driver.remote.DriverRemoteConnection;
+import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
+import org.apache.tinkerpop.gremlin.structure.Transaction;
+import org.junit.Test;
+
+import java.io.File;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+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.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+/**
+ * Integration tests for gremlin-driver and bytecode sessions.
+ *
+ * @author Stephen Mallette (http://stephen.genoprime.com)
+ */
+public class GremlinSessionTxIntegrateTest extends AbstractGremlinServerIntegrationTest {
+
+ /**
+ * Configure specific Gremlin Server settings for specific tests.
+ */
+ @Override
+ public Settings overrideSettings(final Settings settings) {
+ final String nameOfTest = name.getMethodName();
+
+ deleteDirectory(new File("/tmp/neo4j"));
+ settings.graphs.put("graph", "conf/neo4j-empty.properties");
+
+ switch (nameOfTest) {
+ case "shouldExecuteBytecodeInSession":
+ break;
+ }
+
+ return settings;
+ }
+
+ @Test
+ public void shouldCommitTxBytecodeInSession() throws Exception {
+ assumeNeo4jIsPresent();
+
+ final Cluster cluster = TestClientFactory.build().create();
+ final GraphTraversalSource g = traversal().withRemote(DriverRemoteConnection.using(cluster));
+
+ final GraphTraversalSource gtx = g.tx().begin();
+ assertThat(gtx.tx().isOpen(), is(true));
+
+ gtx.addV("person").iterate();
+ assertEquals(1, (long) gtx.V().count().next());
+
+ // outside the session we should be at zero
+ assertEquals(0, (long) g.V().count().next());
+
+ gtx.tx().commit();
+ assertThat(gtx.tx().isOpen(), is(false));
+
+ // sessionless connections should still be good - close() should not affect that
+ assertEquals(1, (long) g.V().count().next());
+
+ // but the spawned gtx should be dead
+ try {
+ gtx.addV("software").iterate();
+ fail("Should have failed since we committed the transaction");
+ } catch (Exception ex) {
+ final Throwable root = ExceptionUtils.getRootCause(ex);
+ assertEquals("Client is closed", root.getMessage());
+ }
+
+ cluster.close();
+ }
+
+ @Test
+ public void shouldCommitTxBytecodeInSessionWithExplicitTransactionObject() throws Exception {
+ assumeNeo4jIsPresent();
+
+ final Cluster cluster = TestClientFactory.build().create();
+ final GraphTraversalSource g = traversal().withRemote(DriverRemoteConnection.using(cluster));
+ final Transaction tx = g.tx();
+ assertThat(tx.isOpen(), is(true));
+
+ final GraphTraversalSource gtx = tx.begin();
+ gtx.addV("person").iterate();
+ assertEquals(1, (long) gtx.V().count().next());
+ tx.commit();
+ assertThat(tx.isOpen(), is(false));
+
+ // sessionless connections should still be good - close() should not affect that
+ assertEquals(1, (long) g.V().count().next());
+
+ cluster.close();
+ }
+
+ @Test
+ public void shouldRollbackTxBytecodeInSession() throws Exception {
+ assumeNeo4jIsPresent();
+
+ final Cluster cluster = TestClientFactory.build().create();
+ final GraphTraversalSource g = traversal().withRemote(DriverRemoteConnection.using(cluster));
+
+ final GraphTraversalSource gtx = g.tx().begin();
+ assertThat(gtx.tx().isOpen(), is(true));
+
+ gtx.addV("person").iterate();
+ assertEquals(1, (long) gtx.V().count().next());
+ gtx.tx().rollback();
+ assertThat(gtx.tx().isOpen(), is(false));
+
+ // sessionless connections should still be good - close() should not affect that
+ assertEquals(0, (long) g.V().count().next());
+
+ // but the spawned gtx should be dead
+ try {
+ gtx.addV("software").iterate();
+ fail("Should have failed since we committed the transaction");
+ } catch (Exception ex) {
+ final Throwable root = ExceptionUtils.getRootCause(ex);
+ assertEquals("Client is closed", root.getMessage());
+ }
+
+ cluster.close();
+ }
+
+ @Test
+ public void shouldCommitTxBytecodeInSessionOnCloseOfGtx() throws Exception {
+ assumeNeo4jIsPresent();
+
+ final Cluster cluster = TestClientFactory.build().create();
+ final GraphTraversalSource g = traversal().withRemote(DriverRemoteConnection.using(cluster));
+
+ final GraphTraversalSource gtx = g.tx().begin();
+ assertThat(gtx.tx().isOpen(), is(true));
+
+ gtx.addV("person").iterate();
+ assertEquals(1, (long) gtx.V().count().next());
+ gtx.close();
+ assertThat(gtx.tx().isOpen(), is(false));
+
+ // sessionless connections should still be good - close() should not affect that
+ assertEquals(1, (long) g.V().count().next());
+
+ // but the spawned gtx should be dead
+ try {
+ gtx.addV("software").iterate();
+ fail("Should have failed since we committed the transaction");
+ } catch (Exception ex) {
+ final Throwable root = ExceptionUtils.getRootCause(ex);
+ assertEquals("Client is closed", root.getMessage());
+ }
+
+ cluster.close();
+ }
+
+ @Test
+ public void shouldCommitTxBytecodeInSessionOnCloseTx() throws Exception {
+ assumeNeo4jIsPresent();
+
+ final Cluster cluster = TestClientFactory.build().create();
+ final GraphTraversalSource g = traversal().withRemote(DriverRemoteConnection.using(cluster));
+
+ final GraphTraversalSource gtx = g.tx().begin();
+ assertThat(gtx.tx().isOpen(), is(true));
+
+ gtx.addV("person").iterate();
+ assertEquals(1, (long) gtx.V().count().next());
+ gtx.tx().close();
+ assertThat(gtx.tx().isOpen(), is(false));
+
+ // sessionless connections should still be good - close() should not affect that
+ assertEquals(1, (long) g.V().count().next());
+
+ // but the spawned gtx should be dead
+ try {
+ gtx.addV("software").iterate();
+ fail("Should have failed since we committed the transaction");
+ } catch (Exception ex) {
+ final Throwable root = ExceptionUtils.getRootCause(ex);
+ assertEquals("Client is closed", root.getMessage());
+ }
+
+ cluster.close();
+ }
+
+ @Test
+ public void shouldCommitTxBytecodeInSessionReusingGtxAcrossThreads() throws Exception {
+ assumeNeo4jIsPresent();
+
+ final ExecutorService service = Executors.newFixedThreadPool(2);
+
+ final Cluster cluster = TestClientFactory.build().create();
+ final GraphTraversalSource g = traversal().withRemote(DriverRemoteConnection.using(cluster));
+
+ final GraphTraversalSource gtx = g.tx().begin();
+ assertThat(gtx.tx().isOpen(), is(true));
+
+ final int verticesToAdd = 64;
+ for (int ix = 0; ix < verticesToAdd; ix++) {
+ service.submit(() -> gtx.addV("person").iterate());
+ }
+
+ service.shutdown();
+ service.awaitTermination(90000, TimeUnit.MILLISECONDS);
+
+ // outside the session we should be at zero
+ assertEquals(0, (long) g.V().count().next());
+
+ assertEquals(verticesToAdd, (long) gtx.V().count().next());
+ gtx.tx().commit();
+ assertThat(gtx.tx().isOpen(), is(false));
+
+ // sessionless connections should still be good - close() should not affect that
+ assertEquals(verticesToAdd, (long) g.V().count().next());
+
+ cluster.close();
+ }
+}