package com.continuuity.explore.executor;

import com.continuuity.api.data.batch.RecordScannable;
import com.continuuity.api.dataset.Dataset;
import com.continuuity.common.conf.Constants;
import com.continuuity.data2.dataset2.DatasetFramework;
import com.continuuity.explore.client.DatasetExploreFacade;
import com.continuuity.explore.service.ExploreService;
import com.continuuity.explore.service.Handle;
import com.continuuity.http.AbstractHttpHandler;
import com.continuuity.http.HttpResponder;
import com.continuuity.internal.io.UnsupportedTypeException;
import com.google.gson.JsonObject;
import com.google.inject.Inject;
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 javax.ws.rs.POST;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;

/**
 * Handler that implements internal explore APIs.
 */
@Path(Constants.Gateway.GATEWAY_VERSION)
public class ExploreExecutorHttpHandler extends AbstractHttpHandler {
  private static final Logger LOG = LoggerFactory.getLogger(QueryExecutorHttpHandler.class);

  private final ExploreService exploreService;
  private final DatasetFramework datasetFramework;

  @Inject
  public ExploreExecutorHttpHandler(ExploreService exploreService, DatasetFramework datasetFramework) {
    this.exploreService = exploreService;
    this.datasetFramework = datasetFramework;
  }

  /**
   * Enable ad-hoc exploration of a dataset instance.
   */
  @POST
  @Path("/explore/instances/{instance}/enable")
  public void enableExplore(@SuppressWarnings("UnusedParameters") HttpRequest request, HttpResponder responder,
                            @PathParam("instance") final String instance) {
    try {
      LOG.debug("Enabling explore for dataset instance {}", instance);
      Dataset dataset = datasetFramework.getDataset(instance, null);
      if (dataset == null) {
        responder.sendError(HttpResponseStatus.NOT_FOUND, "Cannot load dataset " + instance);
        return;
      }

      if (!(dataset instanceof RecordScannable)) {
        // It is not an error to get non-RecordScannable datasets, since the type of dataset may not be known where this
        // call originates from.
        LOG.debug("Dataset {} does not implement {}", instance, RecordScannable.class.getName());
        JsonObject json = new JsonObject();
        json.addProperty("handle", Handle.NO_OP.getHandle());
        responder.sendJson(HttpResponseStatus.OK, json);
        return;
      }

      RecordScannable<?> scannable = (RecordScannable) dataset;
      String createStatement;
      try {
        createStatement = DatasetExploreFacade.generateCreateStatement(instance, scannable);
      } catch (UnsupportedTypeException e) {
        LOG.error("Exception while generating create statement for dataset {}", instance, e);
        responder.sendError(HttpResponseStatus.BAD_REQUEST, e.getMessage());
        return;
      }

      LOG.debug("Running create statement for dataset {} with row scannable {} - {}",
                instance,
                dataset.getClass().getName(),
                createStatement);

      Handle handle = exploreService.execute(createStatement);
      JsonObject json = new JsonObject();
      json.addProperty("handle", handle.getHandle());
      responder.sendJson(HttpResponseStatus.OK, json);
    } catch (Throwable e) {
      LOG.error("Got exception:", e);
      responder.sendError(HttpResponseStatus.INTERNAL_SERVER_ERROR, e.getMessage());
    }
  }

  /**
   * Disable ad-hoc exploration of a dataset instance.
   */
  @POST
  @Path("/explore/instances/{instance}/disable")
  public void disableExplore(@SuppressWarnings("UnusedParameters") HttpRequest request, HttpResponder responder,
                             @PathParam("instance") final String instance) {
    try {
      LOG.debug("Disabling explore for dataset instance {}", instance);

      Dataset dataset = datasetFramework.getDataset(instance, null);
      if (dataset == null) {
        responder.sendError(HttpResponseStatus.NOT_FOUND, "Cannot load dataset " + instance);
        return;
      }

      if (!(dataset instanceof RecordScannable)) {
        // It is not an error to get non-RecordScannable datasets, since the type of dataset may not be known where this
        // call originates from.
        LOG.debug("Dataset {} does not implement {}", instance, RecordScannable.class.getName());
        JsonObject json = new JsonObject();
        json.addProperty("handle", Handle.NO_OP.getHandle());
        responder.sendJson(HttpResponseStatus.OK, json);
        return;
      }

      String deleteStatement = DatasetExploreFacade.generateDeleteStatement(instance);
      LOG.debug("Running delete statement for dataset {} - {}",
                instance,
                deleteStatement);

      Handle handle = exploreService.execute(deleteStatement);
      JsonObject json = new JsonObject();
      json.addProperty("handle", handle.getHandle());
      responder.sendJson(HttpResponseStatus.OK, json);
    } catch (Throwable e) {
      LOG.error("Got exception:", e);
      responder.sendError(HttpResponseStatus.INTERNAL_SERVER_ERROR, e.getMessage());
    }
  }

}

@@ -1775,141 +1775,763 @@
     
-LOG.debug(%22Enabling explore for dataset instance %7B%7D%22, instance);%0A      Dataset dataset = datasetFramework.getDataset(instance, null);
+Dataset dataset;%0A      try %7B%0A        dataset = datasetFramework.getDataset(instance, null);%0A      %7D catch (Exception e) %7B%0A        String className = isClassNotFoundException(e);%0A        if (className == null) %7B%0A          throw e;%0A        %7D%0A        LOG.info(%22Cannot load dataset %7B%7D because class %7B%7D cannot be found. This is probably because class %7B%7D is a %22 +%0A                   %22type parameter of dataset %7B%7D that is not present in the dataset's jar file. See the developer %22 +%0A                   %22guide for more information.%22, instance, className, className, instance);%0A        JsonObject json = new JsonObject();%0A        json.addProperty(%22handle%22, Handle.NO_OP.getHandle());%0A        responder.sendJson(HttpResponseStatus.OK, json);%0A        return;%0A      %7D
 %0A   
@@ -3162,24 +3162,95 @@
 n;%0A      %7D%0A%0A
+      LOG.debug(%22Enabling explore for dataset instance %7B%7D%22, instance);%0A
       Record
@@ -4227,32 +4227,32 @@
 .getMessage());%0A
-
     %7D%0A  %7D%0A%0A  /**
@@ -4242,24 +4242,276 @@
 %0A    %7D%0A  %7D%0A%0A
+  private String isClassNotFoundException(Throwable e) %7B%0A    if (e instanceof ClassNotFoundException) %7B%0A      return e.getMessage();%0A    %7D%0A    if (e.getCause() != null) %7B%0A      return isClassNotFoundException(e.getCause());%0A    %7D%0A    return null;%0A  %7D%0A%0A
   /**%0A   * D

fix misleading error messages if generic dataset cannot be loaded