/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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 org.apache.zeppelin.interpreter.remote;

import org.apache.commons.exec.*;
import org.apache.commons.exec.environment.EnvironmentUtils;
import org.apache.zeppelin.helium.ApplicationEventListener;
import org.apache.zeppelin.interpreter.InterpreterException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Map;

/**
 * This class manages start / stop of remote interpreter process
 */
public class RemoteInterpreterManagedProcess extends RemoteInterpreterProcess
    implements ExecuteResultHandler {
  private static final Logger logger = LoggerFactory.getLogger(
      RemoteInterpreterManagedProcess.class);
  private final String interpreterRunner;

  private DefaultExecutor executor;
  private ExecuteWatchdog watchdog;
  boolean running = false;
  private int port = -1;
  private final String interpreterDir;
  private final String localRepoDir;

  private Map<String, String> env;

  public RemoteInterpreterManagedProcess(
      String intpRunner,
      String intpDir,
      String localRepoDir,
      Map<String, String> env,
      int connectTimeout,
      RemoteInterpreterProcessListener listener,
      ApplicationEventListener appListener) {
    super(new RemoteInterpreterEventPoller(listener, appListener),
        connectTimeout);
    this.interpreterRunner = intpRunner;
    this.env = env;
    this.interpreterDir = intpDir;
    this.localRepoDir = localRepoDir;
  }

  RemoteInterpreterManagedProcess(String intpRunner,
                                  String intpDir,
                                  String localRepoDir,
                                  Map<String, String> env,
                                  RemoteInterpreterEventPoller remoteInterpreterEventPoller,
                                  int connectTimeout) {
    super(remoteInterpreterEventPoller,
        connectTimeout);
    this.interpreterRunner = intpRunner;
    this.env = env;
    this.interpreterDir = intpDir;
    this.localRepoDir = localRepoDir;
  }

  @Override
  public String getHost() {
    return "localhost";
  }

  @Override
  public int getPort() {
    return port;
  }

  @Override
  public void start() {
    // start server process
    try {
      port = RemoteInterpreterUtils.findRandomAvailablePortOnAllLocalInterfaces();
    } catch (IOException e1) {
      throw new InterpreterException(e1);
    }

    CommandLine cmdLine = CommandLine.parse(interpreterRunner);
    cmdLine.addArgument("-d", false);
    cmdLine.addArgument(interpreterDir, false);
    cmdLine.addArgument("-p", false);
    cmdLine.addArgument(Integer.toString(port), false);
    cmdLine.addArgument("-l", false);
    cmdLine.addArgument(localRepoDir, false);

    executor = new DefaultExecutor();

    watchdog = new ExecuteWatchdog(ExecuteWatchdog.INFINITE_TIMEOUT);
    executor.setWatchdog(watchdog);

    try {
      Map procEnv = EnvironmentUtils.getProcEnvironment();
      procEnv.putAll(env);

      logger.info("Run interpreter process {}", cmdLine);
      executor.execute(cmdLine, procEnv, this);
      running = true;
    } catch (IOException e) {
      running = false;
      throw new InterpreterException(e);
    }


    long startTime = System.currentTimeMillis();
    while (System.currentTimeMillis() - startTime < getConnectTimeout()) {
      if (RemoteInterpreterUtils.checkIfRemoteEndpointAccessible("localhost", port)) {
        break;
      } else {
        try {
          Thread.sleep(500);
        } catch (InterruptedException e) {
          logger.error("Exception in RemoteInterpreterProcess while synchronized reference " +
              "Thread.sleep", e);
        }
      }
    }
  }

  public void stop() {
    if (isRunning()) {
      logger.info("kill interpreter process");
      watchdog.destroyProcess();
    }

    executor = null;
    watchdog = null;
    running = false;
    logger.info("Remote process terminated");
  }

  @Override
  public void onProcessComplete(int exitValue) {
    logger.info("Interpreter process exited {}", exitValue);
    running = false;

  }

  @Override
  public void onProcessFailed(ExecuteException e) {
    logger.info("Interpreter process failed {}", e);
    running = false;
  }

  public boolean isRunning() {
    return running;
  }
}

@@ -1142,24 +1142,53 @@
 OException;%0A
+import java.io.OutputStream;%0A
 import java.
@@ -1197,16 +1197,16 @@
 il.Map;%0A
-
 %0A/**%0A * 
@@ -2267,24 +2267,25 @@
 calRepoDir;%0A
+%0A
   %7D%0A%0A  Remot
@@ -3579,17 +3579,106 @@
 utor();%0A
+    executor.setStreamHandler(new PumpStreamHandler(new ProcessLogOutputStream(logger)));
 %0A
-
     watc
@@ -5154,24 +5154,24 @@
 Running() %7B%0A
-
     return r
@@ -5181,11 +5181,301 @@
 ing;%0A  %7D
+%0A%0A  private static class ProcessLogOutputStream extends LogOutputStream %7B%0A%0A    private Logger logger;%0A%0A    public ProcessLogOutputStream(Logger logger) %7B%0A      this.logger = logger;%0A    %7D%0A%0A    @Override%0A    protected void processLine(String s, int i) %7B%0A      this.logger.debug(s);%0A    %7D%0A  %7D
 %0A%7D%0A

ZEPPELIN-1187. Redirect output of the process running interpreter.sh to log4j