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 2016/12/16 17:19:29 UTC
[03/13] tinkerpop git commit: TINKERPOP-1490 Implemented promise API
for Traversal
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/eb089764
Tree: http://git-wip-us.apache.org/repos/asf/tinkerpop/tree/eb089764
Diff: http://git-wip-us.apache.org/repos/asf/tinkerpop/diff/eb089764
Branch: refs/heads/tp32
Commit: eb089764acb91c24023353a2220e7c84febdec8d
Parents: 347a0a9
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:00:40 2016 -0500
----------------------------------------------------------------------
CHANGELOG.asciidoc | 1 +
.../upgrade/release-3.2.x-incubating.asciidoc | 20 +
gremlin-core/pom.xml | 5 +
.../gremlin/process/traversal/Traversal.java | 50 ++
.../traversal/util/DefaultTraversal.java | 37 ++
.../process/traversal/TraversalTest.java | 473 +++++++++++++++++++
gremlin-groovy/pom.xml | 5 -
.../gremlin/python/jsr223/PythonProvider.java | 1 +
.../process/traversal/CoreTraversalTest.java | 42 ++
.../TinkerGraphGroovyTranslatorProvider.java | 1 +
.../TinkerGraphJavaTranslatorProvider.java | 1 +
11 files changed, 631 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/eb089764/CHANGELOG.asciidoc
----------------------------------------------------------------------
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 8e01ce0..24c75a7 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -52,6 +52,7 @@ TinkerPop 3.2.4 (Release Date: NOT OFFICIALLY RELEASED YET)
* Factored `GremlinPlugin` functionality out of gremlin-groovy and into gremlin-core - related classes were deprecated.
* Added a `force` option for killing sessions without waiting for transaction close or timeout of a currently running job or multiple jobs.
* Deprecated `Session.kill()` and `Session.manualKill()`.
+* Added `Traversal.promise()` methods to allow for asynchronous traversal processing.
* Added `choose(predicate,traversal)` and `choose(traversal,traversal)` to effect if/then-semantics (no else). Equivalent to `choose(x,y,identity())`.
* Removed `ImmutablePath.TailPath` as it is no longer required with new recursion model.
* Removed call stack recursion in `ImmutablePath`.
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/eb089764/docs/src/upgrade/release-3.2.x-incubating.asciidoc
----------------------------------------------------------------------
diff --git a/docs/src/upgrade/release-3.2.x-incubating.asciidoc b/docs/src/upgrade/release-3.2.x-incubating.asciidoc
index 8a184b4..34ff427 100644
--- a/docs/src/upgrade/release-3.2.x-incubating.asciidoc
+++ b/docs/src/upgrade/release-3.2.x-incubating.asciidoc
@@ -32,6 +32,26 @@ Please see the link:https://github.com/apache/tinkerpop/blob/3.2.4/CHANGELOG.asc
Upgrading for Users
~~~~~~~~~~~~~~~~~~~
+Traversal Promises
+^^^^^^^^^^^^^^^^^^
+
+The `Traversal` API now has two `promise()` method overloads. These methods return a promise in the form of a
+`CompleteableFuture`. Usage is as follows:
+
+[source,groovy]
+----
+gremlin> promise = g.V().out().promise{it.next()}
+==>java.util.concurrent.CompletableFuture@4aa3d36[Completed normally]
+gremlin> promise.join()
+==>v[3]
+gremlin> promise.isDone()
+==>true
+gremlin> g.V().out().promise{it.toList()}.thenApply{it.size()}.get()
+==>6
+----
+
+See: link:https://issues.apache.org/jira/browse/TINKERPOP-1490[TINKERPOP-1490]
+
If/Then-Semantics with Choose Step
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/eb089764/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/eb089764/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/Traversal.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/Traversal.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/Traversal.java
index edc18e0..e4ba5a6 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/Traversal.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/Traversal.java
@@ -18,6 +18,7 @@
*/
package org.apache.tinkerpop.gremlin.process.traversal;
+import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversal;
import org.apache.tinkerpop.gremlin.process.traversal.step.TraversalParent;
@@ -42,7 +43,13 @@ import java.util.Optional;
import java.util.Set;
import java.util.Spliterator;
import java.util.Spliterators;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import java.util.function.Consumer;
+import java.util.function.Function;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;
@@ -140,6 +147,43 @@ public interface Traversal<S, E> extends Iterator<E>, Serializable, Cloneable, A
}
/**
+ * Starts a promise to execute a function on the current {@code Traversal} that will be completed in the future.
+ * This implementation uses {@link Admin#traversalExecutorService} to execute the supplied
+ * {@code traversalFunction}.
+ */
+ public default <T> CompletableFuture<T> promise(final Function<Traversal, T> traversalFunction) {
+ return promise(traversalFunction, Admin.traversalExecutorService);
+ }
+
+ /**
+ * Starts a promise to execute a function on the current {@code Traversal} that will be completed in the future.
+ * This implementation uses the caller supplied {@code ExecutorService} to execute the {@code traversalFunction}.
+ */
+ public default <T> CompletableFuture<T> promise(final Function<Traversal, T> traversalFunction, final ExecutorService service) {
+ final CompletableFuture<T> promise = new CompletableFuture<>();
+ final Future iterationFuture = service.submit(() -> {
+ try {
+ promise.complete(traversalFunction.apply(this));
+ } catch (Exception ex) {
+ // the promise may have been cancelled by the caller, in which case, there is no need to attempt
+ // another write on completion
+ if (!promise.isDone()) promise.completeExceptionally(ex);
+ }
+ });
+
+ // if the user cancels the promise then attempt to kill the iteration.
+ promise.exceptionally(t -> {
+ if (t instanceof CancellationException) {
+ iterationFuture.cancel(true);
+ }
+
+ return null;
+ });
+
+ return promise;
+ }
+
+ /**
* Add all the results of the traversal to the provided collection.
*
* @param collection the collection to fill
@@ -253,6 +297,12 @@ public interface Traversal<S, E> extends Iterator<E>, Serializable, Cloneable, A
public interface Admin<S, E> extends Traversal<S, E> {
/**
+ * Service that handles promises.
+ */
+ static final ExecutorService traversalExecutorService = Executors.newCachedThreadPool(
+ new BasicThreadFactory.Builder().namingPattern("traversal-executor-%d").build());
+
+ /**
* Get the {@link Bytecode} associated with the construction of this traversal.
*
* @return the byte code representation of the traversal
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/eb089764/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/eb089764/gremlin-core/src/test/java/org/apache/tinkerpop/gremlin/process/traversal/TraversalTest.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/test/java/org/apache/tinkerpop/gremlin/process/traversal/TraversalTest.java b/gremlin-core/src/test/java/org/apache/tinkerpop/gremlin/process/traversal/TraversalTest.java
new file mode 100644
index 0000000..aa1b99b
--- /dev/null
+++ b/gremlin-core/src/test/java/org/apache/tinkerpop/gremlin/process/traversal/TraversalTest.java
@@ -0,0 +1,473 @@
+/*
+ * 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.process.traversal;
+
+import org.apache.tinkerpop.gremlin.process.remote.traversal.DefaultRemoteTraverser;
+import org.apache.tinkerpop.gremlin.process.traversal.step.TraversalParent;
+import org.apache.tinkerpop.gremlin.process.traversal.traverser.TraverserRequirement;
+import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalInterruptedException;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.hamcrest.core.Is.is;
+import static org.hamcrest.core.IsCollectionContaining.hasItems;
+import static org.hamcrest.core.IsInstanceOf.instanceOf;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
+
+/**
+ * @author Stephen Mallette (http://stephen.genoprime.com)
+ */
+public class TraversalTest {
+
+ private final ExecutorService service = Executors.newFixedThreadPool(2);
+
+ @Test
+ public void shouldTryNext() {
+ final MockTraversal<Integer> t = new MockTraversal<>(1, 2, 3);
+ final Optional<Integer> optFirst = t.tryNext();
+ assertEquals(1, optFirst.get().intValue());
+ final Optional<Integer> optSecond = t.tryNext();
+ assertEquals(2, optSecond.get().intValue());
+ final Optional<Integer> optThird = t.tryNext();
+ assertEquals(3, optThird.get().intValue());
+
+ IntStream.range(0, 100).forEach(i -> {
+ assertThat(t.tryNext().isPresent(), is(false));
+ });
+ }
+
+ @Test
+ public void shouldGetTwoAtATime() {
+ final MockTraversal<Integer> t = new MockTraversal<>(1, 2, 3, 4, 5, 6, 7);
+ final List<Integer> batchOne = t.next(2);
+ assertEquals(2, batchOne.size());
+ assertThat(batchOne, hasItems(1 ,2));
+
+ final List<Integer> batchTwo = t.next(2);
+ assertEquals(2, batchTwo.size());
+ assertThat(batchTwo, hasItems(3 ,4));
+
+ final List<Integer> batchThree = t.next(2);
+ assertEquals(2, batchThree.size());
+ assertThat(batchThree, hasItems(5, 6));
+
+ final List<Integer> batchFour = t.next(2);
+ assertEquals(1, batchFour.size());
+ assertThat(batchFour, hasItems(7));
+
+ final List<Integer> batchFive = t.next(2);
+ assertEquals(0, batchFive.size());
+ }
+
+ @Test
+ public void shouldFillList() {
+ final MockTraversal<Integer> t = new MockTraversal<>(1, 2, 3, 4, 5, 6, 7);
+ final List<Integer> listToFill = new ArrayList<>();
+ final List<Integer> batch = t.fill(listToFill);
+ assertEquals(7, batch.size());
+ assertThat(batch, hasItems(1 ,2, 3, 4, 5, 6, 7));
+ assertThat(t.hasNext(), is(false));
+ assertSame(listToFill, batch);
+ }
+
+ @Test
+ public void shouldStream() {
+ final MockTraversal<Integer> t = new MockTraversal<>(1, 2, 3, 4, 5, 6, 7);
+ final List<Integer> batch = t.toStream().collect(Collectors.toList());
+ assertEquals(7, batch.size());
+ assertThat(batch, hasItems(1 ,2, 3, 4, 5, 6, 7));
+ assertThat(t.hasNext(), is(false));
+ }
+
+ @Test
+ public void shouldPromiseNextThreeUsingForkJoin() throws Exception {
+ final MockTraversal<Integer> t = new MockTraversal<>(1, 2, 3, 4, 5, 6, 7);
+ final CompletableFuture<List<Integer>> promiseFirst = t.promise(traversal -> traversal.next(3));
+ final List<Integer> listFirst = promiseFirst.get();
+ assertEquals(3, listFirst.size());
+ assertThat(listFirst, hasItems(1 ,2, 3));
+ assertThat(t.hasNext(), is(true));
+ assertThat(promiseFirst.isDone(), is(true));
+
+ final CompletableFuture<List<Integer>> promiseSecond = t.promise(traversal -> traversal.next(3));
+ final List<Integer> listSecond = promiseSecond.get();
+ assertEquals(3, listSecond.size());
+ assertThat(listSecond, hasItems(4, 5, 6));
+ assertThat(t.hasNext(), is(true));
+ assertThat(promiseSecond.isDone(), is(true));
+
+ final CompletableFuture<List<Integer>> promiseThird = t.promise(traversal -> traversal.next(3));
+ final List<Integer> listThird = promiseThird.get();
+ assertEquals(1, listThird.size());
+ assertThat(listThird, hasItems(7));
+ assertThat(t.hasNext(), is(false));
+ assertThat(promiseThird.isDone(), is(true));
+
+ final CompletableFuture<Integer> promiseDead = t.promise(traversal -> (Integer) traversal.next());
+ final AtomicBoolean dead = new AtomicBoolean(false);
+ promiseDead.exceptionally(tossed -> {
+ dead.set(tossed instanceof NoSuchElementException);
+ return null;
+ });
+
+ try {
+ promiseDead.get(10000, TimeUnit.MILLISECONDS);
+ fail("Should have gotten an exception");
+ } catch (Exception ex) {
+ if (ex instanceof TimeoutException) {
+ fail("This should not have timed out but should have gotten an exception caught above in the exceptionally() clause");
+ }
+
+ assertThat(ex.getCause(), instanceOf(NoSuchElementException.class));
+ }
+
+ assertThat(dead.get(), is(true));
+ assertThat(t.hasNext(), is(false));
+ assertThat(promiseDead.isDone(), is(true));
+ }
+
+ @Test
+ public void shouldPromiseNextThreeUsingSpecificExecutor() throws Exception {
+ final MockTraversal<Integer> t = new MockTraversal<>(1, 2, 3, 4, 5, 6, 7);
+ final CompletableFuture<List<Integer>> promiseFirst = t.promise(traversal -> traversal.next(3), service);
+ final List<Integer> listFirst = promiseFirst.get();
+ assertEquals(3, listFirst.size());
+ assertThat(listFirst, hasItems(1 ,2, 3));
+ assertThat(t.hasNext(), is(true));
+ assertThat(promiseFirst.isDone(), is(true));
+
+ final CompletableFuture<List<Integer>> promiseSecond = t.promise(traversal -> traversal.next(3), service);
+ final List<Integer> listSecond = promiseSecond.get();
+ assertEquals(3, listSecond.size());
+ assertThat(listSecond, hasItems(4, 5, 6));
+ assertThat(t.hasNext(), is(true));
+ assertThat(promiseSecond.isDone(), is(true));
+
+ final CompletableFuture<List<Integer>> promiseThird = t.promise(traversal -> traversal.next(3), service);
+ final List<Integer> listThird = promiseThird.get();
+ assertEquals(1, listThird.size());
+ assertThat(listThird, hasItems(7));
+ assertThat(t.hasNext(), is(false));
+ assertThat(promiseThird.isDone(), is(true));
+
+ final CompletableFuture<Integer> promiseDead = t.promise(traversal -> (Integer) traversal.next(), service);
+ final AtomicBoolean dead = new AtomicBoolean(false);
+ promiseDead.exceptionally(tossed -> {
+ dead.set(tossed instanceof NoSuchElementException);
+ return null;
+ });
+
+ try {
+ promiseDead.get(10000, TimeUnit.MILLISECONDS);
+ fail("Should have gotten an exception");
+ } catch (Exception ex) {
+ if (ex instanceof TimeoutException) {
+ fail("This should not have timed out but should have gotten an exception caught above in the exceptionally() clause");
+ }
+
+ assertThat(ex.getCause(), instanceOf(NoSuchElementException.class));
+ }
+
+ assertThat(dead.get(), is(true));
+ assertThat(t.hasNext(), is(false));
+ assertThat(promiseDead.isDone(), is(true));
+ }
+
+ @Test
+ public void shouldInterruptTraversalFunction() throws Exception {
+ final Random rand = new Random(1234567890);
+
+ // infinite traversal
+ final MockTraversal<Integer> t = new MockTraversal<>(IntStream.generate(rand::nextInt).iterator());
+
+ // iterate a bunch of it
+ final CompletableFuture<List<Integer>> promise10 = t.promise(traversal -> traversal.next(10), service);
+ assertEquals(10, promise10.get(10000, TimeUnit.MILLISECONDS).size());
+ final CompletableFuture<List<Integer>> promise100 = t.promise(traversal -> traversal.next(100), service);
+ assertEquals(100, promise100.get(10000, TimeUnit.MILLISECONDS).size());
+ final CompletableFuture<List<Integer>> promise1000 = t.promise(traversal -> traversal.next(1000), service);
+ assertEquals(1000, promise1000.get(10000, TimeUnit.MILLISECONDS).size());
+
+ // this is endless, so let's cancel
+ final CompletableFuture<List<Integer>> promiseForevers = t.promise(traversal -> traversal.next(Integer.MAX_VALUE), service);
+
+ // specify what to do on exception
+ final AtomicBoolean failed = new AtomicBoolean(false);
+ promiseForevers.exceptionally(ex -> {
+ failed.set(true);
+ return null;
+ });
+
+ try {
+ // let it actually iterate a moment
+ promiseForevers.get(500, TimeUnit.MILLISECONDS);
+ fail("This should have timed out because the traversal has infinite items in it");
+ } catch (TimeoutException tex) {
+
+ }
+
+ assertThat(promiseForevers.isDone(), is(false));
+ promiseForevers.cancel(true);
+ assertThat(failed.get(), is(true));
+ assertThat(promiseForevers.isDone(), is(true));
+ }
+
+ @Test
+ public void shouldIterate() {
+ final MockTraversal<Integer> t = new MockTraversal<>(1, 2, 3, 4, 5, 6, 7);
+ assertThat(t.hasNext(), is(true));
+ t.iterate();
+ assertThat(t.hasNext(), is(false));
+ }
+
+ private static class MockStep<E> implements Step<E,E> {
+
+ private final Iterator<E> itty;
+
+ MockStep(final Iterator<E> itty) {
+ this.itty = itty;
+ }
+
+ @Override
+ public void addStarts(final Iterator starts) {
+
+ }
+
+ @Override
+ public void addStart(final Traverser.Admin start) {
+
+ }
+
+ @Override
+ public void setPreviousStep(final Step step) {
+
+ }
+
+ @Override
+ public Step getPreviousStep() {
+ return null;
+ }
+
+ @Override
+ public void setNextStep(final Step step) {
+
+ }
+
+ @Override
+ public Step getNextStep() {
+ return null;
+ }
+
+ @Override
+ public Traversal.Admin getTraversal() {
+ return null;
+ }
+
+ @Override
+ public void setTraversal(final Traversal.Admin traversal) {
+
+ }
+
+ @Override
+ public void reset() {
+
+ }
+
+ @Override
+ public Step clone() {
+ return null;
+ }
+
+ @Override
+ public Set<String> getLabels() {
+ return null;
+ }
+
+ @Override
+ public void addLabel(final String label) {
+
+ }
+
+ @Override
+ public void removeLabel(final String label) {
+
+ }
+
+ @Override
+ public void setId(final String id) {
+
+ }
+
+ @Override
+ public String getId() {
+ return null;
+ }
+
+ @Override
+ public boolean hasNext() {
+ return itty.hasNext();
+ }
+
+ @Override
+ public Traverser.Admin<E> next() {
+ return new DefaultRemoteTraverser<>(itty.next(), 1L);
+ }
+ }
+
+
+ private static class MockTraversal<T> implements Traversal.Admin<T,T> {
+
+ private Iterator<T> itty;
+
+ private Step mockEndStep;
+
+ private List<Step> steps;
+
+ MockTraversal(final T... objects) {
+ this(Arrays.asList(objects));
+ }
+
+ MockTraversal(final List<T> list) {
+ this(list.iterator());
+ }
+
+ MockTraversal(final Iterator<T> itty) {
+ this.itty = itty;
+ mockEndStep = new MockStep<>(itty);
+ steps = Collections.singletonList(mockEndStep);
+ }
+
+ @Override
+ public Bytecode getBytecode() {
+ return null;
+ }
+
+ @Override
+ public List<Step> getSteps() {
+ return steps;
+ }
+
+ @Override
+ public <S2, E2> Admin<S2, E2> addStep(final int index, final Step<?, ?> step) throws IllegalStateException {
+ return null;
+ }
+
+ @Override
+ public <S2, E2> Admin<S2, E2> removeStep(final int index) throws IllegalStateException {
+ return null;
+ }
+
+ @Override
+ public void applyStrategies() throws IllegalStateException {
+
+ }
+
+ @Override
+ public TraverserGenerator getTraverserGenerator() {
+ return null;
+ }
+
+ @Override
+ public Set<TraverserRequirement> getTraverserRequirements() {
+ return null;
+ }
+
+ @Override
+ public void setSideEffects(final TraversalSideEffects sideEffects) {
+
+ }
+
+ @Override
+ public TraversalSideEffects getSideEffects() {
+ return null;
+ }
+
+ @Override
+ public void setStrategies(final TraversalStrategies strategies) {
+
+ }
+
+ @Override
+ public TraversalStrategies getStrategies() {
+ return null;
+ }
+
+ @Override
+ public void setParent(final TraversalParent step) {
+
+ }
+
+ @Override
+ public TraversalParent getParent() {
+ return null;
+ }
+
+ @Override
+ public Admin<T, T> clone() {
+ return null;
+ }
+
+ @Override
+ public boolean isLocked() {
+ return false;
+ }
+
+ @Override
+ public Optional<Graph> getGraph() {
+ return null;
+ }
+
+ @Override
+ public void setGraph(final Graph graph) {
+
+ }
+
+ @Override
+ public boolean hasNext() {
+ return itty.hasNext();
+ }
+
+ @Override
+ public T next() {
+ if (Thread.interrupted()) throw new TraversalInterruptedException();
+ return itty.next();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/eb089764/gremlin-groovy/pom.xml
----------------------------------------------------------------------
diff --git a/gremlin-groovy/pom.xml b/gremlin-groovy/pom.xml
index acc51b7..dae5e8a 100644
--- a/gremlin-groovy/pom.xml
+++ b/gremlin-groovy/pom.xml
@@ -61,11 +61,6 @@ limitations under the License.
<classifier>indy</classifier>
</dependency>
<dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-lang3</artifactId>
- <version>3.3.1</version>
- </dependency>
- <dependency>
<groupId>com.github.jeremyh</groupId>
<artifactId>jBCrypt</artifactId>
<version>jbcrypt-0.4</version>
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/eb089764/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/eb089764/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/eb089764/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/eb089764/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()));