added connection pooling (issue #103)

Author: mpv1989

ChangeLog

@@ -3,6 +3,7 @@ v4.1.11 (2017-03-xx)
 * added convenience methods (ArangoDatabase.arango(), ArangoCollection.db(), ArangoGraph.db())
 * added convenience methods (ArangoCollection.getIndex(String), .deleteIndex(key))
 * fixed exception handling in Connection (issue #110)
+* added connection pooling (issue #103)
 
 v4.1.10 (2017-02-22)
 ---------------------------

docs/setup.md

@@ -18,7 +18,8 @@ The driver is configured with some default values:
 <tr><td>arangodb.user</td><td>Basic Authentication User</td><td></td></tr>
 <tr><td>arangodb.password</td><td>Basic Authentication Password</td><td></td></tr>
 <tr><td>arangodb.useSsl</td><td>use SSL connection</td><td>false</td></tr>
-<tr><td>harangodb.chunksize</td><td>VelocyStream Chunk content-size(bytes)</td><td>30000</td></tr>
+<tr><td>arangodb.chunksize</td><td>VelocyStream Chunk content-size(bytes)</td><td>30000</td></tr>
+<tr><td>arangodb.connections.max</td><td>max number of connections</td><td>1</td></tr>
 </table>
 
 To customize the configuration the parameters can be changed in the code...
@@ -51,4 +52,14 @@ To use SSL, you have to set the configuration `useSsl` to `true` and set a `SSLC
 
   ArangoDB arangoDB = new ArangoDB.Builder().useSsl(true).sslContext(sc).build();
 
+```
+
+## Connection Pooling
+
+The driver supports connection pooling with a default of 1 maximum connections. To change this value use the method `maxConnections(Integer)` in `ArangoDB.Builder`.
+
+``` Java
+
+  ArangoDB arangoDB = new ArangoDB.Builder().maxConnections(8).build();
+
 ```

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

@@ -80,6 +80,7 @@ public static class Builder {
 		private Boolean useSsl;
 		private SSLContext sslContext;
 		private Integer chunksize;
+		private Integer maxConnections;
 		private final VPack.Builder vpackBuilder;
 		private final CollectionCache collectionCache;
 		private final VPackParser vpackParser;
@@ -109,6 +110,7 @@ public Builder loadProperties(final InputStream in) throws ArangoDBException {
 					password = loadPassword(properties, password);
 					useSsl = loadUseSsl(properties, useSsl);
 					chunksize = loadChunkSize(properties, chunksize);
+					maxConnections = loadMaxConnections(properties, maxConnections);
 				} catch (final IOException e) {
 					throw new ArangoDBException(e);
 				}
@@ -184,6 +186,11 @@ public Builder chunksize(final Integer chunksize) {
 			return this;
 		}
 
+		public Builder maxConnections(final Integer maxConnections) {
+			this.maxConnections = maxConnections;
+			return this;
+		}
+
 		public <T> Builder registerSerializer(final Class<T> clazz, final VPackSerializer<T> serializer) {
 			vpackBuilder.registerSerializer(clazz, serializer);
 			return this;
@@ -259,7 +266,8 @@ public ArangoDB build() {
 			}
 			return new ArangoDB(
 					new CommunicationSync.Builder(new DefaultHostHandler(hosts)).timeout(timeout).user(user)
-							.password(password).useSsl(useSsl).sslContext(sslContext).chunksize(chunksize),
+							.password(password).useSsl(useSsl).sslContext(sslContext).chunksize(chunksize)
+							.maxConnections(maxConnections),
 					vpackBuilder.build(), vpackBuilder.serializeNullValues(true).build(), vpackParser, collectionCache);
 		}
 
@@ -279,6 +287,7 @@ public ArangoDatabase db(final String name) {
 		});
 	}
 
+	@Override
 	protected ArangoExecutorSync executor() {
 		return executor;
 	}

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

@@ -37,6 +37,7 @@ public class ArangoDBConstants {
 	public static final int CHUNK_MIN_HEADER_SIZE = INTEGER_BYTES + INTEGER_BYTES + LONG_BYTES;
 	public static final int CHUNK_MAX_HEADER_SIZE = CHUNK_MIN_HEADER_SIZE + LONG_BYTES;
 	public static final int CHUNK_DEFAULT_CONTENT_SIZE = 30000;
+	public static final int MAX_CONNECTIONS_DEFAULT = 1;
 
 	public static final String PATH_API_DOCUMENT = "/_api/document";
 	public static final String PATH_API_COLLECTION = "/_api/collection";

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

@@ -61,6 +61,7 @@ public class InternalArangoDB<E extends ArangoExecutor<R, C>, R, C extends Conne
 	private static final String PROPERTY_KEY_PASSWORD = "arangodb.password";
 	private static final String PROPERTY_KEY_USE_SSL = "arangodb.usessl";
 	private static final String PROPERTY_KEY_V_STREAM_CHUNK_CONTENT_SIZE = "arangodb.chunksize";
+	private static final String PROPERTY_KEY_MAX_CONNECTIONS = "arangodb.connections.max";
 	protected static final String DEFAULT_PROPERTY_FILE = "/arangodb.properties";
 
 	public InternalArangoDB(final E executor) {
@@ -116,6 +117,11 @@ protected static Integer loadChunkSize(final Properties properties, final Intege
 			ArangoDBConstants.CHUNK_DEFAULT_CONTENT_SIZE));
 	}
 
+	protected static Integer loadMaxConnections(final Properties properties, final Integer currentValue) {
+		return Integer.parseInt(getProperty(properties, PROPERTY_KEY_MAX_CONNECTIONS, currentValue,
+			ArangoDBConstants.MAX_CONNECTIONS_DEFAULT));
+	}
+
 	private static <T> String getProperty(
 		final Properties properties,
 		final String key,

src/main/java/com/arangodb/internal/velocystream/Communication.java

@@ -52,7 +52,7 @@ public abstract class Communication<R, C extends Connection> {
 
 	protected static final AtomicLong mId = new AtomicLong(0L);
 	protected final VPack vpack;
-	protected final C connection;
+	protected final ConnectionPool<C> connectionPool;
 	protected final CollectionCache collectionCache;
 
 	protected final String user;
@@ -62,21 +62,21 @@ public abstract class Communication<R, C extends Connection> {
 
 	protected Communication(final Integer timeout, final String user, final String password, final Boolean useSsl,
 		final SSLContext sslContext, final VPack vpack, final CollectionCache collectionCache, final Integer chunksize,
-		final C connection) {
+		final ConnectionPool<C> connectionPool) {
 		this.user = user;
 		this.password = password;
 		this.vpack = vpack;
 		this.collectionCache = collectionCache;
-		this.connection = connection;
+		this.connectionPool = connectionPool;
 		this.chunksize = chunksize != null ? chunksize : ArangoDBConstants.CHUNK_DEFAULT_CONTENT_SIZE;
 	}
 
-	protected void connect(final Connection connection) {
+	protected void connect(final C connection) {
 		if (!connection.isOpen()) {
 			try {
 				connection.open();
 				if (user != null) {
-					authenticate();
+					authenticate(connection);
 				}
 			} catch (final IOException e) {
 				LOGGER.error(e.getMessage(), e);
@@ -85,17 +85,17 @@ protected void connect(final Connection connection) {
 		}
 	}
 
-	protected abstract void authenticate();
+	protected abstract void authenticate(final C connection);
 
 	public void disconnect() {
-		disconnect(connection);
+		connectionPool.disconnect();
 	}
 
-	public void disconnect(final Connection connection) {
-		connection.close();
+	public R execute(final Request request) throws ArangoDBException {
+		return execute(request, connectionPool.connection());
 	}
 
-	public abstract R execute(final Request request) throws ArangoDBException;
+	public abstract R execute(final Request request, C connection) throws ArangoDBException;
 
 	protected void checkError(final Response response) throws ArangoDBException {
 		try {

src/main/java/com/arangodb/internal/velocystream/CommunicationSync.java

@@ -50,6 +50,7 @@ public static class Builder {
 		private Boolean useSsl;
 		private SSLContext sslContext;
 		private Integer chunksize;
+		private Integer maxConnections;
 
 		public Builder(final HostHandler hostHandler) {
 			super();
@@ -86,26 +87,38 @@ public Builder chunksize(final Integer chunksize) {
 			return this;
 		}
 
+		public Builder maxConnections(final Integer maxConnections) {
+			this.maxConnections = maxConnections;
+			return this;
+		}
+
 		public Communication<Response, ConnectionSync> build(final VPack vpack, final CollectionCache collectionCache) {
 			return new CommunicationSync(hostHandler, timeout, user, password, useSsl, sslContext, vpack,
-					collectionCache, chunksize);
+					collectionCache, chunksize, maxConnections);
 		}
 	}
 
 	protected CommunicationSync(final HostHandler hostHandler, final Integer timeout, final String user,
 		final String password, final Boolean useSsl, final SSLContext sslContext, final VPack vpack,
-		final CollectionCache collectionCache, final Integer chunksize) {
+		final CollectionCache collectionCache, final Integer chunksize, final Integer maxConnections) {
 		super(timeout, user, password, useSsl, sslContext, vpack, collectionCache, chunksize,
-				new ConnectionSync.Builder(hostHandler, new MessageStore()).timeout(timeout).useSsl(useSsl)
-						.sslContext(sslContext).build());
+				new ConnectionPool<ConnectionSync>(maxConnections) {
+					private final ConnectionSync.Builder builder = new ConnectionSync.Builder(hostHandler,
+							new MessageStore()).timeout(timeout).useSsl(useSsl).sslContext(sslContext);
+
+					@Override
+					public ConnectionSync createConnection() {
+						return builder.build();
+					}
+				});
 	}
 
 	@Override
-	public Response execute(final Request request) throws ArangoDBException {
+	public Response execute(final Request request, final ConnectionSync connection) throws ArangoDBException {
 		connect(connection);
 		try {
 			final Message requestMessage = createMessage(request);
-			final Message responseMessage = send(requestMessage);
+			final Message responseMessage = send(requestMessage, connection);
 			collectionCache.setDb(request.getDatabase());
 			final Response response = createResponse(responseMessage);
 			checkError(response);
@@ -116,7 +129,7 @@ public Response execute(final Request request) throws ArangoDBException {
 
 	}
 
-	private Message send(final Message message) throws ArangoDBException {
+	private Message send(final Message message, final ConnectionSync connection) throws ArangoDBException {
 		if (LOGGER.isDebugEnabled()) {
 			LOGGER.debug(String.format("Send Message (id=%s, head=%s, body=%s)", message.getId(), message.getHead(),
 				message.getBody() != null ? message.getBody() : "{}"));
@@ -125,9 +138,10 @@ private Message send(final Message message) throws ArangoDBException {
 	}
 
 	@Override
-	protected void authenticate() {
+	protected void authenticate(final ConnectionSync connection) {
 		final Response response = execute(
-			new AuthenticationRequest(user, password != null ? password : "", ArangoDBConstants.ENCRYPTION_PLAIN));
+			new AuthenticationRequest(user, password != null ? password : "", ArangoDBConstants.ENCRYPTION_PLAIN),
+			connection);
 		checkError(response);
 	}
 

src/main/java/com/arangodb/internal/velocystream/ConnectionPool.java

@@ -0,0 +1,62 @@
+/*
+ * 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.internal.velocystream;
+
+import java.util.LinkedList;
+
+import com.arangodb.internal.ArangoDBConstants;
+
+/**
+ * @author Mark - mark at arangodb.com
+ *
+ */
+public abstract class ConnectionPool<C extends Connection> {
+
+	private final LinkedList<C> connections;
+	private final int maxConnections;
+
+	public ConnectionPool(final Integer maxConnections) {
+		super();
+		this.maxConnections = maxConnections != null ? Math.max(1, maxConnections)
+				: ArangoDBConstants.MAX_CONNECTIONS_DEFAULT;
+		connections = new LinkedList<C>();
+	}
+
+	public abstract C createConnection();
+
+	public synchronized C connection() {
+		final C c;
+		if (connections.size() < maxConnections) {
+			c = createConnection();
+		} else {
+			c = connections.removeFirst();
+		}
+		connections.add(c);
+		return c;
+	}
+
+	public void disconnect() {
+		while (!connections.isEmpty()) {
+			connections.removeLast().close();
+		}
+	}
+
+}

src/test/java/com/arangodb/internal/velocystream/CommunicationTest.java

@@ -91,4 +91,17 @@ public void run() {
 		assertThat(iterator.next(), is(SLOW));
 	}
 
+	@Test
+	public void minOneConnection() {
+		final ArangoDB arangoDB = new ArangoDB.Builder().maxConnections(0).build();
+		final ArangoDBVersion version = arangoDB.getVersion();
+		assertThat(version, is(notNullValue()));
+	}
+
+	@Test
+	public void defaultMaxConnection() {
+		final ArangoDB arangoDB = new ArangoDB.Builder().maxConnections(null).build();
+		final ArangoDBVersion version = arangoDB.getVersion();
+		assertThat(version, is(notNullValue()));
+	}
 }