Add dirty read support

ChangeLog.md

@@ -6,6 +6,16 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) a
 
 ## [Unreleased]
 
+### Added
+
+- added dirty read support ([reading from followers](https://docs.arangodb.com/current/Manual/Administration/ActiveFailover/#reading-from-follower))
+
+  - added option `AqlQueryOptions#allowDirtyRead` for `ArangoDatabase#query`.
+  - added option `DocumentReadOptions#allowDirtyRead` for `ArangoCollection#getDocument`
+  - added option `DocumentReadOptions#allowDirtyRead` for `ArangoCollection#getDocuments`
+  - added option `DocumentReadOptions#allowDirtyRead` for `ArangoVertexCollection#getVertex`
+  - added option `DocumentReadOptions#allowDirtyRead` for `ArangoEdgeCollection#getEdge`
+
 ### Changed
 
 - changed the internal connection pool and host management. There now exists a connection pool for every configured host. This changes the behavior of `ArangoDB.Builder#maxConnections` which now allows to configure the maximal number of connection per host and not overall.

src/main/java/com/arangodb/ArangoCollection.java

@@ -214,6 +214,21 @@ <T> MultiDocumentEntity<DocumentCreateEntity<T>> insertDocuments(
 	 */
 	<T> MultiDocumentEntity<T> getDocuments(Collection<String> keys, Class<T> type) throws ArangoDBException;
 
+	/**
+	 * Retrieves multiple documents with the given {@code _key} from the collection.
+	 * 
+	 * @param keys
+	 *            The keys of the documents
+	 * @param type
+	 *            The type of the documents (POJO class, VPackSlice or String for Json)
+	 * @param options
+	 *            Additional options, can be null
+	 * @return the documents and possible errors
+	 * @throws ArangoDBException
+	 */
+	<T> MultiDocumentEntity<T> getDocuments(Collection<String> keys, Class<T> type, DocumentReadOptions options)
+			throws ArangoDBException;
+
 	/**
 	 * Replaces the document with {@code key} with the one in the body, provided there is such a document and no
 	 * precondition is violated

src/main/java/com/arangodb/ArangoDB.java

@@ -41,6 +41,7 @@
 import com.arangodb.internal.http.HttpCommunication;
 import com.arangodb.internal.http.HttpConnectionFactory;
 import com.arangodb.internal.net.ConnectionFactory;
+import com.arangodb.internal.net.Host;
 import com.arangodb.internal.net.HostHandle;
 import com.arangodb.internal.net.HostHandler;
 import com.arangodb.internal.net.HostResolver;
@@ -620,8 +621,8 @@ public synchronized ArangoDB build() {
 					: new HttpConnectionFactory(timeout, user, password, useSsl, sslContext, custom, protocol,
 							connectionTtl);
 
-			final HostResolver hostResolver = createHostResolver(createHostList(max, connectionFactory), max,
-				connectionFactory);
+			final Collection<Host> hostList = createHostList(max, connectionFactory);
+			final HostResolver hostResolver = createHostResolver(hostList, max, connectionFactory);
 			final HostHandler hostHandler = createHostHandler(hostResolver);
 			return new ArangoDBImpl(
 					new VstCommunicationSync.Builder(hostHandler).timeout(timeout).user(user).password(password)

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

@@ -148,7 +148,14 @@ public <T> T getDocument(final String key, final Class<T> type, final DocumentRe
 	@Override
 	public <T> MultiDocumentEntity<T> getDocuments(final Collection<String> keys, final Class<T> type)
 			throws ArangoDBException {
-		final DocumentReadOptions options = new DocumentReadOptions();
+		return getDocuments(keys, type, new DocumentReadOptions());
+	}
+
+	@Override
+	public <T> MultiDocumentEntity<T> getDocuments(
+		final Collection<String> keys,
+		final Class<T> type,
+		final DocumentReadOptions options) throws ArangoDBException {
 		return executor.execute(getDocumentsRequest(keys, options), getDocumentsResponseDeserializer(type, options));
 	}
 

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

@@ -38,6 +38,7 @@
 import com.arangodb.internal.ArangoExecutor.ResponseDeserializer;
 import com.arangodb.internal.util.ArangoSerializationFactory.Serializer;
 import com.arangodb.internal.util.DocumentUtil;
+import com.arangodb.internal.util.RequestUtils;
 import com.arangodb.model.CollectionPropertiesOptions;
 import com.arangodb.model.CollectionRenameOptions;
 import com.arangodb.model.DocumentCreateOptions;
@@ -221,13 +222,22 @@ protected Request getDocumentRequest(final String key, final DocumentReadOptions
 		final DocumentReadOptions params = (options != null ? options : new DocumentReadOptions());
 		request.putHeaderParam(ArangoRequestParam.IF_NONE_MATCH, params.getIfNoneMatch());
 		request.putHeaderParam(ArangoRequestParam.IF_MATCH, params.getIfMatch());
+		if (params.getAllowDirtyRead() == Boolean.TRUE) {
+			RequestUtils.allowDirtyRead(request);
+		}
 		return request;
 	}
 
 	protected Request getDocumentsRequest(final Collection<String> keys, final DocumentReadOptions options) {
-		return request(db.name(), RequestType.PUT, PATH_API_DOCUMENT, name).putQueryParam("onlyget", true)
-				.putHeaderParam(ArangoRequestParam.IF_NONE_MATCH, options.getIfNoneMatch())
-				.putHeaderParam(ArangoRequestParam.IF_MATCH, options.getIfMatch()).setBody(util().serialize(keys));
+		final DocumentReadOptions params = (options != null ? options : new DocumentReadOptions());
+		final Request request = request(db.name(), RequestType.PUT, PATH_API_DOCUMENT, name)
+				.putQueryParam("onlyget", true)
+				.putHeaderParam(ArangoRequestParam.IF_NONE_MATCH, params.getIfNoneMatch())
+				.putHeaderParam(ArangoRequestParam.IF_MATCH, params.getIfMatch()).setBody(util().serialize(keys));
+		if (params.getAllowDirtyRead() == Boolean.TRUE) {
+			RequestUtils.allowDirtyRead(request);
+		}
+		return request;
 	}
 
 	protected <T> ResponseDeserializer<MultiDocumentEntity<T>> getDocumentsResponseDeserializer(

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

@@ -34,6 +34,7 @@
 import com.arangodb.entity.LoadBalancingStrategy;
 import com.arangodb.internal.net.Connection;
 import com.arangodb.internal.net.ConnectionFactory;
+import com.arangodb.internal.net.DirtyReadHostHandler;
 import com.arangodb.internal.net.ExtendedHostResolver;
 import com.arangodb.internal.net.FallbackHostHandler;
 import com.arangodb.internal.net.Host;
@@ -213,7 +214,7 @@ protected HostHandler createHostHandler(final HostResolver hostResolver) {
 		} else {
 			hostHandler = new FallbackHostHandler(hostResolver);
 		}
-		return hostHandler;
+		return new DirtyReadHostHandler(hostHandler, new RoundRobinHostHandler(hostResolver));
 	}
 
 	private static void loadHosts(final Properties properties, final Collection<HostDescription> hosts) {

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

@@ -39,6 +39,7 @@
 import com.arangodb.entity.ViewType;
 import com.arangodb.internal.ArangoExecutor.ResponseDeserializer;
 import com.arangodb.internal.util.ArangoSerializationFactory.Serializer;
+import com.arangodb.internal.util.RequestUtils;
 import com.arangodb.model.AqlFunctionCreateOptions;
 import com.arangodb.model.AqlFunctionDeleteOptions;
 import com.arangodb.model.AqlFunctionGetOptions;
@@ -192,9 +193,14 @@ protected Request queryRequest(
 		final String query,
 		final Map<String, Object> bindVars,
 		final AqlQueryOptions options) {
-		return request(name, RequestType.POST, PATH_API_CURSOR).setBody(util().serialize(
-			OptionsBuilder.build(options != null ? options : new AqlQueryOptions(), query, bindVars != null
+		final AqlQueryOptions opt = options != null ? options : new AqlQueryOptions();
+		final Request request = request(name, RequestType.POST, PATH_API_CURSOR).setBody(
+			util().serialize(OptionsBuilder.build(opt, query, bindVars != null
 					? util().serialize(bindVars, new ArangoSerializer.Options().serializeNullValues(true)) : null)));
+		if (opt.getAllowDirtyRead() == Boolean.TRUE) {
+			RequestUtils.allowDirtyRead(request);
+		}
+		return request;
 	}
 
 	protected Request queryNextRequest(final String id) {

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

@@ -29,6 +29,7 @@
 import com.arangodb.internal.ArangoExecutor.ResponseDeserializer;
 import com.arangodb.internal.util.ArangoSerializationFactory.Serializer;
 import com.arangodb.internal.util.DocumentUtil;
+import com.arangodb.internal.util.RequestUtils;
 import com.arangodb.model.DocumentReadOptions;
 import com.arangodb.model.EdgeCreateOptions;
 import com.arangodb.model.EdgeDeleteOptions;
@@ -99,6 +100,9 @@ protected Request getEdgeRequest(final String key, final DocumentReadOptions opt
 		final DocumentReadOptions params = (options != null ? options : new DocumentReadOptions());
 		request.putHeaderParam(ArangoRequestParam.IF_NONE_MATCH, params.getIfNoneMatch());
 		request.putHeaderParam(ArangoRequestParam.IF_MATCH, params.getIfMatch());
+		if (params.getAllowDirtyRead()) {
+			RequestUtils.allowDirtyRead(request);
+		}
 		return request;
 	}
 

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

@@ -29,6 +29,7 @@
 import com.arangodb.internal.ArangoExecutor.ResponseDeserializer;
 import com.arangodb.internal.util.ArangoSerializationFactory.Serializer;
 import com.arangodb.internal.util.DocumentUtil;
+import com.arangodb.internal.util.RequestUtils;
 import com.arangodb.model.DocumentReadOptions;
 import com.arangodb.model.VertexCreateOptions;
 import com.arangodb.model.VertexDeleteOptions;
@@ -103,6 +104,9 @@ protected Request getVertexRequest(final String key, final DocumentReadOptions o
 		final DocumentReadOptions params = (options != null ? options : new DocumentReadOptions());
 		request.putHeaderParam(ArangoRequestParam.IF_NONE_MATCH, params.getIfNoneMatch());
 		request.putHeaderParam(ArangoRequestParam.IF_MATCH, params.getIfMatch());
+		if (params.getAllowDirtyRead()) {
+			RequestUtils.allowDirtyRead(request);
+		}
 		return request;
 	}
 

src/main/java/com/arangodb/internal/http/HttpCommunication.java

@@ -28,12 +28,14 @@
 import org.slf4j.LoggerFactory;
 
 import com.arangodb.ArangoDBException;
+import com.arangodb.internal.net.AccessType;
 import com.arangodb.internal.net.ArangoDBRedirectException;
 import com.arangodb.internal.net.Host;
 import com.arangodb.internal.net.HostDescription;
 import com.arangodb.internal.net.HostHandle;
 import com.arangodb.internal.net.HostHandler;
 import com.arangodb.internal.util.HostUtils;
+import com.arangodb.internal.util.RequestUtils;
 import com.arangodb.util.ArangoSerialization;
 import com.arangodb.velocystream.Request;
 import com.arangodb.velocystream.Response;
@@ -77,7 +79,8 @@ public void close() throws IOException {
 	}
 
 	public Response execute(final Request request, final HostHandle hostHandle) throws ArangoDBException, IOException {
-		Host host = hostHandler.get(hostHandle);
+		final AccessType accessType = RequestUtils.determineAccessType(request);
+		Host host = hostHandler.get(hostHandle, accessType);
 		try {
 			while (true) {
 				try {
@@ -89,7 +92,7 @@ public Response execute(final Request request, final HostHandle hostHandle) thro
 				} catch (final SocketException se) {
 					hostHandler.fail();
 					final Host failedHost = host;
-					host = hostHandler.get(hostHandle);
+					host = hostHandler.get(hostHandle, accessType);
 					if (host != null) {
 						LOGGER.warn(String.format("Could not connect to %s. Try connecting to %s",
 							failedHost.getDescription(), host.getDescription()));

src/main/java/com/arangodb/internal/net/AccessType.java

@@ -0,0 +1,31 @@
+/*
+ * DISCLAIMER
+ *
+ * Copyright 2018 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.internal.net;
+
+/**
+ * @author Mark Vollmary
+ *
+ */
+public enum AccessType {
+
+	WRITE, READ, DIRTY_READ
+
+}

src/main/java/com/arangodb/internal/net/DirtyReadHostHandler.java

@@ -0,0 +1,87 @@
+/*
+ * DISCLAIMER
+ *
+ * Copyright 2018 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.internal.net;
+
+import java.io.IOException;
+
+/**
+ * @author Mark Vollmary
+ *
+ */
+public class DirtyReadHostHandler implements HostHandler {
+
+	private final HostHandler master;
+	private final HostHandler follower;
+	private AccessType currentAccessType;
+
+	public DirtyReadHostHandler(final HostHandler master, final HostHandler follower) {
+		super();
+		this.master = master;
+		this.follower = follower;
+	}
+
+	private HostHandler determineHostHandler() {
+		switch (currentAccessType) {
+		case DIRTY_READ:
+			return follower;
+		default:
+			return master;
+		}
+	}
+
+	@Override
+	public Host get(final HostHandle hostHandle, final AccessType accessType) {
+		this.currentAccessType = accessType;
+		return determineHostHandler().get(hostHandle, accessType);
+	}
+
+	@Override
+	public void success() {
+		determineHostHandler().success();
+	}
+
+	@Override
+	public void fail() {
+		determineHostHandler().fail();
+	}
+
+	@Override
+	public void reset() {
+		determineHostHandler().reset();
+	}
+
+	@Override
+	public void confirm() {
+		determineHostHandler().confirm();
+	}
+
+	@Override
+	public void close() throws IOException {
+		master.close();
+		follower.close();
+	}
+
+	@Override
+	public void closeCurrentOnError() {
+		determineHostHandler().closeCurrentOnError();
+	}
+
+}

src/main/java/com/arangodb/internal/net/FallbackHostHandler.java

@@ -43,7 +43,7 @@ public FallbackHostHandler(final HostResolver resolver) {
 	}
 
 	@Override
-	public Host get(final HostHandle hostHandle) {
+	public Host get(final HostHandle hostHandle, AccessType accessType) {
 		return current != lastSuccess || iterations < 3 ? current : null;
 	}
 

src/main/java/com/arangodb/internal/net/HostHandler.java

@@ -28,7 +28,7 @@
  */
 public interface HostHandler {
 
-	Host get(HostHandle hostHandle);
+	Host get(HostHandle hostHandle, AccessType accessType);
 
 	void success();
 

src/main/java/com/arangodb/internal/net/RandomHostHandler.java

@@ -44,7 +44,7 @@ public RandomHostHandler(final HostResolver resolver, final HostHandler fallback
 	}
 
 	@Override
-	public Host get(final HostHandle hostHandle) {
+	public Host get(final HostHandle hostHandle, AccessType accessType) {
 		if (current == null) {
 			origin = current = getRandomHost(false, true);
 		}
@@ -59,7 +59,7 @@ public void success() {
 	@Override
 	public void fail() {
 		fallback.fail();
-		current = fallback.get(null);
+		current = fallback.get(null, null);
 	}
 
 	private Host getRandomHost(final boolean initial, final boolean closeConnections) {

src/main/java/com/arangodb/internal/net/RoundRobinHostHandler.java

@@ -42,7 +42,7 @@ public RoundRobinHostHandler(final HostResolver resolver) {
 	}
 
 	@Override
-	public Host get(final HostHandle hostHandle) {
+	public Host get(final HostHandle hostHandle, AccessType accessType) {
 		final List<Host> hosts = resolver.resolve(false, false);
 		final int size = hosts.size();
 		if (fails > size) {