/*
 * Copyright (C) 2016 Turn Inc. All Rights Reserved.
 * Proprietary and confidential.
 */

package com.turn.edc.router;

import com.turn.edc.exception.KeyNotFoundException;
import com.turn.edc.discovery.CacheInstance;
import com.turn.edc.discovery.DiscoveryListener;
import com.turn.edc.storage.StorageConnection;
import com.turn.edc.storage.ConnectionFactory;

import java.io.IOException;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeoutException;

import com.google.common.collect.Maps;
import com.google.inject.Inject;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Routing layer for edc requests
 *
 * @author tshiou
 */
public class RequestRouter extends DiscoveryListener {
	private static final Logger logger = LoggerFactory.getLogger(RequestRouter.class);

	@Inject
	private ConnectionFactory connectorFactory;

	private volatile Map<Integer, StorageConnection> routingMap = Maps.newConcurrentMap();
	private final static int TIMEOUT = 10;

	public RequestRouter() {
	}

	public byte[] get(CacheInstance source, String key) throws KeyNotFoundException, TimeoutException, IOException {
		StorageConnection connection = routingMap.get(source.hashCode());

		if (connection == null) {
			throw new IOException("Data source not registered!");
		}

		return connection.get(key, TIMEOUT);
	}

	public boolean store(CacheInstance destination, StoreRequest request) {

		StorageConnection connection = routingMap.get(destination.hashCode());

		if (connection == null) {
			return false;
		}

		connection.post(request);

		return true;
	}

	/**
	 * Closes all connections and clears the routing map
	 */
	public void close() {
		for(Iterator<StorageConnection> it = routingMap.values().iterator(); it.hasNext(); ) {
			StorageConnection conn = it.next();
			conn.close();
			it.remove();
		}
	}

	@Override
	public void update(List<CacheInstance> availableInstances) {
		Map<Integer, StorageConnection> newRoutingMap = Maps.newConcurrentMap();

		for (CacheInstance instance : availableInstances) {
			if (this.routingMap.containsKey(instance.hashCode())) {
				newRoutingMap.put(instance.hashCode(), this.routingMap.get(instance.hashCode()));
			} else {
				StorageConnection newConnection;
				try {
					newConnection = connectorFactory.create(
							instance.getHostAndPort().getHostText(),
							Integer.toString(instance.getHostAndPort().getPort()),
							TIMEOUT
					);
					newRoutingMap.put(instance.hashCode(), newConnection);
				} catch (IOException ioe) {
					logger.error("Cache instance {} found but connection was not able to be established", instance.toString());
					logger.debug(ExceptionUtils.getStackTrace(ioe));
					// TODO: need to remove from selection scheme
				}
			}
		}

		Map<Integer, StorageConnection> tempReference = this.routingMap;

		this.routingMap = newRoutingMap;

		for (StorageConnection conn : tempReference.values()) {
			conn.close();
		}
	}

}

@@ -259,24 +259,77 @@
 ryListener;%0A
+import com.turn.edc.selection.CacheInstanceSelector;%0A
 import com.t
@@ -558,16 +558,56 @@
 ption;%0A%0A
+import com.google.common.collect.Lists;%0A
 import c
@@ -1044,16 +1044,73 @@
 ctory;%0A%0A
+%09@Inject%0A%09private CacheInstanceSelector selectionLayer;%0A%0A
 %09private
@@ -2233,59 +2233,296 @@
 %0A%0A%09%09
-for (CacheInstance instance : availableInstances) %7B
+List%3CCacheInstance%3E unestablishedConnections = Lists.newArrayList();%0A%09%09boolean foundBrokenConnection = false;%0A%09%09// Update connection map with new list of available instances%0A%09%09for (CacheInstance instance : availableInstances) %7B%0A%09%09%09// If the connection already exists, reuse old connection
 %0A%09%09%09
@@ -2675,16 +2675,52 @@
  else %7B%0A
+%09%09%09%09// else create a new connection%0A
 %09%09%09%09Stor
@@ -3205,69 +3205,423 @@
 %09%09%09%09
-// TODO: need to remove from selection scheme%0A%09%09%09%09%7D
+unestablishedConnections.add(instance);%0A%09%09%09%09%09foundBrokenConnection = true;%0A%09%09%09%09%7D%0A%09%09%09%7D%0A%09%09%7D%0A%0A%09%09if (foundBrokenConnection) %7B%0A%09%09%09// Remove unestablishable connections from the list of available instances%0A%09%09%09availableInstances.removeAll(unestablishedConnections);%0A%0A%09%09%09// Update the selection layer with the broken connections removed%0A%09%09%09this.selectionLayer.update(availableInstances);
 %0A%09%09
-%09
 %7D
+%0A
 %0A%09%09
-%7D%0A
+// Hold on to old map for GC
 %0A%09%09M
@@ -3651,20 +3651,19 @@
 ection%3E 
-temp
+old
 Referenc
@@ -3684,16 +3684,50 @@
 ngMap;%0A%0A
+%09%09// Update current map reference%0A
 %09%09this.r
@@ -3760,85 +3760,247 @@
 %0A%0A%09%09
-for (StorageConnection conn : tempReference.values()) %7B%0A%09%09%09conn.
+// Remove live connections from the old map%0A%09%09newRoutingMap.keySet().forEach(oldReference::remove);%0A%0A%09%09// Close all remaining (presumed dead) connections in the old connection map%0A%09%09oldReference.values().forEach(StorageConnection::
 close
-(
 );%0A
-%09%09%7D%0A
 %09%7D%0A%0A

Clean up broken connections from routing layer