You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by da...@apache.org on 2016/12/07 21:37:46 UTC
[1/9] tinkerpop git commit: TINKERPOP-1490 Implemented promise API
for Traversal
Repository: tinkerpop
Updated Branches:
refs/heads/TINKERPOP-1490 77b341647 -> 2cb737ef8
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/3a605b9a
Tree: http://git-wip-us.apache.org/repos/asf/tinkerpop/tree/3a605b9a
Diff: http://git-wip-us.apache.org/repos/asf/tinkerpop/diff/3a605b9a
Branch: refs/heads/TINKERPOP-1490
Commit: 3a605b9a08ddd45bba6586ced16e0dbc2f2f033e
Parents: 64609a7
Author: Stephen Mallette <sp...@genoprime.com>
Authored: Tue Nov 1 09:30:28 2016 -0400
Committer: Stephen Mallette <sp...@genoprime.com>
Committed: Tue Nov 15 09:51:57 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/3a605b9a/CHANGELOG.asciidoc
----------------------------------------------------------------------
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 3d963c9..a2e1aac 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -38,6 +38,7 @@ TinkerPop 3.2.4 (Release Date: NOT OFFICIALLY RELEASED YET)
* Improved ability to release resources in `GraphProvider` instances in the test suite.
* 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/3a605b9a/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 8574b88..04f3b79 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/3a605b9a/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/3a605b9a/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/3a605b9a/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/3a605b9a/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/3a605b9a/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/3a605b9a/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 fe04156..3bf4239 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
@@ -72,6 +72,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/3a605b9a/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/3a605b9a/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/3a605b9a/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()));
[4/9] tinkerpop git commit: TINKERPOP-1490 Restructured
Traversal.promise()
Posted by da...@apache.org.
TINKERPOP-1490 Restructured Traversal.promise()
No longer uses an ExecutorService and is only applicable to "remote" traversals. Moved the commons-lang dependency back to gremlin-groovy for now.
Project: http://git-wip-us.apache.org/repos/asf/tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/tinkerpop/commit/ca2ccf2c
Tree: http://git-wip-us.apache.org/repos/asf/tinkerpop/tree/ca2ccf2c
Diff: http://git-wip-us.apache.org/repos/asf/tinkerpop/diff/ca2ccf2c
Branch: refs/heads/TINKERPOP-1490
Commit: ca2ccf2c6085deb6e1af2bab032697bab92d9612
Parents: 3a605b9
Author: Stephen Mallette <sp...@genoprime.com>
Authored: Fri Nov 11 12:51:40 2016 -0500
Committer: Stephen Mallette <sp...@genoprime.com>
Committed: Tue Nov 15 09:52:58 2016 -0500
----------------------------------------------------------------------
gremlin-core/pom.xml | 5 -
.../process/remote/RemoteConnection.java | 12 +-
.../remote/traversal/RemoteTraversal.java | 2 +-
.../remote/traversal/step/map/RemoteStep.java | 32 +++-
.../gremlin/process/traversal/Traversal.java | 57 ++------
.../traversal/util/DefaultTraversal.java | 37 -----
.../process/traversal/TraversalTest.java | 145 -------------------
.../tinkerpop/gremlin/driver/Connection.java | 6 +-
.../driver/remote/DriverRemoteConnection.java | 14 ++
.../driver/remote/DriverRemoteTraversal.java | 16 +-
.../DriverRemoteTraversalSideEffects.java | 22 ++-
.../DriverRemoteTraversalSideEffectsTest.java | 12 +-
gremlin-groovy/pom.xml | 5 +
.../server/GremlinServerIntegrateTest.java | 25 ++++
.../process/traversal/CoreTraversalTest.java | 42 ------
15 files changed, 131 insertions(+), 301 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/ca2ccf2c/gremlin-core/pom.xml
----------------------------------------------------------------------
diff --git a/gremlin-core/pom.xml b/gremlin-core/pom.xml
index 0594448..e8f3a34 100644
--- a/gremlin-core/pom.xml
+++ b/gremlin-core/pom.xml
@@ -61,11 +61,6 @@ 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/ca2ccf2c/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/remote/RemoteConnection.java
----------------------------------------------------------------------
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 8506ad7..f4e3976 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
@@ -24,6 +24,7 @@ import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
import java.util.Iterator;
+import java.util.concurrent.CompletableFuture;
/**
* A simple abstraction of a "connection" to a "server" that is capable of processing a {@link Traversal} and
@@ -43,9 +44,16 @@ public interface RemoteConnection extends AutoCloseable {
public <E> Iterator<Traverser.Admin<E>> submit(final Traversal<?, E> traversal) throws RemoteConnectionException;
/**
- * Submits {@link Traversal} {@link Bytecode} to a server and returns a {@link Traversal}.
- * The {@link Traversal} is an abstraction over two types of results that can be returned as part of the
+ * Submits {@link Traversal} {@link Bytecode} to a server and returns 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.
*/
public <E> RemoteTraversal<?,E> submit(final Bytecode bytecode) throws RemoteConnectionException;
+
+ /**
+ * 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.
+ */
+ public <E> CompletableFuture<RemoteTraversal<?, E>> submitAsync(final Bytecode bytecode) throws RemoteConnectionException;
}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/ca2ccf2c/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/remote/traversal/RemoteTraversal.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/remote/traversal/RemoteTraversal.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/remote/traversal/RemoteTraversal.java
index 9c893c2..57b0cda 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/remote/traversal/RemoteTraversal.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/remote/traversal/RemoteTraversal.java
@@ -39,7 +39,7 @@ import org.apache.tinkerpop.gremlin.process.traversal.step.util.AbstractStep;
public interface RemoteTraversal<S,E> extends Traversal.Admin<S,E> {
/**
- * Returns remote side-effects generated by the traversal so that they can accessible to the client. Note that
+ * Returns remote side-effects generated by the traversal so that they can be accessible to the client. Note that
* "side-effect" refers to the value in "a" in the traversal {@code g.V().aggregate('a').values('name')}.
*/
@Override
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/ca2ccf2c/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/remote/traversal/step/map/RemoteStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/remote/traversal/step/map/RemoteStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/remote/traversal/step/map/RemoteStep.java
index 6b2be96..3e19097 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/remote/traversal/step/map/RemoteStep.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/remote/traversal/step/map/RemoteStep.java
@@ -21,12 +21,17 @@ package org.apache.tinkerpop.gremlin.process.remote.traversal.step.map;
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.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.AbstractStep;
+import org.apache.tinkerpop.gremlin.process.traversal.util.DefaultTraversal;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import java.util.NoSuchElementException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
/**
* Sends a {@link Traversal} to a {@link RemoteConnection} and iterates back the results.
@@ -38,6 +43,7 @@ public final class RemoteStep<S, E> extends AbstractStep<S, E> {
private transient RemoteConnection remoteConnection;
private RemoteTraversal<?, E> remoteTraversal;
+ private final AtomicReference<CompletableFuture<Traversal<?, E>>> traversalFuture = new AtomicReference<>(null);
public RemoteStep(final Traversal.Admin traversal, final RemoteConnection remoteConnection) {
super(traversal);
@@ -51,14 +57,26 @@ public final class RemoteStep<S, E> extends AbstractStep<S, E> {
@Override
protected Traverser.Admin<E> processNextStart() throws NoSuchElementException {
- if (null == this.remoteTraversal) {
- try {
- this.remoteTraversal = this.remoteConnection.submit(this.traversal.getBytecode());
- this.traversal.setSideEffects(this.remoteTraversal.getSideEffects());
- } catch (final RemoteConnectionException sce) {
- throw new IllegalStateException(sce);
+ if (null == this.remoteTraversal) promise().join();
+ return this.remoteTraversal.nextTraverser();
+ }
+
+ /**
+ * Submits the traversal asynchronously to a "remote" using {@link RemoteConnection#submitAsync(Bytecode)}.
+ */
+ public CompletableFuture<Traversal<?, E>> promise() {
+ try {
+ if (null == traversalFuture.get()) {
+ traversalFuture.set(this.remoteConnection.submitAsync(this.traversal.getBytecode()).<Traversal<?, E>>thenApply(t -> {
+ this.remoteTraversal = (RemoteTraversal<?, E>) t;
+ this.traversal.setSideEffects(this.remoteTraversal.getSideEffects());
+ return traversal;
+ }));
}
+
+ return traversalFuture.get();
+ } catch (RemoteConnectionException rce) {
+ throw new IllegalStateException(rce);
}
- return this.remoteTraversal.nextTraverser();
}
}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/ca2ccf2c/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 e4ba5a6..04f5127 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,8 +18,9 @@
*/
package org.apache.tinkerpop.gremlin.process.traversal;
-import org.apache.commons.lang3.concurrent.BasicThreadFactory;
+import org.apache.commons.configuration.Configuration;
import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
+import org.apache.tinkerpop.gremlin.process.remote.traversal.step.map.RemoteStep;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversal;
import org.apache.tinkerpop.gremlin.process.traversal.step.TraversalParent;
import org.apache.tinkerpop.gremlin.process.traversal.step.sideEffect.ProfileSideEffectStep;
@@ -43,11 +44,7 @@ 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;
@@ -148,39 +145,21 @@ 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}.
+ * Note that this method can only be used if the {@code Traversal} is constructed using
+ * {@link TraversalSource#withRemote(Configuration)}. Calling this method otherwise will yield an
+ * {@code IllegalStateException}.
*/
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;
+ // apply strategies to see if RemoteStrategy has any effect (i.e. add RemoteStep)
+ if (!this.asAdmin().isLocked()) this.asAdmin().applyStrategies();
+
+ // use the end step so the results are bulked
+ final Step<?, E> endStep = this.asAdmin().getEndStep();
+ if (endStep instanceof RemoteStep) {
+ return ((RemoteStep) endStep).promise().thenApply(traversalFunction);
+ } else {
+ throw new IllegalStateException("Only traversals created using withRemote() can be used in an async way");
+ }
}
/**
@@ -297,12 +276,6 @@ 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/ca2ccf2c/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 6ce6dfe..3c21e37 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,9 +43,6 @@ 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)
@@ -328,40 +325,6 @@ 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/ca2ccf2c/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
index aa1b99b..c427d8e 100644
--- 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
@@ -30,34 +30,22 @@ 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);
@@ -117,139 +105,6 @@ public class TraversalTest {
}
@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));
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/ca2ccf2c/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
----------------------------------------------------------------------
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
index 972e838..9a2180e 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/Connection.java
@@ -208,7 +208,7 @@ final class Connection {
logger.debug(String.format("Write on connection %s failed", thisConnection.getConnectionInfo()), f.cause());
thisConnection.isDead = true;
thisConnection.returnToPool();
- future.completeExceptionally(f.cause());
+ cluster.executor().submit(() -> future.completeExceptionally(f.cause()));
} else {
final LinkedBlockingQueue<Result> resultLinkedBlockingQueue = new LinkedBlockingQueue<>();
final CompletableFuture<Void> readCompleted = new CompletableFuture<>();
@@ -250,8 +250,8 @@ final class Connection {
final ResultQueue handler = new ResultQueue(resultLinkedBlockingQueue, readCompleted);
pending.put(requestMessage.getRequestId(), handler);
- future.complete(new ResultSet(handler, cluster.executor(), readCompleted,
- requestMessage, pool.host));
+ cluster.executor().submit(() -> future.complete(
+ new ResultSet(handler, cluster.executor(), readCompleted, requestMessage, pool.host)));
}
});
channel.writeAndFlush(requestMessage, requestPromise);
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/ca2ccf2c/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteConnection.java
----------------------------------------------------------------------
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 bb2d33d..be3fa28 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
@@ -37,6 +37,7 @@ import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
import java.util.Iterator;
import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;
/**
@@ -163,6 +164,10 @@ public class DriverRemoteConnection implements RemoteConnection {
}
}
+ /**
+ * @deprecated As of release 3.2.2, replaced by {@link #submit(Bytecode)}.
+ */
+ @Deprecated
@Override
public <E> Iterator<Traverser.Admin<E>> submit(final Traversal<?, E> t) throws RemoteConnectionException {
try {
@@ -189,6 +194,15 @@ public class DriverRemoteConnection implements RemoteConnection {
}
@Override
+ public <E> CompletableFuture<RemoteTraversal<?, E>> submitAsync(final Bytecode bytecode) throws RemoteConnectionException {
+ try {
+ return client.submitAsync(bytecode).thenApply(rs -> new DriverRemoteTraversal<>(rs, client, attachElements, conf));
+ } catch (Exception ex) {
+ throw new RemoteConnectionException(ex);
+ }
+ }
+
+ @Override
public void close() throws Exception {
try {
client.close();
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/ca2ccf2c/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTraversal.java
----------------------------------------------------------------------
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTraversal.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTraversal.java
index 88ee794..d3f290c 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTraversal.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTraversal.java
@@ -66,28 +66,18 @@ public class DriverRemoteTraversal<S, E> extends AbstractRemoteTraversal<S, E> {
}
this.rs = rs;
- this.sideEffects = new DriverRemoteTraversalSideEffects(
- client,
+ this.sideEffects = new DriverRemoteTraversalSideEffects(client,
rs.getOriginalRequestMessage().getRequestId(),
- rs.getHost());
+ rs.getHost(), rs.allItemsAvailableAsync());
}
/**
* Gets a side-effect from the server. Do not call this method prior to completing the iteration of the
- * {@link DriverRemoteTraversal} that spawned this as the side-effect will not be ready. If this method is called
- * prior to iteration being complete, then it will block until the traversal notifies it of completion. Generally
+ * {@link DriverRemoteTraversal} that spawned this as the side-effect will not be ready. Generally
* speaking, the common user would not get side-effects this way - they would use a call to {@code cap()}.
*/
@Override
public RemoteTraversalSideEffects getSideEffects() {
- // wait for the read to complete (i.e. iteration on the server) before allowing the caller to get the
- // side-effect. calling prior to this will result in the side-effect not being found. of course, the
- // bad part here is that the method blocks indefinitely waiting for the result, but it prevents the
- // test failure problems that happen on slower systems. in practice, it's unlikely that a user would
- // try to get a side-effect prior to iteration, but since the API allows it, this at least prevents
- // the error.
- rs.allItemsAvailableAsync().join();
-
return this.sideEffects;
}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/ca2ccf2c/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTraversalSideEffects.java
----------------------------------------------------------------------
diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTraversalSideEffects.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTraversalSideEffects.java
index 8d6fa98..4305567 100644
--- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTraversalSideEffects.java
+++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTraversalSideEffects.java
@@ -33,6 +33,7 @@ import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
/**
* Java driver implementation of {@link TraversalSideEffects}. This class is not thread safe.
@@ -50,15 +51,26 @@ public class DriverRemoteTraversalSideEffects extends AbstractRemoteTraversalSid
private boolean closed = false;
private boolean retrievedAllKeys = false;
+ private final CompletableFuture<Void> ready;
- public DriverRemoteTraversalSideEffects(final Client client, final UUID serverSideEffect, final Host host) {
+ public DriverRemoteTraversalSideEffects(final Client client, final UUID serverSideEffect, final Host host,
+ final CompletableFuture<Void> ready) {
this.client = client;
this.serverSideEffect = serverSideEffect;
this.host = host;
+ this.ready = ready;
}
@Override
public <V> V get(final String key) throws IllegalArgumentException {
+ // wait for the read to complete (i.e. iteration on the server) before allowing the caller to get the
+ // side-effect. calling prior to this will result in the side-effect not being found. of course, the
+ // bad part here is that the method blocks indefinitely waiting for the result, but it prevents the
+ // test failure problems that happen on slower systems. in practice, it's unlikely that a user would
+ // try to get a side-effect prior to iteration, but since the API allows it, this at least prevents
+ // the error.
+ ready.join();
+
if (!keys().contains(key)) throw TraversalSideEffects.Exceptions.sideEffectKeyDoesNotExist(key);
if (!sideEffects.containsKey(key)) {
@@ -91,6 +103,14 @@ public class DriverRemoteTraversalSideEffects extends AbstractRemoteTraversalSid
@Override
public Set<String> keys() {
+ // wait for the read to complete (i.e. iteration on the server) before allowing the caller to get the
+ // side-effect. calling prior to this will result in the side-effect not being found. of course, the
+ // bad part here is that the method blocks indefinitely waiting for the result, but it prevents the
+ // test failure problems that happen on slower systems. in practice, it's unlikely that a user would
+ // try to get a side-effect prior to iteration, but since the API allows it, this at least prevents
+ // the error.
+ ready.join();
+
if (closed && !retrievedAllKeys) throw new IllegalStateException("Traversal has been closed - side-effect keys cannot be retrieved");
if (!retrievedAllKeys) {
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/ca2ccf2c/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTraversalSideEffectsTest.java
----------------------------------------------------------------------
diff --git a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTraversalSideEffectsTest.java b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTraversalSideEffectsTest.java
index 27d0079..4e6df93 100644
--- a/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTraversalSideEffectsTest.java
+++ b/gremlin-driver/src/test/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteTraversalSideEffectsTest.java
@@ -52,7 +52,9 @@ public class DriverRemoteTraversalSideEffectsTest extends AbstractResultQueueTes
mockClientForCall(client);
final UUID sideEffectKey = UUID.fromString("31dec2c6-b214-4a6f-a68b-996608dce0d9");
- final TraversalSideEffects sideEffects = new DriverRemoteTraversalSideEffects(client, sideEffectKey, null);
+ final CompletableFuture<Void> ready = new CompletableFuture<>();
+ ready.complete(null);
+ final TraversalSideEffects sideEffects = new DriverRemoteTraversalSideEffects(client, sideEffectKey, null, ready);
assertEquals(1, sideEffects.keys().size());
sideEffects.close();
@@ -73,7 +75,9 @@ public class DriverRemoteTraversalSideEffectsTest extends AbstractResultQueueTes
mockClientForCall(client);
mockClientForCall(client);
final UUID sideEffectKey = UUID.fromString("31dec2c6-b214-4a6f-a68b-996608dce0d9");
- final TraversalSideEffects sideEffects = new DriverRemoteTraversalSideEffects(client, sideEffectKey, null);
+ final CompletableFuture<Void> ready = new CompletableFuture<>();
+ ready.complete(null);
+ final TraversalSideEffects sideEffects = new DriverRemoteTraversalSideEffects(client, sideEffectKey, null, ready);
assertNotNull(sideEffects.get("test-0"));
sideEffects.close();
@@ -93,7 +97,9 @@ public class DriverRemoteTraversalSideEffectsTest extends AbstractResultQueueTes
mockClientForCall(client);
final UUID sideEffectKey = UUID.fromString("31dec2c6-b214-4a6f-a68b-996608dce0d9");
- final TraversalSideEffects sideEffects = new DriverRemoteTraversalSideEffects(client, sideEffectKey, null);
+ final CompletableFuture<Void> ready = new CompletableFuture<>();
+ ready.complete(null);
+ final TraversalSideEffects sideEffects = new DriverRemoteTraversalSideEffects(client, sideEffectKey, null, ready);
sideEffects.close();
sideEffects.close();
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/ca2ccf2c/gremlin-groovy/pom.xml
----------------------------------------------------------------------
diff --git a/gremlin-groovy/pom.xml b/gremlin-groovy/pom.xml
index dae5e8a..b82c986 100644
--- a/gremlin-groovy/pom.xml
+++ b/gremlin-groovy/pom.xml
@@ -65,6 +65,11 @@ limitations under the License.
<artifactId>jBCrypt</artifactId>
<version>jbcrypt-0.4</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-lang3</artifactId>
+ <version>3.3.1</version>
+ </dependency>
<!-- TEST -->
<dependency>
<groupId>org.slf4j</groupId>
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/ca2ccf2c/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java
----------------------------------------------------------------------
diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java
index b3dbe29..420bd05 100644
--- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java
+++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java
@@ -46,12 +46,14 @@ import org.apache.tinkerpop.gremlin.groovy.jsr223.customizer.InterpreterModeCust
import org.apache.tinkerpop.gremlin.groovy.jsr223.customizer.SimpleSandboxExtension;
import org.apache.tinkerpop.gremlin.groovy.jsr223.customizer.TimedInterruptCustomizerProvider;
import org.apache.tinkerpop.gremlin.process.remote.RemoteGraph;
+import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversal;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.BulkSet;
import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.T;
import org.apache.tinkerpop.gremlin.server.channel.NioChannelizer;
+import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedVertex;
import org.apache.tinkerpop.gremlin.structure.util.empty.EmptyGraph;
import org.apache.tinkerpop.gremlin.util.Log4jRecordingAppender;
@@ -76,6 +78,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@@ -990,4 +993,26 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration
final BulkSet localBSideEffects = se.get("b");
assertThat(localBSideEffects.isEmpty(), is(false));
}
+
+ @Test
+ public void shouldDoNonBlockingPromiseWithRemote() throws Exception {
+ final Graph graph = EmptyGraph.instance();
+ final GraphTraversalSource g = graph.traversal().withRemote(conf);
+ g.addV("person").property("age", 20).promise(Traversal::iterate).join();
+ g.addV("person").property("age", 10).promise(Traversal::iterate).join();
+ assertEquals(50L, g.V().hasLabel("person").map(Lambda.function("it.get().value('age') + 10")).sum().promise(t -> t.next()).join());
+ g.addV("person").property("age", 20).promise(Traversal::iterate).join();
+
+ final Traversal traversal = g.V().hasLabel("person").has("age", 20).values("age");
+ assertEquals(20, traversal.promise(t -> ((Traversal) t).next(1).get(0)).join());
+ assertEquals(20, traversal.next());
+ assertThat(traversal.hasNext(), is(false));
+
+ final Traversal traversalCloned = g.V().hasLabel("person").has("age", 20).values("age");
+ assertEquals(20, traversalCloned.next());
+ assertEquals(20, traversalCloned.promise(t -> ((Traversal) t).next(1).get(0)).join());
+ assertThat(traversalCloned.promise(t -> ((Traversal) t).hasNext()).join(), is(false));
+
+ assertEquals(3, g.V().promise(Traversal::toList).join().size());
+ }
}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/ca2ccf2c/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 050f9de..68f8217 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,7 +20,6 @@ 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;
@@ -41,9 +40,6 @@ 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;
@@ -311,42 +307,4 @@ 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);
- }
}
[2/9] tinkerpop git commit: TINKERPOP-1490 Cleaned up documentation
based on latest changes to promise API
Posted by da...@apache.org.
TINKERPOP-1490 Cleaned up documentation based on latest changes to promise API
Project: http://git-wip-us.apache.org/repos/asf/tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/tinkerpop/commit/08cfa806
Tree: http://git-wip-us.apache.org/repos/asf/tinkerpop/tree/08cfa806
Diff: http://git-wip-us.apache.org/repos/asf/tinkerpop/diff/08cfa806
Branch: refs/heads/TINKERPOP-1490
Commit: 08cfa806dcd838f396485bcb3d7c5c74a68d0a75
Parents: bb0b22e
Author: Stephen Mallette <sp...@genoprime.com>
Authored: Mon Nov 14 06:38:18 2016 -0500
Committer: Stephen Mallette <sp...@genoprime.com>
Committed: Tue Nov 15 09:52:58 2016 -0500
----------------------------------------------------------------------
CHANGELOG.asciidoc | 2 +-
docs/src/upgrade/release-3.2.x-incubating.asciidoc | 4 +++-
2 files changed, 4 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/08cfa806/CHANGELOG.asciidoc
----------------------------------------------------------------------
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index a2e1aac..93e418c 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -38,7 +38,7 @@ TinkerPop 3.2.4 (Release Date: NOT OFFICIALLY RELEASED YET)
* Improved ability to release resources in `GraphProvider` instances in the test suite.
* 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 `Traversal.promise()` method to allow for asynchronous traversal processing on "remote" traversals.
* 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/08cfa806/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 04f3b79..283688f 100644
--- a/docs/src/upgrade/release-3.2.x-incubating.asciidoc
+++ b/docs/src/upgrade/release-3.2.x-incubating.asciidoc
@@ -35,7 +35,7 @@ Upgrading for Users
Traversal Promises
^^^^^^^^^^^^^^^^^^
-The `Traversal` API now has two `promise()` method overloads. These methods return a promise in the form of a
+The `Traversal` API now has a new `promise()` method. These methods return a promise in the form of a
`CompleteableFuture`. Usage is as follows:
[source,groovy]
@@ -50,6 +50,8 @@ gremlin> g.V().out().promise{it.toList()}.thenApply{it.size()}.get()
==>6
----
+At this time, this method is only used for traversals that are configured using `withRemote()`.
+
See: link:https://issues.apache.org/jira/browse/TINKERPOP-1490[TINKERPOP-1490]
If/Then-Semantics with Choose Step
[9/9] tinkerpop git commit: raise error if side effect method called
while loop is running
Posted by da...@apache.org.
raise error if side effect method called while loop is running
Project: http://git-wip-us.apache.org/repos/asf/tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/tinkerpop/commit/2cb737ef
Tree: http://git-wip-us.apache.org/repos/asf/tinkerpop/tree/2cb737ef
Diff: http://git-wip-us.apache.org/repos/asf/tinkerpop/diff/2cb737ef
Branch: refs/heads/TINKERPOP-1490
Commit: 2cb737ef8f467390d9fb01443963852e3973e0dc
Parents: 522960c
Author: davebshow <da...@gmail.com>
Authored: Wed Dec 7 16:37:05 2016 -0500
Committer: davebshow <da...@gmail.com>
Committed: Wed Dec 7 16:37:05 2016 -0500
----------------------------------------------------------------------
.../driver/driver_remote_connection.py | 4 ++--
.../gremlin_python/driver/remote_connection.py | 12 +++++++++-
.../driver/test_driver_remote_connection.py | 24 +++++++++++---------
3 files changed, 26 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/2cb737ef/gremlin-python/src/main/jython/gremlin_python/driver/driver_remote_connection.py
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/jython/gremlin_python/driver/driver_remote_connection.py b/gremlin-python/src/main/jython/gremlin_python/driver/driver_remote_connection.py
index bcf6d15..4d8d7e2 100644
--- a/gremlin-python/src/main/jython/gremlin_python/driver/driver_remote_connection.py
+++ b/gremlin-python/src/main/jython/gremlin_python/driver/driver_remote_connection.py
@@ -54,13 +54,13 @@ class DriverRemoteConnection(RemoteConnection):
request_id = str(uuid.uuid4())
traversers = self._loop.run_sync(lambda: self.submit_traversal_bytecode(request_id, bytecode))
keys, value, close = self._get_side_effect_lambdas(request_id)
- return RemoteTraversal(iter(traversers), RemoteTraversalSideEffects(keys, value, close))
+ return RemoteTraversal(iter(traversers), RemoteTraversalSideEffects(keys, value, close, self._loop))
def submit_async(self, bytecode):
request_id = str(uuid.uuid4())
future_traversers = self.submit_traversal_bytecode(request_id, bytecode)
keys, value, close = self._get_side_effect_lambdas(request_id)
- side_effects = RemoteTraversalSideEffects(keys, value, close)
+ side_effects = RemoteTraversalSideEffects(keys, value, close, self._loop)
return RemoteTraversal(future_traversers, side_effects)
@gen.coroutine
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/2cb737ef/gremlin-python/src/main/jython/gremlin_python/driver/remote_connection.py
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/jython/gremlin_python/driver/remote_connection.py b/gremlin-python/src/main/jython/gremlin_python/driver/remote_connection.py
index 93c92b7..f7ed48e 100644
--- a/gremlin-python/src/main/jython/gremlin_python/driver/remote_connection.py
+++ b/gremlin-python/src/main/jython/gremlin_python/driver/remote_connection.py
@@ -57,20 +57,27 @@ class RemoteTraversal(Traversal):
class RemoteTraversalSideEffects(TraversalSideEffects):
- def __init__(self, keys_lambda, value_lambda, close_lambda):
+ def __init__(self, keys_lambda, value_lambda, close_lambda, loop):
self._keys_lambda = keys_lambda
self._value_lambda = value_lambda
self._close_lambda = close_lambda
+ self._loop = loop
self._keys = set()
self._side_effects = {}
self._closed = False
def keys(self):
+ if self._loop._running:
+ raise RuntimeError("Cannot call side effect methods"
+ "while event loop is running")
if not self._closed:
self._keys = self._keys_lambda()
return self._keys
def get(self, key):
+ if self._loop._running:
+ raise RuntimeError("Cannot call side effect methods"
+ "while event loop is running")
if not self._side_effects.get(key):
if not self._closed:
results = self._value_lambda(key)
@@ -81,6 +88,9 @@ class RemoteTraversalSideEffects(TraversalSideEffects):
return self._side_effects[key]
def close(self):
+ if self._loop._running:
+ raise RuntimeError("Cannot call side effect methods"
+ "while event loop is running")
results = self._close_lambda()
self._closed = True
return results
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/2cb737ef/gremlin-python/src/main/jython/tests/driver/test_driver_remote_connection.py
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/jython/tests/driver/test_driver_remote_connection.py b/gremlin-python/src/main/jython/tests/driver/test_driver_remote_connection.py
index c9e64c5..783cf7e 100644
--- a/gremlin-python/src/main/jython/tests/driver/test_driver_remote_connection.py
+++ b/gremlin-python/src/main/jython/tests/driver/test_driver_remote_connection.py
@@ -211,6 +211,7 @@ class TestDriverRemoteConnection(TestCase):
assert count == 6
loop.run_sync(go)
+ connection.close()
def test_promise_side_effects(self):
loop = ioloop.IOLoop.current()
@@ -224,22 +225,17 @@ class TestDriverRemoteConnection(TestCase):
@gen.coroutine
def go():
traversal = yield g.V().aggregate('a').promise()
- # Trying to get side effect keys throws error - BAD
+ # Calling synchronous side effect methods from coroutine raises.
with pytest.raises(RuntimeError):
keys = traversal.side_effects.keys()
- # IOLoop is now hosed.
- loop.run_sync(go)
+ with pytest.raises(RuntimeError):
+ keys = traversal.side_effects.get('a')
- # Get a new IOLoop - this should happen for each test case.
- connection.close()
- ioloop.IOLoop.clear_instance()
- loop.close()
- loop = ioloop.IOLoop()
- loop.make_current()
+ with pytest.raises(RuntimeError):
+ keys = traversal.side_effects.close()
- connection = DriverRemoteConnection('ws://localhost:45940/gremlin', 'g')
- g = Graph().traversal().withRemote(connection)
+ loop.run_sync(go)
# If we return the traversal though, we can use side effects per usual.
@gen.coroutine
@@ -251,6 +247,12 @@ class TestDriverRemoteConnection(TestCase):
traversal = loop.run_sync(go)
a, = traversal.side_effects.keys()
assert a == 'a'
+ results = traversal.side_effects.get('a')
+ assert results
+ results = traversal.side_effects.close()
+ assert not results
+
+ connection.close()
if __name__ == '__main__':
[5/9] tinkerpop git commit: TINKERPOP-1490 Deprecated the old
RemoteConnection submit method.
Posted by da...@apache.org.
TINKERPOP-1490 Deprecated the old RemoteConnection submit method.
That method isn't called at all anymore - needs to be removed completely for 3.3.x.
Project: http://git-wip-us.apache.org/repos/asf/tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/tinkerpop/commit/57eb5ec6
Tree: http://git-wip-us.apache.org/repos/asf/tinkerpop/tree/57eb5ec6
Diff: http://git-wip-us.apache.org/repos/asf/tinkerpop/diff/57eb5ec6
Branch: refs/heads/TINKERPOP-1490
Commit: 57eb5ec6238a319a0c844bd2ddce87d735846c1d
Parents: 69e0f66
Author: Stephen Mallette <sp...@genoprime.com>
Authored: Mon Nov 14 10:08:48 2016 -0500
Committer: Stephen Mallette <sp...@genoprime.com>
Committed: Tue Nov 15 09:52:58 2016 -0500
----------------------------------------------------------------------
CHANGELOG.asciidoc | 1 +
.../gremlin/process/remote/RemoteConnection.java | 18 ++++++++++++++++--
.../process/remote/traversal/RemoteTraversal.java | 6 +++---
.../driver/remote/DriverRemoteConnection.java | 6 +++++-
4 files changed, 25 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/57eb5ec6/CHANGELOG.asciidoc
----------------------------------------------------------------------
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index 93e418c..6bfb8ad 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -39,6 +39,7 @@ TinkerPop 3.2.4 (Release Date: NOT OFFICIALLY RELEASED YET)
* 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()` method to allow for asynchronous traversal processing on "remote" traversals.
+* Deprecated `RemoteConnection.submit(Bytecode)` in favor of `submitAsync(Bytecode)`.
* 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/57eb5ec6/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/remote/RemoteConnection.java
----------------------------------------------------------------------
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 f4e3976..1831276 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
@@ -38,7 +38,7 @@ import java.util.concurrent.CompletableFuture;
public interface RemoteConnection extends AutoCloseable {
/**
- * @deprecated As of release 3.2.2, replaced by {@link #submit(Bytecode)}.
+ * @deprecated As of release 3.2.2, replaced by {@link #submitAsync(Bytecode)}.
*/
@Deprecated
public <E> Iterator<Traverser.Admin<E>> submit(final Traversal<?, E> traversal) throws RemoteConnectionException;
@@ -47,13 +47,27 @@ public interface RemoteConnection extends AutoCloseable {
* Submits {@link Traversal} {@link Bytecode} to a server and returns 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.
+ *
+ * @deprecated As of release 3.2.4, replaced by {@link #submitAsync(Bytecode)}.
*/
+ @Deprecated
public <E> RemoteTraversal<?,E> submit(final Bytecode bytecode) throws RemoteConnectionException;
/**
* 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.
+ * <p/>
+ * The default implementation calls the {@link #submit(Bytecode)} method for backward compatibility, but generally
+ * speaking this method should be implemented directly as {@link #submit(Bytecode)} is not called directly by
+ * any part of TinkerPop. Even if the {@code RemoteConnection} itself is not capable of asynchronous behaviour, it
+ * should simply implement this method in a blocking form.
*/
- public <E> CompletableFuture<RemoteTraversal<?, E>> submitAsync(final Bytecode bytecode) throws RemoteConnectionException;
+ public default <E> CompletableFuture<RemoteTraversal<?, E>> submitAsync(final Bytecode bytecode) throws RemoteConnectionException {
+ // default implementation for backward compatibility to 3.2.4 - this method will probably just become
+ // the new submit() in 3.3.x when the deprecation is removed
+ final CompletableFuture<RemoteTraversal<?, E>> promise = new CompletableFuture<>();
+ promise.complete(submit(bytecode));
+ return promise;
+ }
}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/57eb5ec6/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/remote/traversal/RemoteTraversal.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/remote/traversal/RemoteTraversal.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/remote/traversal/RemoteTraversal.java
index 57b0cda..70ec27f 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/remote/traversal/RemoteTraversal.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/remote/traversal/RemoteTraversal.java
@@ -25,9 +25,9 @@ import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.step.util.AbstractStep;
/**
- * A {@link RemoteTraversal} is returned from {@link RemoteConnection#submit(Bytecode)}. It is iterated from within
- * {@link RemoteStep} using {@link #nextTraverser()}. Implementations should typically be given a "result" from a
- * remote source where the traversal was executed. The "result" should be an iterator which preferably has its data
+ * A {@link RemoteTraversal} is returned from {@link RemoteConnection#submitAsync(Bytecode)}. It is iterated from
+ * within {@link RemoteStep} using {@link #nextTraverser()}. Implementations should typically be given a "result" from
+ * a remote source where the traversal was executed. The "result" should be an iterator which preferably has its data
* bulked.
* <p/>
* Note that internally {@link #nextTraverser()} is called from within a loop (specifically in
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/57eb5ec6/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/remote/DriverRemoteConnection.java
----------------------------------------------------------------------
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 be3fa28..d6415de 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
@@ -165,7 +165,7 @@ public class DriverRemoteConnection implements RemoteConnection {
}
/**
- * @deprecated As of release 3.2.2, replaced by {@link #submit(Bytecode)}.
+ * @deprecated As of release 3.2.2, replaced by {@link #submitAsync(Bytecode)}.
*/
@Deprecated
@Override
@@ -183,6 +183,10 @@ public class DriverRemoteConnection implements RemoteConnection {
}
}
+ /**
+ * @deprecated As of release 3.2.4, replaced by {@link #submitAsync(Bytecode)}.
+ */
+ @Deprecated
@Override
public <E> RemoteTraversal<?,E> submit(final Bytecode bytecode) throws RemoteConnectionException {
try {
[6/9] tinkerpop git commit: TINKERPOP-1490 Updated dev docs on
RemoteConnection API
Posted by da...@apache.org.
TINKERPOP-1490 Updated dev docs on RemoteConnection API
Project: http://git-wip-us.apache.org/repos/asf/tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/tinkerpop/commit/69e0f66d
Tree: http://git-wip-us.apache.org/repos/asf/tinkerpop/tree/69e0f66d
Diff: http://git-wip-us.apache.org/repos/asf/tinkerpop/diff/69e0f66d
Branch: refs/heads/TINKERPOP-1490
Commit: 69e0f66d53a55eca85adb8c94d244335e7324129
Parents: 08cfa80
Author: Stephen Mallette <sp...@genoprime.com>
Authored: Mon Nov 14 10:08:17 2016 -0500
Committer: Stephen Mallette <sp...@genoprime.com>
Committed: Tue Nov 15 09:52:58 2016 -0500
----------------------------------------------------------------------
docs/src/dev/provider/index.asciidoc | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/69e0f66d/docs/src/dev/provider/index.asciidoc
----------------------------------------------------------------------
diff --git a/docs/src/dev/provider/index.asciidoc b/docs/src/dev/provider/index.asciidoc
index 8760d01..71d071f 100644
--- a/docs/src/dev/provider/index.asciidoc
+++ b/docs/src/dev/provider/index.asciidoc
@@ -427,7 +427,7 @@ iterate.
There is one method to implement on `RemoteConnection`:
[source,java]
-public <E> RemoteTraversal<?,E> submit(final Bytecode bytecode) throws RemoteConnectionException;
+public <E> CompletableFuture<RemoteTraversal<?, E>> submitAsync(final Bytecode bytecode) throws RemoteConnectionException;
Note that it returns a `RemoteTraversal`. This interface should also be implemented and in most cases implementers can
simply extend the `AbstractRemoteTraversal`, which provides default implementations for all the `Traversal` methods
[3/9] tinkerpop git commit: TINKERPOP-1490 Remove some dead tests
from ignores.
Posted by da...@apache.org.
TINKERPOP-1490 Remove some dead tests from ignores.
Project: http://git-wip-us.apache.org/repos/asf/tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/tinkerpop/commit/bb0b22e5
Tree: http://git-wip-us.apache.org/repos/asf/tinkerpop/tree/bb0b22e5
Diff: http://git-wip-us.apache.org/repos/asf/tinkerpop/diff/bb0b22e5
Branch: refs/heads/TINKERPOP-1490
Commit: bb0b22e5a17a990ab2c785c3c40eaa1406c5f9bd
Parents: ca2ccf2
Author: Stephen Mallette <sp...@genoprime.com>
Authored: Fri Nov 11 13:39:40 2016 -0500
Committer: Stephen Mallette <sp...@genoprime.com>
Committed: Tue Nov 15 09:52:58 2016 -0500
----------------------------------------------------------------------
.../org/apache/tinkerpop/gremlin/python/jsr223/PythonProvider.java | 1 -
.../process/groovy/jsr223/TinkerGraphGroovyTranslatorProvider.java | 1 -
.../process/jsr223/TinkerGraphJavaTranslatorProvider.java | 1 -
3 files changed, 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/bb0b22e5/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 3bf4239..fe04156 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
@@ -72,7 +72,6 @@ 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/bb0b22e5/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 a595c34..dd118d7 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,7 +51,6 @@ 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/bb0b22e5/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 15a6c0b..f40a8c7 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,7 +49,6 @@ public class TinkerGraphJavaTranslatorProvider extends TinkerGraphProvider {
TraversalInterruptionComputerTest.class.getCanonicalName(),
"shouldNeverPropagateANoBulkTraverser",
"shouldNeverPropagateANullValuedTraverser",
- "shouldUsePromiseAndControlTransactionsIfAvailable",
ElementIdStrategyProcessTest.class.getCanonicalName(),
EventStrategyProcessTest.class.getCanonicalName(),
ProgramTest.Traversals.class.getCanonicalName()));
[8/9] tinkerpop git commit: rebased
Posted by da...@apache.org.
rebased
Project: http://git-wip-us.apache.org/repos/asf/tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/tinkerpop/commit/522960c3
Tree: http://git-wip-us.apache.org/repos/asf/tinkerpop/tree/522960c3
Diff: http://git-wip-us.apache.org/repos/asf/tinkerpop/diff/522960c3
Branch: refs/heads/TINKERPOP-1490
Commit: 522960c345c1d684151d3996607bd3778e1156b5
Parents: 54fac51 77b3416
Author: davebshow <da...@gmail.com>
Authored: Wed Dec 7 15:15:27 2016 -0500
Committer: davebshow <da...@gmail.com>
Committed: Wed Dec 7 15:15:27 2016 -0500
----------------------------------------------------------------------
CHANGELOG.asciidoc | 19 +
docker/build.sh | 2 +-
docker/scripts/build.sh | 7 +
.../developer/development-environment.asciidoc | 18 +-
docs/src/dev/io/graphson.asciidoc | 4 +-
.../src/reference/gremlin-applications.asciidoc | 6 +-
docs/src/reference/gremlin-variants.asciidoc | 5 -
.../reference/implementations-neo4j.asciidoc | 2 +-
.../gremlin-language-variants/index.asciidoc | 3 -
.../upgrade/release-3.2.x-incubating.asciidoc | 38 +-
.../src/test/resources/giraph-site.xml | 12 +
.../archetype-resources/src/main/java/App.java | 2 +-
.../src/main/java/Service.java | 2 +-
.../src/test/java/ServiceTest.java | 2 +-
.../SingleGremlinScriptEngineManager.java | 7 +
.../gremlin/process/traversal/Bindings.java | 39 +-
.../gremlin/process/traversal/Bytecode.java | 19 +-
.../process/traversal/TraversalEngine.java | 2 +-
.../process/traversal/TraversalSideEffects.java | 8 +-
.../process/traversal/TraversalSource.java | 7 +-
.../traversal/dsl/graph/GraphTraversal.java | 130 +++++--
.../dsl/graph/GraphTraversalSource.java | 4 +-
.../gremlin/process/traversal/dsl/graph/__.java | 43 ++-
.../traversal/step/map/GroupStepV3d0.java | 2 +-
.../step/map/LambdaCollectingBarrierStep.java | 2 +-
.../traversal/step/util/HasContainer.java | 26 +-
.../strategy/decoration/SubgraphStrategy.java | 4 +-
.../optimization/FilterRankingStrategy.java | 43 ++-
.../IncidentToAdjacentStrategy.java | 4 +-
.../optimization/InlineFilterStrategy.java | 20 +-
.../process/traversal/util/TraversalHelper.java | 19 +-
.../process/traversal/util/TraversalRing.java | 8 +-
.../gremlin/structure/io/IoRegistry.java | 2 +-
.../GraphSONTraversalSerializersV2d0.java | 9 +-
.../gremlin/structure/io/gryo/GryoPool.java | 23 +-
.../gremlin/structure/util/StringFactory.java | 19 +-
.../optimization/GraphFilterStrategyTest.java | 2 +-
.../gremlin/process/traversal/BytecodeTest.java | 32 +-
.../traversal/step/filter/HasStepTest.java | 12 +-
.../optimization/FilterRankingStrategyTest.java | 27 +-
.../optimization/InlineFilterStrategyTest.java | 12 +-
.../gremlin/structure/io/IoXIoRegistry.java | 2 +-
.../gremlin/structure/io/IoYIoRegistry.java | 2 +-
.../structure/io/gryo/GryoMapperTest.java | 12 +-
.../ser/AbstractGryoMessageSerializerV1d0.java | 4 +-
.../driver/ser/AbstractMessageSerializer.java | 26 +-
.../ser/GryoBaseMessageSerializerV1d0Test.java | 93 +++++
.../step/filter/GroovyWhereTest.groovy | 11 +
.../traversal/step/map/GroovyPathTest.groovy | 5 +
.../groovy/util/DependencyGrabber.groovy | 55 +--
.../util/DependencyGrabberIntegrateTest.java | 2 +-
.../python/GraphTraversalSourceGenerator.groovy | 15 +-
.../python/TraversalSourceGenerator.groovy | 6 +
.../gremlin_python/process/graph_traversal.py | 19 +-
.../jython/gremlin_python/process/traversal.py | 6 +
.../gremlin_python/structure/io/graphson.py | 10 +
.../driver/test_driver_remote_connection.py | 5 +
.../main/jython/tests/process/test_traversal.py | 26 ++
.../jython/tests/structure/io/test_graphson.py | 10 +
.../gremlin/process/ProcessComputerSuite.java | 6 +-
.../gremlin/process/ProcessStandardSuite.java | 4 +-
.../TraversalInterruptionComputerTest.java | 15 +-
.../process/traversal/step/ComplexTest.java | 205 +++++++++++
.../traversal/step/filter/WhereTest.java | 31 ++
.../process/traversal/step/map/PathTest.java | 35 +-
.../hadoop/structure/HadoopConfiguration.java | 4 +-
.../step/sideEffect/Neo4jGraphStep.java | 11 +-
.../Neo4jGraphStepStrategyTest.java | 12 +-
pom.xml | 2 +-
.../SparkGraphComputerProcessIntegrateTest.java | 32 ++
.../computer/SparkGraphComputerProcessTest.java | 32 --
...GraphComputerGroovyProcessIntegrateTest.java | 33 ++
.../SparkGraphComputerGroovyProcessTest.java | 33 --
.../PersistedInputOutputRDDIntegrateTest.java | 358 +++++++++++++++++++
.../io/PersistedInputOutputRDDTest.java | 358 -------------------
.../step/sideEffect/TinkerGraphStep.java | 9 +-
.../tinkergraph/structure/TinkerGraph.java | 2 +-
.../tinkergraph/structure/TinkerIoRegistry.java | 8 +
.../structure/TinkerIoRegistryV2d0.java | 8 +
.../TinkerGraphStepStrategyTest.java | 10 +-
.../TinkerGraphGraphSONSerializerV2d0Test.java | 4 +-
.../tinkergraph/structure/TinkerGraphTest.java | 2 +-
82 files changed, 1487 insertions(+), 683 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/522960c3/gremlin-python/src/main/jython/gremlin_python/process/traversal.py
----------------------------------------------------------------------
diff --cc gremlin-python/src/main/jython/gremlin_python/process/traversal.py
index 3fdd50c,07afeac..61f8aee
--- a/gremlin-python/src/main/jython/gremlin_python/process/traversal.py
+++ b/gremlin-python/src/main/jython/gremlin_python/process/traversal.py
@@@ -396,3 -400,10 +396,9 @@@ class Binding(object)
def __init__(self,key,value):
self.key = key
self.value = value
+ def __eq__(self, other):
+ return isinstance(other, self.__class__) and self.key == other.key and self.value == other.value
+ def __hash__(self):
+ return hash(self.key) + hash(self.value)
+ def __repr__(self):
+ return "binding[" + self.key + "=" + str(self.value) + "]"
-
[7/9] tinkerpop git commit: implemented promise API for
gremlin-python. side effect retrieval is problematic
Posted by da...@apache.org.
implemented promise API for gremlin-python. side effect retrieval is problematic
Project: http://git-wip-us.apache.org/repos/asf/tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/tinkerpop/commit/54fac514
Tree: http://git-wip-us.apache.org/repos/asf/tinkerpop/tree/54fac514
Diff: http://git-wip-us.apache.org/repos/asf/tinkerpop/diff/54fac514
Branch: refs/heads/TINKERPOP-1490
Commit: 54fac5146a5ac067e75f2fbfe135fae2da641c35
Parents: 57eb5ec
Author: davebshow <da...@gmail.com>
Authored: Wed Dec 7 13:35:42 2016 -0500
Committer: davebshow <da...@gmail.com>
Committed: Wed Dec 7 13:35:42 2016 -0500
----------------------------------------------------------------------
.../python/TraversalSourceGenerator.groovy | 27 +++++++++
.../driver/driver_remote_connection.py | 21 +++++--
.../gremlin_python/driver/remote_connection.py | 7 +++
.../jython/gremlin_python/process/traversal.py | 32 +++++++++--
.../driver/test_driver_remote_connection.py | 59 ++++++++++++++++++++
5 files changed, 137 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/54fac514/gremlin-python/src/main/groovy/org/apache/tinkerpop/gremlin/python/TraversalSourceGenerator.groovy
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/groovy/org/apache/tinkerpop/gremlin/python/TraversalSourceGenerator.groovy b/gremlin-python/src/main/groovy/org/apache/tinkerpop/gremlin/python/TraversalSourceGenerator.groovy
index a3684cb..debd61d 100644
--- a/gremlin-python/src/main/groovy/org/apache/tinkerpop/gremlin/python/TraversalSourceGenerator.groovy
+++ b/gremlin-python/src/main/groovy/org/apache/tinkerpop/gremlin/python/TraversalSourceGenerator.groovy
@@ -114,6 +114,28 @@ class Traversal(object):
except StopIteration: return tempList
tempList.append(temp)
return tempList
+ def promise(self, cb=None):
+ self.traversal_strategies.apply_async_strategies(self)
+ future_traversers = self.traversers
+ future = type(future_traversers)()
+ def process(f):
+ try:
+ traversers = f.result()
+ except Exception as e:
+ future.set_exception(e)
+ else:
+ self.traversers = iter(traversers)
+ if cb:
+ try:
+ result = cb(self)
+ except Exception as e:
+ future.set_exception(e)
+ else:
+ future.set_result(result)
+ else:
+ future.set_result(self)
+ future_traversers.add_done_callback(process)
+ return future
""")
@@ -223,6 +245,9 @@ class TraversalStrategies(object):
def apply_strategies(self, traversal):
for traversal_strategy in self.traversal_strategies:
traversal_strategy.apply(traversal)
+ def apply_async_strategies(self, traversal):
+ for traversal_strategy in self.traversal_strategies:
+ traversal_strategy.apply_async(traversal)
def __repr__(self):
return str(self.traversal_strategies)
@@ -233,6 +258,8 @@ class TraversalStrategy(object):
self.configuration = {} if configuration is None else configuration
def apply(self, traversal):
return
+ def apply_async(self, traversal):
+ return
def __eq__(self, other):
return isinstance(other, self.__class__)
def __hash__(self):
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/54fac514/gremlin-python/src/main/jython/gremlin_python/driver/driver_remote_connection.py
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/jython/gremlin_python/driver/driver_remote_connection.py b/gremlin-python/src/main/jython/gremlin_python/driver/driver_remote_connection.py
index eab06f1..bcf6d15 100644
--- a/gremlin-python/src/main/jython/gremlin_python/driver/driver_remote_connection.py
+++ b/gremlin-python/src/main/jython/gremlin_python/driver/driver_remote_connection.py
@@ -17,9 +17,11 @@ specific language governing permissions and limitations
under the License.
"""
import base64
+import functools
import json
import uuid
from tornado import gen
+from tornado import concurrent
from tornado import ioloop
from tornado import websocket
@@ -51,10 +53,15 @@ class DriverRemoteConnection(RemoteConnection):
'''
request_id = str(uuid.uuid4())
traversers = self._loop.run_sync(lambda: self.submit_traversal_bytecode(request_id, bytecode))
- side_effect_keys = lambda: self._loop.run_sync(lambda: self.submit_sideEffect_keys(request_id))
- side_effect_value = lambda key: self._loop.run_sync(lambda: self.submit_sideEffect_value(request_id, key))
- side_effect_close = lambda: self._loop.run_sync(lambda: self.submit_sideEffect_close(request_id))
- return RemoteTraversal(iter(traversers), RemoteTraversalSideEffects(side_effect_keys, side_effect_value, side_effect_close))
+ keys, value, close = self._get_side_effect_lambdas(request_id)
+ return RemoteTraversal(iter(traversers), RemoteTraversalSideEffects(keys, value, close))
+
+ def submit_async(self, bytecode):
+ request_id = str(uuid.uuid4())
+ future_traversers = self.submit_traversal_bytecode(request_id, bytecode)
+ keys, value, close = self._get_side_effect_lambdas(request_id)
+ side_effects = RemoteTraversalSideEffects(keys, value, close)
+ return RemoteTraversal(future_traversers, side_effects)
@gen.coroutine
def submit_traversal_bytecode(self, request_id, bytecode):
@@ -135,6 +142,12 @@ class DriverRemoteConnection(RemoteConnection):
result = yield self._execute_message(message)
raise gen.Return(result)
+ def _get_side_effect_lambdas(self, request_id):
+ side_effect_keys = lambda: self._loop.run_sync(lambda: self.submit_sideEffect_keys(request_id))
+ side_effect_value = lambda key: self._loop.run_sync(lambda: self.submit_sideEffect_value(request_id, key))
+ side_effect_close = lambda: self._loop.run_sync(lambda: self.submit_sideEffect_close(request_id))
+ return side_effect_keys, side_effect_value, side_effect_close
+
@gen.coroutine
def _execute_message(self, send_message):
send_message = b"".join([b"\x21",
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/54fac514/gremlin-python/src/main/jython/gremlin_python/driver/remote_connection.py
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/jython/gremlin_python/driver/remote_connection.py b/gremlin-python/src/main/jython/gremlin_python/driver/remote_connection.py
index 46fb760..93c92b7 100644
--- a/gremlin-python/src/main/jython/gremlin_python/driver/remote_connection.py
+++ b/gremlin-python/src/main/jython/gremlin_python/driver/remote_connection.py
@@ -95,3 +95,10 @@ class RemoteStrategy(TraversalStrategy):
remote_traversal = self.remote_connection.submit(traversal.bytecode)
traversal.side_effects = remote_traversal.side_effects
traversal.traversers = remote_traversal.traversers
+
+ def apply_async(self, traversal):
+ if traversal.traversers is None:
+ remote_traversal = self.remote_connection.submit_async(
+ traversal.bytecode)
+ traversal.side_effects = remote_traversal.side_effects
+ traversal.traversers = remote_traversal.traversers
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/54fac514/gremlin-python/src/main/jython/gremlin_python/process/traversal.py
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/jython/gremlin_python/process/traversal.py b/gremlin-python/src/main/jython/gremlin_python/process/traversal.py
index d30db35..3fdd50c 100644
--- a/gremlin-python/src/main/jython/gremlin_python/process/traversal.py
+++ b/gremlin-python/src/main/jython/gremlin_python/process/traversal.py
@@ -77,6 +77,28 @@ class Traversal(object):
except StopIteration: return tempList
tempList.append(temp)
return tempList
+ def promise(self, cb=None):
+ self.traversal_strategies.apply_async_strategies(self)
+ future_traversers = self.traversers
+ future = type(future_traversers)()
+ def process(f):
+ try:
+ traversers = f.result()
+ except Exception as e:
+ future.set_exception(e)
+ else:
+ self.traversers = iter(traversers)
+ if cb:
+ try:
+ result = cb(self)
+ except Exception as e:
+ future.set_exception(e)
+ else:
+ future.set_result(result)
+ else:
+ future.set_result(self)
+ future_traversers.add_done_callback(process)
+ return future
Barrier = Enum('Barrier', 'normSack')
@@ -118,10 +140,6 @@ statics.add_static('keyDecr', Order.keyDecr)
statics.add_static('valueDecr', Order.valueDecr)
statics.add_static('shuffle', Order.shuffle)
-Pick = Enum('Pick', 'any none')
-statics.add_static('any', Pick.any)
-statics.add_static('none', Pick.none)
-
Pop = Enum('Pop', 'all_ first last')
statics.add_static('first', Pop.first)
statics.add_static('last', Pop.last)
@@ -286,6 +304,9 @@ class TraversalStrategies(object):
def apply_strategies(self, traversal):
for traversal_strategy in self.traversal_strategies:
traversal_strategy.apply(traversal)
+ def apply_async_strategies(self, traversal):
+ for traversal_strategy in self.traversal_strategies:
+ traversal_strategy.apply_async(traversal)
def __repr__(self):
return str(self.traversal_strategies)
@@ -296,6 +317,8 @@ class TraversalStrategy(object):
self.configuration = {} if configuration is None else configuration
def apply(self, traversal):
return
+ def apply_async(self, traversal):
+ return
def __eq__(self, other):
return isinstance(other, self.__class__)
def __hash__(self):
@@ -373,4 +396,3 @@ class Binding(object):
def __init__(self,key,value):
self.key = key
self.value = value
-
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/54fac514/gremlin-python/src/main/jython/tests/driver/test_driver_remote_connection.py
----------------------------------------------------------------------
diff --git a/gremlin-python/src/main/jython/tests/driver/test_driver_remote_connection.py b/gremlin-python/src/main/jython/tests/driver/test_driver_remote_connection.py
index 9f3e466..d825dad 100644
--- a/gremlin-python/src/main/jython/tests/driver/test_driver_remote_connection.py
+++ b/gremlin-python/src/main/jython/tests/driver/test_driver_remote_connection.py
@@ -24,6 +24,8 @@ from unittest import TestCase
import pytest
+from tornado import ioloop, gen
+
from gremlin_python import statics
from gremlin_python.statics import long
from gremlin_python.driver.driver_remote_connection import DriverRemoteConnection
@@ -188,6 +190,63 @@ class TestDriverRemoteConnection(TestCase):
t.side_effects.value_lambda('b')
connection.close()
+ def test_promise(self):
+ loop = ioloop.IOLoop.current()
+ connection = DriverRemoteConnection('ws://localhost:45940/gremlin', 'g')
+ g = Graph().traversal().withRemote(connection)
+
+ @gen.coroutine
+ def go():
+ future_traversal = g.V().promise(lambda x: x.toList())
+ assert not future_traversal.done()
+ resp = yield future_traversal
+ assert future_traversal.done()
+ assert len(resp) == 6
+ count = yield g.V().count().promise(lambda x: x.next())
+ assert count == 6
+
+ loop.run_sync(go)
+
+ def test_promise_side_effects(self):
+ loop = ioloop.IOLoop.current()
+ connection = DriverRemoteConnection('ws://localhost:45940/gremlin', 'g')
+ g = Graph().traversal().withRemote(connection)
+
+ # Side effects are problematic in coroutines.
+ # Because they are designed to be synchronous (calling `run_sync`)
+ # they result in an error if called from a coroutine because
+ # the event loop is already running
+ @gen.coroutine
+ def go():
+ traversal = yield g.V().aggregate('a').promise()
+ # Trying to get side effect keys throws error - BAD
+ with pytest.raises(RuntimeError):
+ keys = traversal.side_effects.keys()
+ # IOLoop is now hosed.
+
+ loop.run_sync(go)
+
+ # Get a new IOLoop - this should happen for each test case.
+ connection.close()
+ ioloop.IOLoop.clear_instance()
+ loop.close()
+ loop = ioloop.IOLoop()
+ loop.make_current()
+
+ connection = DriverRemoteConnection('ws://localhost:45940/gremlin', 'g')
+ g = Graph().traversal().withRemote(connection)
+
+ # If we return the traversal though, we can use side effects per usual.
+ @gen.coroutine
+ def go():
+ traversal = yield g.V().aggregate('a').promise()
+ raise gen.Return(traversal) # Weird legacy Python compatible idiom
+
+ # See, normal side effects.
+ traversal = loop.run_sync(go)
+ a, = traversal.side_effects.keys()
+ assert a == 'a'
+
if __name__ == '__main__':
test = False