package com.continuuity.gateway.handlers;

import com.continuuity.common.conf.Constants;
import com.continuuity.common.discovery.RandomEndpointStrategy;
import com.continuuity.common.discovery.TimeLimitEndpointStrategy;
import com.continuuity.http.AbstractHttpHandler;
import com.continuuity.http.HttpResponder;
import com.google.common.base.Throwables;
import com.google.inject.Inject;
import com.ning.http.client.Response;
import com.ning.http.client.SimpleAsyncHttpClient;
import org.apache.twill.discovery.Discoverable;
import org.apache.twill.discovery.DiscoveryServiceClient;
import org.jboss.netty.handler.codec.http.HttpRequest;
import org.jboss.netty.handler.codec.http.HttpResponseStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;

/**
 * Monitor Handler returns the status of different discoverable services
 */
@Path(Constants.Gateway.GATEWAY_VERSION)
public class MonitorHandler extends AbstractHttpHandler {
  private static final Logger LOG = LoggerFactory.getLogger(MonitorHandler.class);

  private static final String VERSION = Constants.Gateway.GATEWAY_VERSION;

  private final DiscoveryServiceClient discoveryServiceClient;

  /**
   * Timeout to get response from discovered service.
   */
  private static final long SERVICE_PING_RESPONSE_TIMEOUT = TimeUnit.MILLISECONDS.convert(1, TimeUnit.MINUTES);

  /**
   * Number of seconds for timing out a service endpoint discovery.
   */
  private static final long DISCOVERY_TIMEOUT_SECONDS = 3;

  private enum Services {
    METRICS (Constants.Service.METRICS),
    TRANSACTION (Constants.Service.TRANSACTION),
    STREAMS (Constants.Service.STREAM_HANDLER),
    APPFABRIC (Constants.Service.APP_FABRIC_HTTP);

    private final String name;

    private Services(String name) {
      this.name = name;
    }

    public String getName() { return name; }

    public static Services valueofName(String name) { return valueOf(name.toUpperCase()); }
  }

  @Inject
  public MonitorHandler(DiscoveryServiceClient discoveryServiceClient) {
    this.discoveryServiceClient = discoveryServiceClient;
  }

  @Path("/monitor/{service-id}")
  @GET
  public void monitor(final HttpRequest request, final HttpResponder responder,
                      @PathParam("service-id") final String service) {
    if (discoverService(service)) {
      //Service is discoverable
      String response = String.format("%s is OK\n", service);
      responder.sendString(HttpResponseStatus.OK, response);
    } else {
      String response = String.format("%s not found\n", service);
      responder.sendString(HttpResponseStatus.NOT_FOUND, service);
    }
  }

  private boolean discoverService(String serviceName) {
    try {
      Iterable<Discoverable> discoverables = this.discoveryServiceClient.discover(Services.valueofName(
                                                                                  serviceName).getName());
      Discoverable discoverable = new TimeLimitEndpointStrategy(new RandomEndpointStrategy(discoverables),
                                                                DISCOVERY_TIMEOUT_SECONDS, TimeUnit.SECONDS).pick();
      if (discoverable == null) {
        return false;
      } else {
        // Ping the discovered service to check its status
        String url = String.format("http://%s:%d/ping", discoverable.getSocketAddress().getHostName(),
                                   discoverable.getSocketAddress().getPort());
        //TODO: Figure out a way to Ping Transaction Service (Hint: Use RUOK?)
        if (!Services.valueofName(serviceName).equals(Services.TRANSACTION)) {
          return checkGetStatus(url).equals(HttpResponseStatus.OK);
        }
        return true;
      }
    } catch (IllegalArgumentException e) {
      return false;
    } catch (Exception e) {
      LOG.warn("Unable to ping {} : Reason : {}", serviceName, e.getMessage());
      return false;
    }
  }

  private HttpResponseStatus checkGetStatus(String url) throws Exception {
    SimpleAsyncHttpClient client = new SimpleAsyncHttpClient.Builder()
      .setUrl(url)
      .setRequestTimeoutInMs((int) SERVICE_PING_RESPONSE_TIMEOUT)
      .build();

    try {
      Future<Response> future = client.get();
      Response response = future.get(SERVICE_PING_RESPONSE_TIMEOUT, TimeUnit.MILLISECONDS);
      return HttpResponseStatus.valueOf(response.getStatusCode());
    } catch (Exception e) {
      Throwables.propagate(e);
    } finally {
      client.close();
    }
    return HttpResponseStatus.NOT_FOUND;
  }
}

@@ -82,16 +82,74 @@
 stants;%0A
+import com.continuuity.common.discovery.EndpointStrategy;%0A
 import c
@@ -196,32 +196,32 @@
 dpointStrategy;%0A
-
 import com.conti
@@ -263,32 +263,91 @@
 dpointStrategy;%0A
+import com.continuuity.gateway.handlers.util.ThriftHelper;%0A
 import com.conti
@@ -578,32 +578,77 @@
 syncHttpClient;%0A
+import org.apache.thrift.protocol.TProtocol;%0A
 import org.apach
@@ -1066,16 +1066,111 @@
 Param;%0A%0A
+import static com.continuuity.data2.transaction.distributed.thrift.TTransactionServer.Client;%0A%0A
 /**%0A * M
@@ -3201,82 +3201,8 @@
 me(%0A
-                                                                          
     
@@ -3240,33 +3240,41 @@
     
-Discoverable discoverable
+EndpointStrategy endpointStrategy
  = n
@@ -3302,16 +3302,25 @@
 trategy(
+%0A        
 new Rand
@@ -3357,117 +3357,105 @@
 es),
-%0A                                                                DISCOVERY_TIMEOUT_SECONDS, TimeUnit.SECONDS)
+ DISCOVERY_TIMEOUT_SECONDS, TimeUnit.SECONDS);%0A      Discoverable discoverable = endpointStrategy
 .pic
@@ -3767,24 +3767,25 @@
 getPort());%0A
+%0A
         //TO
@@ -3786,76 +3786,82 @@
   //
-TODO: Figure out a way to Ping Transaction Service (Hint: Use RUOK?)
+Use Thrift Client for Transaction and HTTP PingHandlers for other services
 %0A   
@@ -4009,24 +4009,31 @@
 );%0A        %7D
+ else %7B
 %0A        ret
@@ -4033,20 +4033,217 @@
     
-return true;
+  TProtocol protocol =  ThriftHelper.getThriftProtocol(Constants.Service.TRANSACTION, endpointStrategy);%0A          Client client = new Client(protocol);%0A          return client.status().equals(%22OK%22);%0A        %7D
 %0A   

Use thrift client to check status of Transaction service