package com.hubspot.mesos.client;

import java.util.List;

import org.apache.mesos.Protos.MasterInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fasterxml.jackson.core.type.TypeReference;
import com.google.inject.Inject;
import com.google.inject.name.Named;
import com.hubspot.horizon.HttpClient;
import com.hubspot.horizon.HttpRequest;
import com.hubspot.horizon.HttpResponse;
import com.hubspot.mesos.JavaUtils;
import com.hubspot.mesos.MesosUtils;
import com.hubspot.mesos.json.MesosMasterStateObject;
import com.hubspot.mesos.json.MesosSlaveStateObject;
import com.hubspot.mesos.json.MesosTaskMonitorObject;

public class MesosClient {

  public static final String HTTP_CLIENT_NAME = "mesos.http.client";

  private static final Logger LOG = LoggerFactory.getLogger(MesosClient.class);

  private static final String MASTER_STATE_FORMAT = "http://%s/master/state.json";
  private static final String MESOS_SLAVE_JSON_URL = "http://%s:5051/slave(1)/state.json";
  private static final String MESOS_SLAVE_STATISTICS_URL = "http://%s:5051/monitor/statistics.json";

  private static final TypeReference<List<MesosTaskMonitorObject>> TASK_MONITOR_TYPE_REFERENCE = new TypeReference<List<MesosTaskMonitorObject>>() {};

  private final HttpClient httpClient;

  @Inject
  public MesosClient(@Named(HTTP_CLIENT_NAME) HttpClient httpClient) {
    this.httpClient = httpClient;
  }

  public String getMasterUri(MasterInfo masterInfo) {
    return String.format(MASTER_STATE_FORMAT, MesosUtils.getMasterHostAndPort(masterInfo));
  }

  @SuppressWarnings("serial")
  public static class MesosClientException extends RuntimeException {

    public MesosClientException(String message) {
      super(message);
    }

    public MesosClientException(String message, Throwable cause) {
      super(message, cause);
    }

  }

  private HttpResponse getFromMesos(String uri) {
    HttpResponse response = null;

    final long start = System.currentTimeMillis();

    LOG.debug("Fetching {} from mesos", uri);

    try {
      response = httpClient.execute(HttpRequest.newBuilder().setUrl(uri).build());

      LOG.debug("Response {} - {} after {}", response.getStatusCode(), uri, JavaUtils.duration(start));
    } catch (Exception e) {
      throw new MesosClientException(String.format("Exception fetching %s after %s", uri, JavaUtils.duration(start)), e);
    }

    if (!response.isSuccess()) {
      throw new MesosClientException(String.format("Invalid response code from %s : %s", uri, response.getStatusCode()));
    }

    return response;
  }

  private <T> T getFromMesos(String uri, Class<T> clazz) {
    HttpResponse response = getFromMesos(uri);

    try {
      return response.getAs(clazz);
    } catch (Exception e) {
      throw new MesosClientException(String.format("Couldn't deserialize %s from %s", clazz.getSimpleName(), uri), e);
    }
  }

  public MesosMasterStateObject getMasterState(String uri) {
    return getFromMesos(uri, MesosMasterStateObject.class);
  }

  public String getSlaveUri(String hostname) {
    return String.format(MESOS_SLAVE_JSON_URL, hostname);
  }

  public MesosSlaveStateObject getSlaveState(String uri) {
    return getFromMesos(uri, MesosSlaveStateObject.class);
  }

  public List<MesosTaskMonitorObject> getSlaveResourceUsage(String hostname) {
    final String uri = String.format(MESOS_SLAVE_STATISTICS_URL, hostname);

    HttpResponse response = getFromMesos(uri);

    try {
      return response.getAs(TASK_MONITOR_TYPE_REFERENCE);
    } catch (Exception e) {
      throw new MesosClientException(String.format("Unable to deserialize task monitor object from %s", uri), e);
    }
  }

}

@@ -1468,43 +1468,21 @@
 urn 
-String.format(MASTER_STATE_FORMAT, 
+getMasterUri(
 Meso
@@ -1528,16 +1528,140 @@
 );%0A  %7D%0A%0A
+  public String getMasterUri(String hostnameAndPort) %7B%0A    return String.format(MASTER_STATE_FORMAT, hostnameAndPort);%0A  %7D%0A%0A
   @Suppr

add other method for master uri