/*
 *  Copyright (c) 2012-2013 Malhar, Inc.
 *  All Rights Reserved.
 */
package com.datatorrent.stram.client;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Map;
import org.apache.commons.io.IOUtils;
import org.codehaus.jettison.json.JSONException;
import org.codehaus.jettison.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author David Yan <david@datatorrent.com>
 */
public class CLIProxy
{
  private static final Logger LOG = LoggerFactory.getLogger(CLIProxy.class);

  public static class CommandException extends Exception
  {
    public CommandException(String message)
    {
      super(message);
    }
  }

  public static JSONObject getLogicalPlan(String jarUri, String appName) throws Exception
  {
    return issueCommand("show-logical-plan \"" + jarUri + "\" \"" + appName + "\"");
  }

  public static JSONObject launchApp(String jarUri, String appName, Map<String, String> properties) throws Exception
  {
    StringBuilder sb = new StringBuilder("launch ");
    for (Map.Entry<String, String> entry : properties.entrySet()) {
      sb.append("-D ");
      sb.append(entry.getKey());
      sb.append("=\"");
      sb.append(entry.getValue());
      sb.append("\" ");
    }
    sb.append(jarUri);
    sb.append(" \"");
    sb.append(appName);
    sb.append("\"");
    return issueCommand(sb.toString());
  }

  public static JSONObject getApplications(String jarUri) throws Exception
  {
    return issueCommand("show-logical-plan \"" + jarUri + "\"");
  }

  private static JSONObject issueCommand(String command) throws Exception
  {
    String shellCommand = "dtcli -r -e '" + command + "'";
    Process p = Runtime.getRuntime().exec(new String[] { "sh", "-c", shellCommand } );
    String error = IOUtils.toString(p.getErrorStream());
    String result = IOUtils.toString(p.getInputStream());
    p.waitFor();
    int exitValue = p.exitValue();
    LOG.debug("Executed: {}", shellCommand);
    LOG.debug("Output: {}", result);
    LOG.debug("Error: {}", error);
    if (exitValue == 0) {
      return new JSONObject(result);
    } else {
      throw new CommandException(error);
    }
  }
}

@@ -124,84 +124,9 @@
 .io.
-ByteArrayInputStream;%0Aimport java.io.IOException;%0Aimport java.io.InputStream
+*
 ;%0Aim
@@ -629,24 +629,668 @@
 age);%0A    %7D%0A
+%0A  %7D%0A%0A  private static class StreamGobbler extends Thread%0A  %7B%0A    InputStream is;%0A    StringBuilder content = new StringBuilder();%0A%0A    StreamGobbler(InputStream is)%0A    %7B%0A      this.is = is;%0A    %7D%0A%0A    String getContent()%0A    %7B%0A      return content.toString();%0A    %7D%0A%0A    @Override%0A    public void run()%0A    %7B%0A      try %7B%0A        InputStreamReader isr = new InputStreamReader(is);%0A        BufferedReader br = new BufferedReader(isr);%0A        String line;%0A        while ((line = br.readLine()) != null) %7B%0A          content.append(line);%0A        %7D%0A      %7D%0A      catch (IOException ex) %7B%0A        LOG.error(%22Caught exception%22, ex);%0A      %7D%0A    %7D%0A%0A
   %7D%0A%0A  publi
@@ -2331,17 +2331,16 @@
 ring%5B%5D %7B
- 
 %22sh%22, %22-
@@ -2355,19 +2355,17 @@
 lCommand
- %7D 
+%7D
 );%0A    S
@@ -2370,129 +2370,195 @@
  Str
-ing error = IOUtils.toString(p.getErrorStream());%0A    String result = IOUtils.toString(p.getInputStream());%0A    p.waitFor
+eamGobbler errorGobbler = new StreamGobbler(p.getErrorStream());%0A    StreamGobbler outputGobbler = new StreamGobbler(p.getInputStream());%0A    errorGobbler.start();%0A    outputGobbler.start
 ();%0A
@@ -2579,25 +2579,23 @@
 lue = p.
-exitValue
+waitFor
 ();%0A    
@@ -2663,22 +2663,42 @@
 t: %7B%7D%22, 
-result
+outputGobbler.getContent()
 );%0A    L
@@ -2720,24 +2720,44 @@
 : %7B%7D%22, error
+Gobbler.getContent()
 );%0A    if (e
@@ -2805,22 +2805,42 @@
 ect(
-result
+outputGobbler.getContent()
 );%0A    %7D
  els
@@ -2835,16 +2835,20 @@
 );%0A    %7D
+%0A   
  else %7B%0A
@@ -2885,16 +2885,36 @@
 on(error
+Gobbler.getContent()
 );%0A    %7D
@@ -2918,10 +2918,11 @@
   %7D%0A  %7D%0A
+%0A
 %7D%0A

use streamgobbler to gobble the stdout and stderr to prevent blocking