<commit_before>/*
 * 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();
		}
	}

}<commit_msg>Clean up broken connections from routing layer<commit_after>/*
 * 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.selection.CacheInstanceSelector;
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.Lists;
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;

	@Inject
	private CacheInstanceSelector selectionLayer;

	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();

		List<CacheInstance> unestablishedConnections = Lists.newArrayList();
		boolean foundBrokenConnection = false;
		// Update connection map with new list of available instances
		for (CacheInstance instance : availableInstances) {
			// If the connection already exists, reuse old connection
			if (this.routingMap.containsKey(instance.hashCode())) {
				newRoutingMap.put(instance.hashCode(), this.routingMap.get(instance.hashCode()));
			} else {
				// else create a new connection
				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));
					unestablishedConnections.add(instance);
					foundBrokenConnection = true;
				}
			}
		}

		if (foundBrokenConnection) {
			// Remove unestablishable connections from the list of available instances
			availableInstances.removeAll(unestablishedConnections);

			// Update the selection layer with the broken connections removed
			this.selectionLayer.update(availableInstances);
		}

		// Hold on to old map for GC
		Map<Integer, StorageConnection> oldReference = this.routingMap;

		// Update current map reference
		this.routingMap = newRoutingMap;

		// Remove live connections from the old map
		newRoutingMap.keySet().forEach(oldReference::remove);

		// Close all remaining (presumed dead) connections in the old connection map
		oldReference.values().forEach(StorageConnection::close);
	}

}
{'fraction_non_alphanumeric': 0.10464300743442279, 'fraction_numerical': 0.002524898302707252, 'mean_word_length': 4.609756097560975, 'pattern_counts': {'":': 0, '<': 14, '<?xml version=': 0, '>': 14, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7149834', 'n_tokens_mistral': 2229, 'n_tokens_neox': 1951, 'n_words': 494}