import java.io.UnsupportedEncodingException;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.List;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

public class ConsistentHashLoadBalance extends AbstractLoadBalance {
	private final ConcurrentMap<String, ConsistentHashSelector> selectors = new ConcurrentHashMap<String, ConsistentHashSelector>();

	@Override
	protected String doSelect(List<String> servers, String type, String key) {
		int identityHashCode = System.identityHashCode(servers);
		ConsistentHashSelector selector = selectors.get(type);
		if (selector == null
				|| selector.getIdentityHashCode() != identityHashCode) {
			selectors.put(type, new ConsistentHashSelector(servers, type));
			selector = selectors.get(type);
		}
		return selector.select(key);

	}

	private static final class ConsistentHashSelector {

		private final TreeMap<Long, String> virtualServers;

		private final int replicaNumber;

		private final int identityHashCode;

		public ConsistentHashSelector(List<String> servers, String type) {
			this.virtualServers = new TreeMap<Long, String>();
			this.identityHashCode = System.identityHashCode(servers);
			this.replicaNumber = 160;
			for (String invoker : servers) {
				for (int i = 0; i < replicaNumber / 4; i++) {
					byte[] digest = md5(type + invoker + i);
					for (int h = 0; h < 4; h++) {
						long m = hash(digest, h);
						virtualServers.put(m, invoker);
					}
				}
			}
		}

		public int getIdentityHashCode() {
			return identityHashCode;
		}

		public String select(String key) {
			byte[] digest = md5(key);
			String invoker = sekectForKey(hash(digest, 0));
			return invoker;
		}

		private String sekectForKey(long hash) {
			String invoker;
			Long key = hash;
			if (!virtualServers.containsKey(key)) {
				SortedMap<Long, String> tailMap = virtualServers.tailMap(key);
				if (tailMap.isEmpty()) {
					key = virtualServers.firstKey();
				} else {
					key = tailMap.firstKey();
				}
			}
			invoker = virtualServers.get(key);
			return invoker;
		}

		private long hash(byte[] digest, int number) {
			return (((long) (digest[3 + number * 4] & 0xFF) << 24)
					| ((long) (digest[2 + number * 4] & 0xFF) << 16)
					| ((long) (digest[1 + number * 4] & 0xFF) << 8) | (digest[0 + number * 4] & 0xFF)) & 0xFFFFFFFFL;
		}

		private byte[] md5(String value) {
			MessageDigest md5;
			try {
				md5 = MessageDigest.getInstance("MD5");
			} catch (NoSuchAlgorithmException e) {
				throw new IllegalStateException(e.getMessage(), e);
			}
			md5.reset();
			byte[] bytes = null;
			try {
				bytes = value.getBytes("UTF-8");
			} catch (UnsupportedEncodingException e) {
				throw new IllegalStateException(e.getMessage(), e);
			}
			md5.update(bytes);
			return md5.digest();
		}
	}

}