package com.linkedin.helix.webapp.resources;

import java.io.IOException;

import org.codehaus.jackson.JsonGenerationException;
import org.codehaus.jackson.map.JsonMappingException;
import org.restlet.data.MediaType;
import org.restlet.data.Status;
import org.restlet.resource.Representation;
import org.restlet.resource.Resource;
import org.restlet.resource.StringRepresentation;
import org.restlet.resource.Variant;

import com.linkedin.helix.HelixException;
import com.linkedin.helix.HelixProperty;
import com.linkedin.helix.PropertyKey;
import com.linkedin.helix.PropertyKey.Builder;
import com.linkedin.helix.ZNRecord;
import com.linkedin.helix.manager.zk.ZKHelixDataAccessor;
import com.linkedin.helix.manager.zk.ZkBaseDataAccessor;
import com.linkedin.helix.manager.zk.ZkClient;
import com.linkedin.helix.tools.ClusterSetup;
import com.linkedin.helix.webapp.RestAdminApplication;

public class ControllerResource extends Resource
{
  @Override
  public boolean allowGet()
  {
    return true;
  }

  @Override
  public boolean allowPost()
  {
    return true;
  }

  @Override
  public boolean allowPut()
  {
    return false;
  }

  @Override
  public boolean allowDelete()
  {
    return false;
  }

  StringRepresentation getControllerRepresentation(String clusterName) throws JsonGenerationException,
      JsonMappingException,
      IOException
  {
    Builder keyBuilder = new PropertyKey.Builder(clusterName);
    ZkClient zkClient =
        (ZkClient) getContext().getAttributes().get(RestAdminApplication.ZKCLIENT);

    ZKHelixDataAccessor accessor =
        new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<ZNRecord>(zkClient));

    // return an empty znrecord if leader not exist
    String value = ClusterRepresentationUtil.ZNRecordToJson(new ZNRecord(""));
    HelixProperty property = accessor.getProperty(keyBuilder.controllerLeader());
    if (property != null)
    {
      value = ClusterRepresentationUtil.ZNRecordToJson(property.getRecord());
    }

    StringRepresentation representation =
        new StringRepresentation(value, MediaType.APPLICATION_JSON);

    return representation;
  }

  @Override
  public Representation represent(Variant variant)
  {
    StringRepresentation presentation = null;
    try
    {
      String clusterName = (String) getRequest().getAttributes().get("clusterName");
      presentation = getControllerRepresentation(clusterName);
    }
    catch (Exception e)
    {
      String error = ClusterRepresentationUtil.getErrorAsJsonStringFromException(e);
      presentation = new StringRepresentation(error, MediaType.APPLICATION_JSON);
      e.printStackTrace();
    }
    return presentation;
  }

  @Override
  public void acceptRepresentation(Representation entity)
  {
    try
    {
      String clusterName = (String) getRequest().getAttributes().get("clusterName");
      ZkClient zkClient =
          (ZkClient) getContext().getAttributes().get(RestAdminApplication.ZKCLIENT);
      ClusterSetup setupTool = new ClusterSetup(zkClient);

      JsonParameters jsonParameters = new JsonParameters(entity);
      String command = jsonParameters.getCommand();

      if (command == null)
      {
        throw new HelixException("Could NOT find 'command' in parameterMap: " + jsonParameters._parameterMap);
      }
      else if (command.equalsIgnoreCase(ClusterSetup.enableCluster))
      {
        boolean enabled =
            Boolean.parseBoolean(jsonParameters.getParameter(JsonParameters.ENABLED));

        setupTool.getClusterManagementTool().enableCluster(clusterName, enabled);
      }
      else
      {
        throw new HelixException("Unsupported command: " + command
                                 + ". Should be one of [" + ClusterSetup.enableCluster + "]");
      }
      
      getResponse().setEntity(getControllerRepresentation(clusterName));
      getResponse().setStatus(Status.SUCCESS_OK);

    }
    catch (Exception e)
    {
      getResponse().setEntity(ClusterRepresentationUtil.getErrorAsJsonStringFromException(e),
                              MediaType.APPLICATION_JSON);
      getResponse().setStatus(Status.SUCCESS_OK);
    }
  }
}

@@ -66,16 +66,151 @@
 ception;
+%0Aimport java.text.DateFormat;%0Aimport java.text.SimpleDateFormat;%0Aimport java.util.Date;%0Aimport java.util.Map;%0Aimport java.util.TreeMap;
 %0A%0Aimport
@@ -716,16 +716,56 @@
 uilder;%0A
+import com.linkedin.helix.PropertyType;%0A
 import c
@@ -984,26 +984,172 @@
 lix.
-tools.ClusterSetup
+model.LiveInstance;%0Aimport com.linkedin.helix.model.PauseSignal;%0Aimport com.linkedin.helix.tools.ClusterSetup;%0Aimport com.linkedin.helix.util.StatusUpdateUtil.Level
 ;%0Aim
@@ -1978,168 +1978,594 @@
 );%0A%0A
+%0A
     
-// return an empty zn
+ZNRecord record = null;%0A    LiveInstance leader = accessor.getProperty(keyBuilder.controllerLeader());%0A    if (leader != null)%0A    %7B%0A      
 record 
-if
+=
  leader
- not exist%0A    String value = ClusterRepresentationUtil.ZNRecordToJson(new ZNRecord(%22%22));%0A    HelixProperty property
+.getRecord();%0A    %7D%0A    else%0A    %7B%0A      record = new ZNRecord(%22%22);%0A      DateFormat formatter = new SimpleDateFormat(%22yyyyMMdd-HHmmss.SSSSSS%22);%0A      String time = formatter.format(new Date());%0A      Map%3CString, String%3E contentMap = new TreeMap%3CString, String%3E();%0A      contentMap.put(%22AdditionalInfo%22, %22No leader exists%22);%0A      record.setMapField(Level.HELIX_INFO + %22-%22 + time, contentMap);%0A    %7D%0A    %0A    boolean paused
  = 
+(
 acce
@@ -2596,72 +2596,130 @@
 der.
-controllerLeader());%0A    if (property != null)%0A    %7B%0A      value
+pause()) == null? false : true);%0A    record.setSimpleField(PropertyType.PAUSE.toString(), %22%22 + paused);%0A%0A    String retVal
  = C
@@ -2762,37 +2762,16 @@
 son(
-property.getRecord());%0A    %7D%0A
+record);
 %0A   
@@ -2846,13 +2846,14 @@
 ion(
-value
+retVal
 , Me

add pause info to helix web addmin of enableCluster