only remove not longer used hosts (and connection pools)

Author: BCaxelbecker

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

@@ -23,6 +23,9 @@
 import java.io.IOException;
 import java.util.Collection;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import com.arangodb.ArangoDB;
 import com.arangodb.ArangoDBException;
 import com.arangodb.ArangoDatabase;
@@ -58,6 +61,8 @@
  *
  */
 public class ArangoDBImpl extends InternalArangoDB<ArangoExecutorSync> implements ArangoDB {
+
+private static final Logger LOGGER = LoggerFactory.getLogger(ArangoDBImpl.class);
 
 private ArangoCursorInitializer cursorInitializer;
 private CommunicationProtocol cp;
@@ -81,6 +86,8 @@ public ArangoDBImpl(final VstCommunicationSync.Builder vstBuilder, final HttpCom
 
 hostResolver.init(this.executor(), util());
 
+LOGGER.info("ArangoDB Client is ready to use");
+
 }
 
 private static CommunicationProtocol createProtocol(

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

@@ -55,13 +55,14 @@
 import com.arangodb.velocypack.VPack;
 import com.arangodb.velocypack.VPackParser;
 
+
 /**
  * @author Mark Vollmary
  *
  */
 public abstract class InternalArangoDBBuilder {
 
-private static final Logger LOGGER = LoggerFactory.getLogger(InternalArangoDBBuilder.class);
+private static final Logger LOG = LoggerFactory.getLogger(InternalArangoDBBuilder.class);
 
 private static final String PROPERTY_KEY_HOSTS = "arangodb.hosts";
 private static final String PROPERTY_KEY_HOST = "arangodb.host";
@@ -200,17 +201,19 @@ protected void setSerializer(final ArangoSerialization serializer) {
 protected HostResolver createHostResolver(final Collection<Host> hosts, final int maxConnections,final ConnectionFactory connectionFactory) {
 
 if(acquireHostList) {
-LOGGER.debug("acquireHostList -> Use ExtendedHostResolver");
+LOG.debug("acquireHostList -> Use ExtendedHostResolver");
 return new ExtendedHostResolver(new ArrayList<Host>(hosts), maxConnections, connectionFactory, acquireHostListInterval);
 } else {
-LOGGER.debug("Use SimpleHostResolver");
+LOG.debug("Use SimpleHostResolver");
 return new SimpleHostResolver(new ArrayList<Host>(hosts));
 }
 
 }
 
 protected HostHandler createHostHandler(final HostResolver hostResolver) {
+
 final HostHandler hostHandler;
+
 if (loadBalancingStrategy != null) {
 switch (loadBalancingStrategy) {
 case ONE_RANDOM:
@@ -227,6 +230,9 @@ protected HostHandler createHostHandler(final HostResolver hostResolver) {
 } else {
 hostHandler = new FallbackHostHandler(hostResolver);
 }
+
+LOG.info("HostHandler is " + hostHandler.getClass().getSimpleName());
+
 return new DirtyReadHostHandler(hostHandler, new RoundRobinHostHandler(hostResolver));
 }
 

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

@@ -24,11 +24,19 @@
 import java.util.ArrayList;
 import java.util.List;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.arangodb.internal.velocystream.internal.VstConnection;
+import com.arangodb.internal.velocystream.internal.VstConnectionSync;
+
 /**
  * @author Mark Vollmary
  *
  */
 public class ConnectionPoolImpl implements ConnectionPool {
+
+private static final Logger LOGGER = LoggerFactory.getLogger(ConnectionPoolImpl.class);
 
 private final HostDescription host;
 private final int maxConnections;
@@ -53,7 +61,9 @@ public Connection createConnection(final HostDescription host) {
 
 @Override
 public synchronized Connection connection() {
+
 final Connection connection;
+
 if (connections.size() < maxConnections) {
 connection = createConnection(host);
 connections.add(connection);
@@ -62,6 +72,11 @@ public synchronized Connection connection() {
 final int index = (current++) % connections.size();
 connection = connections.get(index);
 }
+
+if(connection instanceof VstConnectionSync) {
+LOGGER.debug("Return Connection " + ((VstConnection)connection).getConnectionName());
+}
+
 return connection;
 }
 
@@ -73,4 +88,10 @@ public void close() throws IOException {
 connections.clear();
 }
 
+@Override
+public String toString() {
+return "ConnectionPoolImpl [host=" + host + ", maxConnections=" + maxConnections + ", connections="
++ connections.size() + ", current=" + current + ", factory=" + factory.getClass().getSimpleName() + "]";
+}
+
 }

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

@@ -20,6 +20,7 @@
 
 package com.arangodb.internal.net;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -81,18 +82,18 @@ public void init(ArangoExecutorSync executor, ArangoSerialization arangoSerializ
 
 @Override
 
-public HostSet resolve(final boolean initial, final boolean closeConnections) {
-
+public HostSet resolve(boolean initial, boolean closeConnections) {
+
 if (!initial && isExpired()) {
 
 lastUpdate = System.currentTimeMillis();
 
 final Collection<String> endpoints = resolveFromServer();
-LOGGER.debug("Resolve " + endpoints.size() + " Endpoints");
+LOGGER.info("Resolve " + endpoints.size() + " Endpoints");
 LOGGER.debug("Endpoints " + Arrays.deepToString(endpoints.toArray()));
 
 if (!endpoints.isEmpty()) {
-hosts.clear();
+hosts.markAllForDeletion();
 }
 
 for (final String endpoint : endpoints) {
@@ -117,6 +118,13 @@ public HostSet resolve(final boolean initial, final boolean closeConnections) {
 LOGGER.warn("Skip Endpoint (Format)" + endpoint);
 }
 }
+
+try {
+hosts.clearAllMarkedForDeletion();
+} catch (IOException e) {
+LOGGER.error("Cant close all Hosts with MarkedForDeletion", e);
+}
+
 }
 
 return hosts;

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

@@ -35,4 +35,8 @@ public interface Host {
 void closeOnError();
 
 void close() throws IOException;
+
+void setMarkforDeletion(boolean markforDeletion);
+
+boolean isMarkforDeletion();
 }

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

@@ -32,6 +32,7 @@ public class HostImpl implements Host {
 
 private final ConnectionPool connectionPool;
 private final HostDescription description;
+private boolean markforDeletion = false;
 
 public HostImpl(final ConnectionPool connectionPool, final HostDescription description) {
 super();
@@ -65,7 +66,43 @@ public void closeOnError() {
 
 @Override
 public String toString() {
-return "HostImpl [connectionPool=" + connectionPool + ", description=" + description + "]";
+return "HostImpl [connectionPool=" + connectionPool + ", description=" + description + ", markforDeletion="
++ markforDeletion + "]";
 }
+
+public boolean isMarkforDeletion() {
+return markforDeletion;
+}
+
+public void setMarkforDeletion(boolean markforDeletion) {
+this.markforDeletion = markforDeletion;
+}
+
+@Override
+public int hashCode() {
+final int prime = 31;
+int result = 1;
+result = prime * result + ((description == null) ? 0 : description.hashCode());
+return result;
+}
+
+@Override
+public boolean equals(Object obj) {
+if (this == obj)
+return true;
+if (obj == null)
+return false;
+if (getClass() != obj.getClass())
+return false;
+HostImpl other = (HostImpl) obj;
+if (description == null) {
+if (other.description != null)
+return false;
+} else if (!description.equals(other.description))
+return false;
+return true;
+}
+
+
 
 }

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

@@ -31,8 +31,16 @@ public List<Host> getHostsList() {
 }
 
 public void addHost(Host newHost) {
+
 if(hosts.contains(newHost)) {
 LOGGER.debug("Host" + newHost + " allready in Set");
+
+for (Host host : hosts) {
+if(host.equals(newHost)) {
+host.setMarkforDeletion(false);
+}
+}
+
 } else {
 hosts.add(newHost);
 LOGGER.debug("Added Host " + newHost + " - now " + hosts.size() + " Hosts in List");
@@ -55,6 +63,33 @@ public void close() {
 }
 }
 
+public void markAllForDeletion() {
+
+for (Host host : hosts) {
+host.setMarkforDeletion(true);
+}
+
+}
+
+public void clearAllMarkedForDeletion() throws IOException {
+
+LOGGER.debug("Clear all Hosts in Set with markForDeletion");
+
+for (Host host : hosts) {
+if(host.isMarkforDeletion()) {
+try {
+
+LOGGER.debug("Try to close Host " + host);
+host.close();
+
+} catch (IOException e) {
+LOGGER.warn("Error during closing the Host " + host, e);
+}
+}
+}
+
+}
+
 public void clear() {
 LOGGER.debug("Clear all Hosts in Set");
 

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

@@ -30,6 +30,7 @@
 import java.nio.ByteOrder;
 import java.util.Collection;
 import java.util.Date;
+import java.util.HashMap;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -69,6 +70,10 @@ public abstract class VstConnection implements Connection {
 private InputStream inputStream;
 
 private final HostDescription host;
+
+private HashMap<Long, Long> sendTimestamps = new HashMap<Long, Long>();
+
+private String connectionName;
 
 protected VstConnection(final HostDescription host, final Integer timeout, final Long ttl, final Boolean useSsl,
 final SSLContext sslContext, final MessageStore messageStore) {
@@ -79,6 +84,9 @@ protected VstConnection(final HostDescription host, final Integer timeout, final
 this.useSsl = useSsl;
 this.sslContext = sslContext;
 this.messageStore = messageStore;
+
+connectionName = "conenction_" + System.currentTimeMillis() + "_" + Math.random();
+LOGGER.debug("Connection " + connectionName + " created");
 }
 
 public boolean isOpen() {
@@ -118,11 +126,12 @@ public synchronized void open() throws IOException {
 ((SSLSocket) socket).startHandshake();
 }
 sendProtocolHeader();
+
 executor = Executors.newSingleThreadExecutor();
 executor.submit(new Callable<Void>() {
 @Override
 public Void call() throws Exception {
-LOGGER.debug("Start Callable for " + this.toString());
+LOGGER.debug("Start Callable for " + connectionName);
 
 final long openTime = new Date().getTime();
 final Long ttlTime = ttl != null ? openTime + ttl : null;
@@ -153,7 +162,7 @@ public Void call() throws Exception {
 }
 }
 
-LOGGER.debug("Stop Callable for " + this.toString());
+LOGGER.debug("Stop Callable for " + connectionName);
 
 return null;
 }
@@ -193,6 +202,7 @@ protected synchronized void writeIntern(final Message message, final Collection<
 if (LOGGER.isDebugEnabled()) {
 LOGGER.debug(String.format("Send chunk %s:%s from message %s", chunk.getChunk(),
 chunk.isFirstChunk() ? 1 : 0, chunk.getMessageId()));
+sendTimestamps.put(chunk.getMessageId(), System.currentTimeMillis());
 }
 writeChunkHead(chunk);
 final int contentOffset = chunk.getContentOffset();
@@ -210,6 +220,7 @@ protected synchronized void writeIntern(final Message message, final Collection<
 }
 outputStream.flush();
 } catch (final IOException e) {
+LOGGER.error("Error on Connection " + connectionName);
 throw new ArangoDBException(e);
 }
 }
@@ -245,10 +256,13 @@ protected Chunk readChunk() throws IOException {
 contentLength = length - ArangoDefaults.CHUNK_MIN_HEADER_SIZE;
 }
 final Chunk chunk = new Chunk(messageId, chunkX, messageLength, 0, contentLength);
+
 if (LOGGER.isDebugEnabled()) {
-LOGGER.debug(String.format("Received chunk %s:%s from message %s", chunk.getChunk(),
-chunk.isFirstChunk() ? 1 : 0, chunk.getMessageId()));
+
+LOGGER.debug(String.format("Received chunk %s:%s from message %s", chunk.getChunk(), chunk.isFirstChunk() ? 1 : 0, chunk.getMessageId()));
+LOGGER.debug("Responsetime for Message " + chunk.getMessageId() + " is " + (sendTimestamps.get(chunk.getMessageId())-System.currentTimeMillis()));
 }
+
 return chunk;
 }
 
@@ -268,5 +282,9 @@ protected void readBytesIntoBuffer(final byte[] buf, final int off, final int le
 }
 }
 }
+
+public String getConnectionName() {
+return this.connectionName;
+}
 
 }