TINKERPOP-1490 Implemented promise API for Traversal Added two promise() methods that return CompletableFuture on Traversal. Provided an override on DefaultTraversal for those methods because the function that transforms the Traversal is executed in a different thread and therefore requires Graph transaction management (or else we would orphan transactions). Did not update gremlin-python with the promise API because it seemed to beg discussion on the "right" way to do that (i.e. what library to use to support promises?, just use futures from the core lib?, etc).
Project: http://git-wip-us.apache.org/repos/asf/tinkerpop/repo Commit: http://git-wip-us.apache.org/repos/asf/tinkerpop/commit/6c4cbc84 Tree: http://git-wip-us.apache.org/repos/asf/tinkerpop/tree/6c4cbc84 Diff: http://git-wip-us.apache.org/repos/asf/tinkerpop/diff/6c4cbc84 Branch: refs/heads/TINKERPOP-1545-tp32 Commit: 6c4cbc849f83b9ed44d3c5bbef5e59d67b702593 Parents: b0b9330 Author: Stephen Mallette <sp...@genoprime.com> Authored: Tue Nov 1 09:30:28 2016 -0400 Committer: Stephen Mallette <sp...@genoprime.com> Committed: Fri Dec 16 10:10:14 2016 -0500 ---------------------------------------------------------------------- gremlin-core/pom.xml | 5 +++ .../traversal/util/DefaultTraversal.java | 37 +++++++++++++++++ .../gremlin/python/jsr223/PythonProvider.java | 1 + .../process/traversal/CoreTraversalTest.java | 42 ++++++++++++++++++++ .../TinkerGraphGroovyTranslatorProvider.java | 1 + .../TinkerGraphJavaTranslatorProvider.java | 1 + 6 files changed, 87 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/6c4cbc84/gremlin-core/pom.xml ---------------------------------------------------------------------- diff --git a/gremlin-core/pom.xml b/gremlin-core/pom.xml index e8f3a34..0594448 100644 --- a/gremlin-core/pom.xml +++ b/gremlin-core/pom.xml @@ -61,6 +61,11 @@ limitations under the License. </exclusion> </exclusions> </dependency> + <dependency> + <groupId>org.apache.commons</groupId> + <artifactId>commons-lang3</artifactId> + <version>3.3.1</version> + </dependency> <!-- LOGGING --> <dependency> <groupId>org.slf4j</groupId> http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/6c4cbc84/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/util/DefaultTraversal.java ---------------------------------------------------------------------- diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/util/DefaultTraversal.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/util/DefaultTraversal.java index 3c21e37..6ce6dfe 100644 --- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/util/DefaultTraversal.java +++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/util/DefaultTraversal.java @@ -43,6 +43,9 @@ import java.util.List; import java.util.NoSuchElementException; import java.util.Optional; import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.function.Function; /** * @author Marko A. Rodriguez (http://markorodriguez.com) @@ -325,6 +328,40 @@ public class DefaultTraversal<S, E> implements Traversal.Admin<S, E> { this.graph = graph; } + /** + * Override of {@link Traversal#promise(Function)} that is aware of graph transactions. + */ + @Override + public <T2> CompletableFuture<T2> promise(final Function<Traversal, T2> traversalFunction) { + return this.promise(traversalFunction, Traversal.Admin.traversalExecutorService); + } + + /** + * Override of {@link Traversal#promise(Function)} that is aware of graph transactions. In a transactional graph + * a promise represents the full scope of a transaction, even if the graph is only partially iterated. + */ + @Override + public <T2> CompletableFuture<T2> promise(final Function<Traversal, T2> traversalFunction, final ExecutorService service) { + if (graph != null && graph.features().graph().supportsTransactions()) { + final Function<Traversal, T2> transactionAware = traversal -> { + + try { + if (graph.tx().isOpen()) graph.tx().rollback(); + final T2 obj = traversalFunction.apply(traversal); + if (graph.tx().isOpen()) graph.tx().commit(); + return obj; + } catch (Exception ex) { + if (graph.tx().isOpen()) graph.tx().rollback(); + throw ex; + } + }; + + return Traversal.Admin.super.promise(transactionAware, service); + } else { + return Traversal.Admin.super.promise(traversalFunction, service); + } + } + @Override public boolean equals(final Object other) { return other != null && other.getClass().equals(this.getClass()) && this.equals(((Traversal.Admin) other)); http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/6c4cbc84/gremlin-python/src/test/java/org/apache/tinkerpop/gremlin/python/jsr223/PythonProvider.java ---------------------------------------------------------------------- diff --git a/gremlin-python/src/test/java/org/apache/tinkerpop/gremlin/python/jsr223/PythonProvider.java b/gremlin-python/src/test/java/org/apache/tinkerpop/gremlin/python/jsr223/PythonProvider.java index 9e03884..5dd9c28 100644 --- a/gremlin-python/src/test/java/org/apache/tinkerpop/gremlin/python/jsr223/PythonProvider.java +++ b/gremlin-python/src/test/java/org/apache/tinkerpop/gremlin/python/jsr223/PythonProvider.java @@ -69,6 +69,7 @@ public class PythonProvider extends AbstractGraphProvider { "shouldNeverPropagateANullValuedTraverser", "shouldHidePartitionKeyForValues", "g_withSackXBigInteger_TEN_powX1000X_assignX_V_localXoutXknowsX_barrierXnormSackXX_inXknowsX_barrier_sack", + "shouldUsePromiseAndControlTransactionsIfAvailable", // ProgramTest.Traversals.class.getCanonicalName(), TraversalInterruptionTest.class.getCanonicalName(), http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/6c4cbc84/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/CoreTraversalTest.java ---------------------------------------------------------------------- diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/CoreTraversalTest.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/CoreTraversalTest.java index 68f8217..050f9de 100644 --- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/CoreTraversalTest.java +++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/CoreTraversalTest.java @@ -20,6 +20,7 @@ package org.apache.tinkerpop.gremlin.process.traversal; import org.apache.tinkerpop.gremlin.ExceptionCoverage; import org.apache.tinkerpop.gremlin.FeatureRequirement; +import org.apache.tinkerpop.gremlin.FeatureRequirementSet; import org.apache.tinkerpop.gremlin.LoadGraphWith; import org.apache.tinkerpop.gremlin.process.AbstractGremlinProcessTest; import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversal; @@ -40,6 +41,9 @@ import java.util.Map; import java.util.NoSuchElementException; import java.util.Random; import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.stream.Collectors; import static org.apache.tinkerpop.gremlin.LoadGraphWith.GraphData.MODERN; @@ -307,4 +311,42 @@ public class CoreTraversalTest extends AbstractGremlinProcessTest { } } + + @Test + @FeatureRequirementSet(FeatureRequirementSet.Package.SIMPLE) + public void shouldUsePromiseAndControlTransactionsIfAvailable() throws Exception { + // this test will validate that transactional graphs can properly open/close transactions within a promise. + // as there is a feature check, non-transactional graphs can use this to simply exercise the promise API + final Vertex vAdded = g.addV("person").property("name", "stephen").promise(t -> (Vertex) t.next()).get(10000, TimeUnit.MILLISECONDS); + final Vertex vRead = g.V().has("name", "stephen").next(); + assertEquals(vAdded.id(), vRead.id()); + + // transaction should have been committed at this point so test the count in this thread to validate persistence + assertVertexEdgeCounts(graph, 1, 0); + + // cancel a promise and ensure the transaction ended in failure. hold the traversal in park until it can be + // interrupted, then the promise will have to rollback the transaction. + final CompletableFuture promiseToCancel = g.addV("person").property("name", "marko").sideEffect(traverser -> { + try { + Thread.sleep(100000); + } catch (Exception ignored) { + + } + }).promise(t -> (Vertex) t.next()); + + try { + promiseToCancel.get(500, TimeUnit.MILLISECONDS); + fail("Should have timed out"); + } catch (TimeoutException te) { + + } + + promiseToCancel.cancel(true); + + // graphs that support transactions will rollback the transaction + if (graph.features().graph().supportsTransactions()) + assertVertexEdgeCounts(graph, 1, 0); + else + assertVertexEdgeCounts(graph, 2, 0); + } } http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/6c4cbc84/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/groovy/jsr223/TinkerGraphGroovyTranslatorProvider.java ---------------------------------------------------------------------- diff --git a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/groovy/jsr223/TinkerGraphGroovyTranslatorProvider.java b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/groovy/jsr223/TinkerGraphGroovyTranslatorProvider.java index dd118d7..a595c34 100644 --- a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/groovy/jsr223/TinkerGraphGroovyTranslatorProvider.java +++ b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/groovy/jsr223/TinkerGraphGroovyTranslatorProvider.java @@ -51,6 +51,7 @@ public class TinkerGraphGroovyTranslatorProvider extends TinkerGraphProvider { "g_VX1X_out_injectXv2X_name", "shouldNeverPropagateANoBulkTraverser", "shouldNeverPropagateANullValuedTraverser", + "shouldUsePromiseAndControlTransactionsIfAvailable", GraphComputerTest.class.getCanonicalName(), ProgramTest.Traversals.class.getCanonicalName(), TraversalInterruptionTest.class.getCanonicalName(), http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/6c4cbc84/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/jsr223/TinkerGraphJavaTranslatorProvider.java ---------------------------------------------------------------------- diff --git a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/jsr223/TinkerGraphJavaTranslatorProvider.java b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/jsr223/TinkerGraphJavaTranslatorProvider.java index f40a8c7..15a6c0b 100644 --- a/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/jsr223/TinkerGraphJavaTranslatorProvider.java +++ b/tinkergraph-gremlin/src/test/java/org/apache/tinkerpop/gremlin/tinkergraph/process/jsr223/TinkerGraphJavaTranslatorProvider.java @@ -49,6 +49,7 @@ public class TinkerGraphJavaTranslatorProvider extends TinkerGraphProvider { TraversalInterruptionComputerTest.class.getCanonicalName(), "shouldNeverPropagateANoBulkTraverser", "shouldNeverPropagateANullValuedTraverser", + "shouldUsePromiseAndControlTransactionsIfAvailable", ElementIdStrategyProcessTest.class.getCanonicalName(), EventStrategyProcessTest.class.getCanonicalName(), ProgramTest.Traversals.class.getCanonicalName()));
