You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by sp...@apache.org on 2015/08/25 14:34:50 UTC
[31/50] incubator-tinkerpop git commit: Decomposed
AbstractTransaction around listener implementations.
Decomposed AbstractTransaction around listener implementations.
There are now two sub-classes of AbstractTransaction: one for thread local and one for concurrent use. This should be helpful to those implementing threaded transactions as the Graph returned from that is not ThreadLocal based (which was how AbstractTransaction was implemented).
Project: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/commit/3c9356ed
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/tree/3c9356ed
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/diff/3c9356ed
Branch: refs/heads/tp30
Commit: 3c9356edbab2dbc98640b0ccf675cb8a7ecd4c5d
Parents: 54b114b
Author: Stephen Mallette <sp...@genoprime.com>
Authored: Tue Aug 18 08:41:38 2015 -0400
Committer: Stephen Mallette <sp...@genoprime.com>
Committed: Tue Aug 18 08:41:38 2015 -0400
----------------------------------------------------------------------
.../gremlin/structure/Transaction.java | 1 -
.../util/AbstractThreadLocalTransaction.java | 75 +++++++++++++++++++
.../util/AbstractThreadedTransaction.java | 70 +++++++++++++++++
.../structure/util/AbstractTransaction.java | 79 +++++++++++---------
.../gremlin/neo4j/structure/Neo4jGraph.java | 8 +-
5 files changed, 192 insertions(+), 41 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/3c9356ed/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/Transaction.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/Transaction.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/Transaction.java
index 56b22e0..03bd8cc 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/Transaction.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/Transaction.java
@@ -20,7 +20,6 @@ package org.apache.tinkerpop.gremlin.structure;
import org.apache.tinkerpop.gremlin.structure.util.AbstractTransaction;
-import java.io.Closeable;
import java.util.Collections;
import java.util.Set;
import java.util.function.BiFunction;
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/3c9356ed/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/AbstractThreadLocalTransaction.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/AbstractThreadLocalTransaction.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/AbstractThreadLocalTransaction.java
new file mode 100644
index 0000000..b47eb79
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/AbstractThreadLocalTransaction.java
@@ -0,0 +1,75 @@
+/*
+ * 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.structure.util;
+
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.Transaction;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Consumer;
+
+/**
+ * A base implementation of {@link Transaction} that provides core functionality for transaction listeners using
+ * {@link ThreadLocal}. In this implementation, the listeners are bound to the current thread of execution (usually
+ * the same as the transaction for most graph database implementations). Therefore, when {@link #commit()} is
+ * called on a particular thread, the only listeners that get notified are those bound to that thread.
+ *
+ * @see AbstractThreadedTransaction
+ *
+ * @author Stephen Mallette (http://stephen.genoprime.com)
+ */
+public abstract class AbstractThreadLocalTransaction extends AbstractTransaction {
+
+ protected final ThreadLocal<List<Consumer<Transaction.Status>>> transactionListeners = new ThreadLocal<List<Consumer<Transaction.Status>>>() {
+ @Override
+ protected List<Consumer<Transaction.Status>> initialValue() {
+ return new ArrayList<>();
+ }
+ };
+
+ public AbstractThreadLocalTransaction(final Graph g) {
+ super(g);
+ }
+
+ @Override
+ protected void fireOnCommit() {
+ transactionListeners.get().forEach(c -> c.accept(Status.COMMIT));
+ }
+
+ @Override
+ protected void fireOnRollback() {
+ transactionListeners.get().forEach(c -> c.accept(Status.ROLLBACK));
+ }
+
+ @Override
+ public void addTransactionListener(final Consumer<Status> listener) {
+ transactionListeners.get().add(listener);
+ }
+
+ @Override
+ public void removeTransactionListener(final Consumer<Status> listener) {
+ transactionListeners.get().remove(listener);
+ }
+
+ @Override
+ public void clearTransactionListeners() {
+ transactionListeners.get().clear();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/3c9356ed/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/AbstractThreadedTransaction.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/AbstractThreadedTransaction.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/AbstractThreadedTransaction.java
new file mode 100644
index 0000000..57f8ec0
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/AbstractThreadedTransaction.java
@@ -0,0 +1,70 @@
+/*
+ * 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.structure.util;
+
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.Transaction;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.function.Consumer;
+
+/**
+ * A base implementation of {@link Transaction} that provides core functionality for transaction listeners using a
+ * shared set of transaction listeners. Therefore, when {@link #commit()} is called from any thread, all listeners
+ * get notified. This implementation would be useful for graph implementations that support threaded transactions,
+ * specifically in the {@link Graph} instance returned from {@link Transaction#createThreadedTx()}.
+ *
+ * @see AbstractThreadLocalTransaction
+ *
+ * @author Stephen Mallette (http://stephen.genoprime.com)
+ */
+public abstract class AbstractThreadedTransaction extends AbstractTransaction {
+
+ protected final List<Consumer<Status>> transactionListeners = new CopyOnWriteArrayList<>();
+
+ public AbstractThreadedTransaction(final Graph g) {
+ super(g);
+ }
+
+ @Override
+ protected void fireOnCommit() {
+ transactionListeners.forEach(c -> c.accept(Status.COMMIT));
+ }
+
+ @Override
+ protected void fireOnRollback() {
+ transactionListeners.forEach(c -> c.accept(Status.ROLLBACK));
+ }
+
+ @Override
+ public void addTransactionListener(final Consumer<Status> listener) {
+ transactionListeners.add(listener);
+ }
+
+ @Override
+ public void removeTransactionListener(final Consumer<Status> listener) {
+ transactionListeners.remove(listener);
+ }
+
+ @Override
+ public void clearTransactionListeners() {
+ transactionListeners.clear();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/3c9356ed/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/AbstractTransaction.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/AbstractTransaction.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/AbstractTransaction.java
index 26a7d78..56c6f36 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/AbstractTransaction.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/AbstractTransaction.java
@@ -28,13 +28,11 @@ import java.util.function.Consumer;
import java.util.function.Function;
/**
- * A simple base class for {@link Transaction} that provides some common functionality and default behavior. Vendors
- * can use this class as a starting point for their own implementations. Implementers should note that this
- * class assumes that threaded transactions are not enabled. Vendors should explicitly override
- * {@link #createThreadedTx} to implement that functionality if required.
- * <p/>
- * Note that transaction listeners are registered in a {@link ThreadLocal} fashion which matches the pattern
- * expected of vendor implementations of a {@link Transaction}.
+ * A simple base class for {@link Transaction} that provides some common functionality and default behavior.
+ * While vendors can choose to use this class, it is generally better to extend from
+ * {@link AbstractThreadedTransaction} or {@link AbstractThreadLocalTransaction} which include default "listener"
+ * functionality. Implementers should note that this class assumes that threaded transactions are not enabled
+ * and should explicitly override {@link #createThreadedTx} to implement that functionality if required.
*
* @author Stephen Mallette (http://stephen.genoprime.com)
*/
@@ -44,13 +42,6 @@ public abstract class AbstractTransaction implements Transaction {
private Graph g;
- private ThreadLocal<List<Consumer<Status>>> transactionListeners = new ThreadLocal<List<Consumer<Status>>>() {
- @Override
- protected List<Consumer<Status>> initialValue() {
- return new ArrayList<>();
- }
- };
-
public AbstractTransaction(final Graph g) {
// auto transaction behavior
readWriteConsumer = READ_WRITE_BEHAVIOR.AUTO;
@@ -79,6 +70,23 @@ public abstract class AbstractTransaction implements Transaction {
*/
protected abstract void doRollback() throws TransactionException;
+ /**
+ * Called within {@link #commit()} just after the internal call to {@link #doCommit()}. Implementations of this
+ * method should raise {@link Status#COMMIT} events to any listeners added via
+ * {@link #addTransactionListener(Consumer)}.
+ */
+ protected abstract void fireOnCommit();
+
+ /**
+ * Called within {@link #rollback()} just after the internal call to {@link #doRollback()} ()}. Implementations
+ * of this method should raise {@link Status#ROLLBACK} events to any listeners added via
+ * {@link #addTransactionListener(Consumer)}.
+ */
+ protected abstract void fireOnRollback();
+
+ /**
+ * {@inheritDoc}
+ */
@Override
public void open() {
if (isOpen())
@@ -87,40 +95,52 @@ public abstract class AbstractTransaction implements Transaction {
doOpen();
}
+ /**
+ * {@inheritDoc}
+ */
@Override
public void commit() {
readWriteConsumer.accept(this);
try {
doCommit();
+ fireOnCommit();
} catch (TransactionException te) {
throw new RuntimeException(te);
}
-
- transactionListeners.get().forEach(c -> c.accept(Status.COMMIT));
}
+ /**
+ * {@inheritDoc}
+ */
@Override
public void rollback() {
readWriteConsumer.accept(this);
try {
doRollback();
+ fireOnRollback();
} catch (TransactionException te) {
throw new RuntimeException(te);
}
-
- transactionListeners.get().forEach(c -> c.accept(Status.ROLLBACK));
}
+ /**
+ * {@inheritDoc}
+ */
@Override
public <R> Workload<R> submit(final Function<Graph, R> work) {
return new Workload<>(g, work);
}
-
+ /**
+ * {@inheritDoc}
+ */
@Override
public <G extends Graph> G createThreadedTx() {
throw Transaction.Exceptions.threadedTransactionsNotSupported();
}
+ /**
+ * {@inheritDoc}
+ */
@Override
public void readWrite() {
readWriteConsumer.accept(this);
@@ -134,33 +154,24 @@ public abstract class AbstractTransaction implements Transaction {
closeConsumer.accept(this);
}
+ /**
+ * {@inheritDoc}
+ */
@Override
public synchronized Transaction onReadWrite(final Consumer<Transaction> consumer) {
readWriteConsumer = Optional.ofNullable(consumer).orElseThrow(Transaction.Exceptions::onReadWriteBehaviorCannotBeNull);
return this;
}
+ /**
+ * {@inheritDoc}
+ */
@Override
public synchronized Transaction onClose(final Consumer<Transaction> consumer) {
closeConsumer = Optional.ofNullable(consumer).orElseThrow(Transaction.Exceptions::onCloseBehaviorCannotBeNull);
return this;
}
- @Override
- public void addTransactionListener(final Consumer<Status> listener) {
- transactionListeners.get().add(listener);
- }
-
- @Override
- public void removeTransactionListener(final Consumer<Status> listener) {
- transactionListeners.get().remove(listener);
- }
-
- @Override
- public void clearTransactionListeners() {
- transactionListeners.get().clear();
- }
-
/**
* An "internal" exception thrown by vendors when calls to {@link AbstractTransaction#doCommit} or
* {@link AbstractTransaction#doRollback} fail.
http://git-wip-us.apache.org/repos/asf/incubator-tinkerpop/blob/3c9356ed/neo4j-gremlin/src/main/java/org/apache/tinkerpop/gremlin/neo4j/structure/Neo4jGraph.java
----------------------------------------------------------------------
diff --git a/neo4j-gremlin/src/main/java/org/apache/tinkerpop/gremlin/neo4j/structure/Neo4jGraph.java b/neo4j-gremlin/src/main/java/org/apache/tinkerpop/gremlin/neo4j/structure/Neo4jGraph.java
index 17b9976..2ea254c 100644
--- a/neo4j-gremlin/src/main/java/org/apache/tinkerpop/gremlin/neo4j/structure/Neo4jGraph.java
+++ b/neo4j-gremlin/src/main/java/org/apache/tinkerpop/gremlin/neo4j/structure/Neo4jGraph.java
@@ -36,7 +36,7 @@ import org.apache.tinkerpop.gremlin.structure.Graph;
import org.apache.tinkerpop.gremlin.structure.Transaction;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.VertexProperty;
-import org.apache.tinkerpop.gremlin.structure.util.AbstractTransaction;
+import org.apache.tinkerpop.gremlin.structure.util.AbstractThreadLocalTransaction;
import org.apache.tinkerpop.gremlin.structure.util.ElementHelper;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import org.apache.tinkerpop.gremlin.structure.util.wrapped.WrappedGraph;
@@ -316,7 +316,7 @@ public final class Neo4jGraph implements Graph, WrappedGraph<Neo4jGraphAPI> {
return traversal;
}
- class Neo4jTransaction extends AbstractTransaction {
+ class Neo4jTransaction extends AbstractThreadLocalTransaction {
protected final ThreadLocal<Neo4jTx> threadLocalTx = ThreadLocal.withInitial(() -> null);
@@ -344,10 +344,6 @@ public final class Neo4jGraph implements Graph, WrappedGraph<Neo4jGraphAPI> {
@Override
public void doRollback() throws TransactionException {
try {
-// javax.transaction.Transaction t = transactionManager.getTransaction();
-// if (null == t || t.getStatus() == javax.transaction.Status.STATUS_ROLLEDBACK)
-// return;
-
threadLocalTx.get().failure();
} catch (Exception e) {
throw new TransactionException(e);