package scray.client.finagle;

import java.net.SocketAddress;
import java.sql.SQLException;
import java.util.List;

import scray.client.jdbc.ScrayURL;
import scray.service.qservice.thriftjava.ScrayMetaTService;
import scray.service.qservice.thriftjava.ScrayTServiceEndpoint;

import com.twitter.finagle.Thrift;
import com.twitter.util.Await;
import com.twitter.util.Duration;
import com.twitter.util.Future;
import com.google.common.base.Joiner;

public class ScrayTServiceManager {

	private org.slf4j.Logger log = org.slf4j.LoggerFactory
			.getLogger(ScrayTServiceManager.class);

	private ScrayTServiceManager() {
	}

	private static class SingletonHolder {
		static ScrayTServiceManager instance = new ScrayTServiceManager();
	}

	public static ScrayTServiceManager getInstance() {
		return SingletonHolder.instance;
	}

	private class MetaServiceConnection {
		private ScrayMetaTService.FutureIface metaServiceClient;
		private ScrayURL scrayURL;

		MetaServiceConnection(ScrayURL scrayURL) {
			this.scrayURL = scrayURL;
		}

		String seedEndpoint() {
			return scrayURL.getHostAndPort();
		}

		ScrayMetaTService.FutureIface getMetaClient() {
			if (metaServiceClient == null) {
				metaServiceClient = Thrift.newIface(seedEndpoint(),
						ScrayMetaTService.FutureIface.class);
			}
			return metaServiceClient;
		}
	}

	private List<ScrayTServiceEndpoint> endpointCache = null;
	private MetaServiceConnection connection = null;

	private java.util.Random rand = new java.util.Random();

	private int TIMEOUT = 10; // secs
	private long REFRESH = 3 * 60 * 1000; // milis

	public void init(ScrayURL scrayURL) {
		// initialize if new, (re)initialize if different, else reuse
		if (connection == null) {
			connection = new MetaServiceConnection(scrayURL);
		} else if (connection.scrayURL.equals(scrayURL)) {
			connection = new MetaServiceConnection(scrayURL);
		}

		// initially fill the endpoint cache
		refreshEndpoints();

		// schedule refresh of endpointCache
		new java.util.Timer().schedule(new java.util.TimerTask() {
			@Override
			public void run() {
				refreshEndpoints();
			}
		}, REFRESH, REFRESH);
	}

	void refreshEndpoints() {
		try {
			Future<List<ScrayTServiceEndpoint>> eplist = connection
					.getMetaClient().getServiceEndpoints();
			endpointCache = Await.result(eplist, Duration.fromSeconds(TIMEOUT));
			log.info("Refreshed scray service endpoints: "
					+ Joiner.on(", ").join(endpointCache));
		} catch (Exception e) {
			log.error("Error refreshing scray service enpoint cache.", e);
		}
	}

	public String getRandomEndpoint() throws SQLException {
		try {
			int nextIdx = rand.nextInt(endpointCache.size());
			ScrayTServiceEndpoint nextEp = endpointCache.get(nextIdx);
			return getHostAndPort(nextEp);
		} catch (Exception e) {
			String msg = "Error connecting with BDQ Scray Service: no endpoint available.";
			log.error(msg, e);
			throw new SQLException(msg);
		}
	}

	private String getHostAndPort(ScrayTServiceEndpoint endpoint) {
		String DELIMITER = ":";
		return endpoint.getHost() + DELIMITER + endpoint.getPort();
	}

	public List<ScrayTServiceEndpoint> getEndpointCache() {
		return endpointCache;
	}

	public ScrayURL getScrayURL() {
		return connection.scrayURL;
	}

	public boolean isStatefulTService() {
		return connection.scrayURL.getProtocolMode().equals(
				ScrayURL.ProtocolModes.stateful.name());
	}

	public boolean isStatelessTService() {
		return connection.scrayURL.getProtocolMode().equals(
				ScrayURL.ProtocolModes.stateless.name());
	}
}

@@ -28,39 +28,8 @@
 e;%0A%0A
-import java.net.SocketAddress;%0A
 impo
@@ -76,16 +76,68 @@
 il.List;
+%0Aimport java.util.Timer;%0Aimport java.util.TimerTask;
 %0A%0Aimport
@@ -290,16 +290,54 @@
 point;%0A%0A
+import com.google.common.base.Joiner;%0A
 import c
@@ -463,46 +463,8 @@
 ure;
-%0Aimport com.google.common.base.Joiner;
 %0A%0Apu
@@ -1958,78 +1958,23 @@
 %0A%0A%09%09
-// schedule refresh of endpointCache%0A%09%09new java.util.Timer().schedule(
+TimerTask tt = 
 new 
@@ -1992,24 +1992,27 @@
 imerTask() %7B
+%09%09%09
 %0A%09%09%09@Overrid
@@ -2068,16 +2068,125 @@
 %09%09%09%7D%0A%09%09%7D
+;%0A%0A%09%09// schedule refresh of endpointCache (as daemon)%0A%09%09Timer t = new Timer(true);%0A%09%09t.scheduleAtFixedRate(tt
 , REFRES

make scray endpoint refreshing a daemon