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/11/01 13:32:42 UTC
tinkerpop git commit: TINKERPOP-1490 Implemented promise API for
Traversal
Repository: tinkerpop
Updated Branches:
refs/heads/TINKERPOP-1490 [created] 4b37339c6
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).
Project: http://git-wip-us.apache.org/repos/asf/tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/tinkerpop/commit/4b37339c
Tree: http://git-wip-us.apache.org/repos/asf/tinkerpop/tree/4b37339c
Diff: http://git-wip-us.apache.org/repos/asf/tinkerpop/diff/4b37339c
Branch: refs/heads/TINKERPOP-1490
Commit: 4b37339c62f20807900c942d391d6e93458f77e3
Parents: c4682f1
Author: Stephen Mallette <sp...@genoprime.com>
Authored: Tue Nov 1 09:30:28 2016 -0400
Committer: Stephen Mallette <sp...@genoprime.com>
Committed: Tue Nov 1 09:30:28 2016 -0400
----------------------------------------------------------------------
CHANGELOG.asciidoc | 1 +
.../upgrade/release-3.2.x-incubating.asciidoc | 23 +
.../gremlin/process/traversal/Traversal.java | 42 ++
.../traversal/util/DefaultTraversal.java | 39 ++
.../process/traversal/TraversalTest.java | 473 +++++++++++++++++++
.../process/traversal/CoreTraversalTest.java | 42 ++
6 files changed, 620 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/4b37339c/CHANGELOG.asciidoc
----------------------------------------------------------------------
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 48f08ef..9c283ce 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -26,6 +26,7 @@ image::https://raw.githubusercontent.com/apache/tinkerpop/master/docs/static/ima
TinkerPop 3.2.4 (Release Date: NOT OFFICIALLY RELEASED YET)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+* Added `Traversal.promise()` methods to allow for asynchronous traversal processing.
* Added `VertexFeatures.supportsDuplicateMultiProperties()` for graphs that only support unique values in multi-properties.
* Deprecated the "performance" tests in `OptIn`.
* Added `Pick.none` and `Pick.any` to the serializers and importers.
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/4b37339c/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 30daae1..40c3fd9 100644
--- a/docs/src/upgrade/release-3.2.x-incubating.asciidoc
+++ b/docs/src/upgrade/release-3.2.x-incubating.asciidoc
@@ -29,6 +29,29 @@ TinkerPop 3.2.4
Please see the link:https://github.com/apache/tinkerpop/blob/3.2.4/CHANGELOG.asciidoc#release-3-2-4[changelog] for a complete list of all the modifications that are part of this release.
+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]
+
Upgrading for Providers
~~~~~~~~~~~~~~~~~~~~~~~
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/4b37339c/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..62e2f8e 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
@@ -42,7 +42,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.ForkJoinPool;
+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 +146,42 @@ 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 {@code ForkJoinPool} to execute the supplied {@code traversalFunction}.
+ */
+ public default <T> CompletableFuture<T> promise(final Function<Traversal, T> traversalFunction) {
+ return promise(traversalFunction, ForkJoinPool.commonPool());
+ }
+
+ /**
+ * 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 ot 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
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/4b37339c/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 5bd01da..ada0e15 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
@@ -42,6 +42,11 @@ import java.util.Iterator;
import java.util.List;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ForkJoinPool;
+import java.util.function.Function;
+import java.util.function.UnaryOperator;
/**
* @author Marko A. Rodriguez (http://markorodriguez.com)
@@ -316,6 +321,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, ForkJoinPool.commonPool());
+ }
+
+ /**
+ * 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/4b37339c/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/4b37339c/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 bf3acf3..e68d872 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;
@@ -37,6 +38,9 @@ import java.util.List;
import java.util.Map;
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;
@@ -272,4 +276,42 @@ public class CoreTraversalTest extends AbstractGremlinProcessTest {
assertEquals(1, IteratorUtils.count(t));
g.tx().rollback();
}
+
+ @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);
+ }
}