get stream transactions (#301)

Author: rashtao

docs/Drivers/Java/Reference/Database/StreamTransactions.md

@@ -29,6 +29,13 @@ Gets information about a Stream Transaction.
   transaction id
 
 
+## ArangoDatabase.getStreamTransactions
+
+`ArangoDatabase.getStreamTransactions() : Collection<TransactionEntity>`
+
+Gets all the currently running Stream Transactions.
+
+
 ## ArangoDatabase.commitStreamTransaction
 
 `ArangoDatabase.commitStreamTransaction(String id) : StreamTransactionEntity`

src/main/java/com/arangodb/ArangoDatabase.java

@@ -577,6 +577,17 @@ GraphEntity createGraph(String name, Collection<EdgeDefinition> edgeDefinitions,
 	 */
 	StreamTransactionEntity getStreamTransaction(String id) throws ArangoDBException;
 
+	/**
+	 * Gets all the currently running Stream Transactions.
+	 *
+	 * @return all the currently running Stream Transactions
+	 * @throws ArangoDBException
+	 * @see <a href="https://docs.arangodb.com/current/HTTP/transaction-stream-transaction.html#list-currently-ongoing-transactions">
+	 * API Documentation</a>
+	 * @since ArangoDB 3.5.0
+	 */
+	Collection<TransactionEntity> getStreamTransactions() throws ArangoDBException;
+
 	/**
 	 * Commits a Stream Transaction.
 	 *

src/main/java/com/arangodb/entity/StreamTransactionEntity.java

@@ -31,10 +31,6 @@ public class StreamTransactionEntity implements Entity {
 	private String id;
 	private StreamTransactionStatus status;
 
-	public enum StreamTransactionStatus {
-		running, committed, aborted
-	}
-
 	public String getId() {
 		return id;
 	}

src/main/java/com/arangodb/entity/StreamTransactionStatus.java

@@ -0,0 +1,28 @@
+/*
+ * DISCLAIMER
+ *
+ * Copyright 2016 ArangoDB GmbH, Cologne, Germany
+ *
+ * Licensed 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.
+ *
+ * Copyright holder is ArangoDB GmbH, Cologne, Germany
+ */
+
+package com.arangodb.entity;
+
+/**
+ * @author Michele Rastelli
+ */
+public enum StreamTransactionStatus {
+	running, committed, aborted
+}

src/main/java/com/arangodb/entity/TransactionEntity.java

@@ -0,0 +1,42 @@
+/*
+ * DISCLAIMER
+ *
+ * Copyright 2016 ArangoDB GmbH, Cologne, Germany
+ *
+ * Licensed 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.
+ *
+ * Copyright holder is ArangoDB GmbH, Cologne, Germany
+ */
+
+package com.arangodb.entity;
+
+/**
+ * @author Michele Rastelli
+ * @see <a href=
+ * "https://docs.arangodb.com/current/HTTP/transaction-stream-transaction.html#list-currently-ongoing-transactions</a>
+ * @since ArangoDB 3.5.0
+ */
+public class TransactionEntity implements Entity {
+
+	private String id;
+	private StreamTransactionStatus state;
+
+	public String getId() {
+		return id;
+	}
+
+	public StreamTransactionStatus getStatus() {
+		return state;
+	}
+
+}

src/main/java/com/arangodb/internal/ArangoDatabaseImpl.java

@@ -351,6 +351,11 @@ public StreamTransactionEntity getStreamTransaction(String id) throws ArangoDBEx
 		return executor.execute(getStreamTransactionRequest(id), streamTransactionResponseDeserializer());
 	}
 
+	@Override
+	public Collection<TransactionEntity> getStreamTransactions() throws ArangoDBException {
+		return executor.execute(getStreamTransactionsRequest(), transactionsResponseDeserializer());
+	}
+
 	@Override
 	public StreamTransactionEntity commitStreamTransaction(String id) throws ArangoDBException {
 		return executor.execute(commitStreamTransactionRequest(id), streamTransactionResponseDeserializer());

src/main/java/com/arangodb/internal/InternalArangoDatabase.java

@@ -387,10 +387,25 @@ protected Request abortStreamTransactionRequest(String id) {
 		return request(name, RequestType.DELETE, PATH_API_TRANSACTION, id);
 	}
 
+	protected Request getStreamTransactionsRequest() {
+		return request(name, RequestType.GET, PATH_API_TRANSACTION);
+	}
+
 	protected Request getStreamTransactionRequest(String id) {
 		return request(name, RequestType.GET, PATH_API_TRANSACTION, id);
 	}
 
+	protected ResponseDeserializer<Collection<TransactionEntity>> transactionsResponseDeserializer() {
+		return new ResponseDeserializer<Collection<TransactionEntity>>() {
+			@Override
+			public Collection<TransactionEntity> deserialize(final Response response) throws VPackException {
+				final VPackSlice result = response.getBody().get("transactions");
+				return util().deserialize(result, new Type<Collection<TransactionEntity>>() {
+				}.getType());
+			}
+		};
+	}
+
 	protected Request commitStreamTransactionRequest(String id) {
 		return request(name, RequestType.PUT, PATH_API_TRANSACTION, id);
 	}

src/test/java/com/arangodb/StreamTransactionTest.java

@@ -73,7 +73,7 @@ public void beginStreamTransaction() {
 
 		StreamTransactionEntity tx = db.beginStreamTransaction(null);
 		assertThat(tx.getId(), is(notNullValue()));
-		assertThat(tx.getStatus(), is(StreamTransactionEntity.StreamTransactionStatus.running));
+		assertThat(tx.getStatus(), is(StreamTransactionStatus.running));
 		db.abortStreamTransaction(tx.getId());
 	}
 
@@ -97,7 +97,7 @@ public void abortStreamTransaction() {
 
 		assertThat(abortedTx.getId(), is(notNullValue()));
 		assertThat(abortedTx.getId(), is(begunTx.getId()));
-		assertThat(abortedTx.getStatus(), is(StreamTransactionEntity.StreamTransactionStatus.aborted));
+		assertThat(abortedTx.getStatus(), is(StreamTransactionStatus.aborted));
 	}
 
 	@Test
@@ -151,7 +151,7 @@ public void getStreamTransaction() {
 
 		assertThat(gotTx.getId(), is(notNullValue()));
 		assertThat(gotTx.getId(), is(createdTx.getId()));
-		assertThat(gotTx.getStatus(), is(StreamTransactionEntity.StreamTransactionStatus.running));
+		assertThat(gotTx.getStatus(), is(StreamTransactionStatus.running));
 
 		db.abortStreamTransaction(createdTx.getId());
 	}
@@ -185,7 +185,7 @@ public void commitStreamTransaction() {
 
 		assertThat(committedTx.getId(), is(notNullValue()));
 		assertThat(committedTx.getId(), is(createdTx.getId()));
-		assertThat(committedTx.getStatus(), is(StreamTransactionEntity.StreamTransactionStatus.committed));
+		assertThat(committedTx.getStatus(), is(StreamTransactionStatus.committed));
 	}
 
 	@Test
@@ -705,4 +705,25 @@ public void nextCursor() {
 		db.abortStreamTransaction(tx.getId());
 	}
 
+	@Test
+	public void getStreamTransactions() {
+		assumeTrue(requireSingleServer());
+		assumeTrue(requireVersion(3, 5));
+		assumeTrue(requireStorageEngine(ArangoDBEngine.StorageEngineName.rocksdb));
+
+		StreamTransactionEntity tx1 = db.beginStreamTransaction(null);
+		StreamTransactionEntity tx2 = db.beginStreamTransaction(null);
+
+		List<String> createdIds = Arrays.asList(tx1.getId(), tx2.getId());
+		Set<TransactionEntity> gotTxs = db.getStreamTransactions().stream().
+				filter(it -> createdIds.contains(it.getId())).collect(Collectors.toSet());
+
+		assertThat(gotTxs.size(), is(createdIds.size()));
+		assertThat(gotTxs.stream()
+				.allMatch(it -> it.getStatus() == StreamTransactionStatus.running), is(true));
+
+		db.abortStreamTransaction(tx1.getId());
+		db.abortStreamTransaction(tx2.getId());
+	}
+
 }