/*
 * Copyright © 2016 Cask Data, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package co.cask.cdap.gateway.handlers.preview;

import co.cask.cdap.api.data.schema.Schema;
import co.cask.cdap.app.preview.PreviewManager;
import co.cask.cdap.common.NotFoundException;
import co.cask.cdap.common.conf.CConfiguration;
import co.cask.cdap.common.conf.Constants;
import co.cask.cdap.gateway.handlers.util.AbstractAppFabricHttpHandler;
import co.cask.cdap.internal.io.SchemaTypeAdapter;
import co.cask.cdap.proto.id.NamespaceId;
import co.cask.cdap.proto.id.PreviewId;
import co.cask.http.HttpResponder;
import com.google.common.base.Charsets;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.inject.Inject;
import com.google.inject.Singleton;
import org.jboss.netty.handler.codec.http.HttpRequest;
import org.jboss.netty.handler.codec.http.HttpResponseStatus;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;

/**
 * {@link co.cask.http.HttpHandler} to manage program lifecycle for v3 REST APIs
 */
@Singleton
@Path(Constants.Gateway.API_VERSION_3 + "/namespaces/{namespace-id}")
public class PreviewHttpHandler extends AbstractAppFabricHttpHandler {
  private static final Gson GSON = new GsonBuilder().registerTypeAdapter(Schema.class,
                                                                         new SchemaTypeAdapter()).create();
  private final PreviewManager previewManager;
  private final int limit;

  @Inject
  PreviewHttpHandler(CConfiguration cConf, PreviewManager previewManager) {
    this.previewManager = previewManager;
    limit = cConf.getInt("preview.records.limit", 50);
  }

  @POST
  @Path("/preview")
  public void startPreview(HttpRequest request, HttpResponder responder,
                           @PathParam("namespace-id") String namespaceId) throws Exception {
    String config = request.getContent().toString(Charsets.UTF_8);
    responder.sendString(HttpResponseStatus.OK,
                         GSON.toJson(previewManager.start(new NamespaceId(namespaceId), config)));
  }

  @GET
  @Path("/previews/{preview-id}/status")
  public void getPreviewStatus(HttpRequest request, HttpResponder responder,
                               @PathParam("namespace-id") String namespaceId,
                               @PathParam("preview-id") String previewId) throws NotFoundException {
    responder.sendString(HttpResponseStatus.OK,
                         GSON.toJson(previewManager.getStatus(new PreviewId(namespaceId, previewId))));
  }

  @GET
  @Path("/previews/{preview-id}/stages/{stage-name}")
  public void getPreviewData(HttpRequest request, HttpResponder responder,
                             @PathParam("namespace-id") String namespaceId,
                             @PathParam("preview-id") String previewId,
                             @PathParam("stage-name") String stageName) throws NotFoundException {
    Map<String, List<String>> data = previewManager.getData(new PreviewId(namespaceId, previewId), stageName);
    Map<String, List<String>> result = new HashMap<>();
    for (Map.Entry<String, List<String>> entry : data.entrySet()) {
      String stage = entry.getKey();
      List<String> stageData = entry.getValue();
      List<String> values = stageData.size() <= limit ? stageData : stageData.subList(0, limit);
      result.put(stage, values);
    }
    responder.sendString(HttpResponseStatus.OK, GSON.toJson(result));
  }
}

@@ -781,56 +781,8 @@
 on;%0A
-import co.cask.cdap.common.conf.CConfiguration;%0A
 impo
@@ -1356,79 +1356,8 @@
 s;%0A%0A
-import java.util.HashMap;%0Aimport java.util.List;%0Aimport java.util.Map;%0A
 impo
@@ -1943,35 +1943,8 @@
 ger;
-%0A  private final int limit;
 %0A%0A  
@@ -1976,30 +1976,8 @@
 ler(
-CConfiguration cConf, 
 Prev
@@ -2051,63 +2051,8 @@
 er;%0A
-    limit = cConf.getInt(%22preview.records.limit%22, 50);%0A
   %7D%0A
@@ -3317,527 +3317,161 @@
     
-Map%3CString, List%3CString%3E%3E data = previewManager.getData(new PreviewId(namespaceId, previewId), stageName);%0A    Map%3CString, List%3CString%3E%3E result = new HashMap%3C%3E();%0A    for (Map.Entry%3CString, List%3CString%3E%3E entry : data.entrySet()) %7B%0A      String stage = entry.getKey();%0A      List%3CString%3E stageData = entry.getValue();%0A      List%3CString%3E values = sta
+responder.sendString(HttpResponseStatus.OK,%0A                         GSON.toJson(previewManager.
 ge
+t
 Data
-.size() %3C= limit ? stageData : stageData.subList(0, limit);%0A      result.put(stage, values);%0A    %7D%0A    responder.sendString(HttpResponseStatus.OK, GSON.toJson(result
+(new PreviewId(namespaceId, previewId), stageName)
 ));%0A

Revert "Limit the data returned by the preview handler."