Extract JobListener and ResultDistributor, adapt to telnet and http api

core/src/main/java/com/taobao/arthas/core/command/model/MessageModel.java

@@ -0,0 +1,24 @@
+package com.taobao.arthas.core.command.model;
+
+/**
+ * @author gongdewei 2020/4/2
+ */
+public class MessageModel extends ResultModel {
+    private String message;
+
+    public MessageModel() {
+    }
+
+    public MessageModel(String message) {
+        this.message = message;
+    }
+
+    public String getMessage() {
+        return message;
+    }
+
+    @Override
+    public String getType() {
+        return "message";
+    }
+}

core/src/main/java/com/taobao/arthas/core/command/model/ResultModel.java

@@ -0,0 +1,27 @@
+package com.taobao.arthas.core.command.model;
+
+/**
+ * Command execute result
+ *
+ * @author gongdewei 2020-03-26
+ */
+public abstract class ResultModel {
+
+    private Integer jobId;
+
+    /**
+     * Command type (name)
+     *
+     * @return
+     */
+    public abstract String getType();
+
+
+    public Integer getJobId() {
+        return jobId;
+    }
+
+    public void setJobId(Integer jobId) {
+        this.jobId = jobId;
+    }
+}

core/src/main/java/com/taobao/arthas/core/command/model/StatusModel.java

@@ -0,0 +1,52 @@
+package com.taobao.arthas.core.command.model;
+
+public class StatusModel extends ResultModel {
+    private int statusCode;
+    private String message;
+
+    public StatusModel() {
+    }
+
+    public StatusModel(int statusCode) {
+        this.statusCode = statusCode;
+    }
+
+    public StatusModel(int statusCode, String message) {
+        this.statusCode = statusCode;
+        this.message = message;
+    }
+
+    public int getStatusCode() {
+        return statusCode;
+    }
+
+    public StatusModel setStatusCode(int statusCode) {
+        this.statusCode = statusCode;
+        return this;
+    }
+
+    public String getMessage() {
+        return message;
+    }
+
+    public StatusModel setMessage(String message) {
+        this.message = message;
+        return this;
+    }
+
+    public StatusModel setStatus(int statusCode, String message) {
+        this.statusCode = statusCode;
+        this.message = message;
+        return this;
+    }
+
+    public StatusModel setStatus(int statusCode) {
+        return this.setStatus(statusCode, null);
+    }
+
+    @Override
+    public String getType() {
+        return "status";
+    }
+
+}

core/src/main/java/com/taobao/arthas/core/command/view/MessageView.java

@@ -0,0 +1,14 @@
+package com.taobao.arthas.core.command.view;
+
+import com.taobao.arthas.core.command.model.MessageModel;
+import com.taobao.arthas.core.shell.command.CommandProcess;
+
+/**
+ * @author gongdewei 2020/4/2
+ */
+public class MessageView extends ResultView<MessageModel> {
+    @Override
+    public void draw(CommandProcess process, MessageModel result) {
+        writeln(process, result.getMessage());
+    }
+}

core/src/main/java/com/taobao/arthas/core/command/view/ResultView.java

@@ -0,0 +1,30 @@
+package com.taobao.arthas.core.command.view;
+
+import com.taobao.arthas.core.command.model.ResultModel;
+import com.taobao.arthas.core.shell.command.CommandProcess;
+
+/**
+ * Command result view for telnet term/tty.
+ * Note: Result view is a reusable and stateless instance
+ *
+ * @author gongdewei 2020/3/27
+ */
+public abstract class ResultView<T extends ResultModel> {
+
+    /**
+     * formatted printing data to term/tty
+     *
+     * @param process
+     */
+    public abstract void draw(CommandProcess process, T result);
+
+    /**
+     * write str and append a new line
+     *
+     * @param process
+     * @param str
+     */
+    protected void writeln(CommandProcess process, String str) {
+        process.write(str).write("\n");
+    }
+}

core/src/main/java/com/taobao/arthas/core/command/view/ResultViewResolver.java

@@ -0,0 +1,65 @@
+package com.taobao.arthas.core.command.view;
+
+import com.alibaba.arthas.deps.org.slf4j.Logger;
+import com.alibaba.arthas.deps.org.slf4j.LoggerFactory;
+import com.taobao.arthas.core.command.model.*;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * @author gongdewei 2020/3/27
+ */
+public class ResultViewResolver {
+    private static final Logger logger = LoggerFactory.getLogger(ResultViewResolver.class);
+
+    private Map<String, ResultView> resultViewMap = new ConcurrentHashMap<String, ResultView>();
+
+    private static ResultViewResolver viewResolver;
+
+    public static ResultViewResolver getInstance() {
+        if (viewResolver == null) {
+            synchronized (ResultViewResolver.class) {
+                viewResolver = new ResultViewResolver();
+            }
+        }
+        return viewResolver;
+    }
+
+    static {
+        getInstance().registerResultViews();
+    }
+
+    private void registerResultViews() {
+        try {
+            registerView(new StatusModel(), new StatusView());
+            registerView(new VersionModel(), new VersionView());
+            registerView(new MessageModel(), new MessageView());
+            registerView(new HelpModel(), new HelpView());
+            //registerView(new HistoryModel(), new HistoryView());
+            registerView(new EchoModel(), new EchoView());
+        } catch (Throwable e) {
+            logger.error("register result view failed", e);
+        }
+    }
+
+    private ResultViewResolver() {
+    }
+
+//    public void registerView(Class<? extends ExecResult> resultClass, ResultView view) throws IllegalAccessException, InstantiationException {
+//        ExecResult instance = resultClass.newInstance();
+//        this.registerView(instance.getType(), view);
+//    }
+
+    public <T extends ResultModel> void registerView(T resultObject, ResultView view) {
+        this.registerView(resultObject.getType(), view);
+    }
+
+    public void registerView(String resultType, ResultView view) {
+        resultViewMap.put(resultType, view);
+    }
+
+    public ResultView getResultView(String resultType) {
+        return resultViewMap.get(resultType);
+    }
+}

core/src/main/java/com/taobao/arthas/core/command/view/StatusView.java

@@ -0,0 +1,18 @@
+package com.taobao.arthas.core.command.view;
+
+import com.taobao.arthas.core.command.model.StatusModel;
+import com.taobao.arthas.core.shell.command.CommandProcess;
+
+/**
+ * @author gongdewei 2020/3/27
+ */
+public class StatusView extends ResultView<StatusModel> {
+
+    @Override
+    public void draw(CommandProcess process, StatusModel result) {
+        if (result.getMessage() != null) {
+            writeln(process, result.getMessage());
+        }
+    }
+
+}

core/src/main/java/com/taobao/arthas/core/distribution/ResultDistributor.java

@@ -0,0 +1,21 @@
+package com.taobao.arthas.core.distribution;
+
+import com.taobao.arthas.core.command.model.ResultModel;
+
+/**
+ * Command result distributor, sending results to consumers who joins in the same session.
+ * @author gongdewei 2020-03-26
+ */
+public interface ResultDistributor {
+
+    /**
+     * Append the phased result to queue
+     * @param result a phased result of the command
+     */
+    void appendResult(ResultModel result);
+
+    /**
+     * Close result distribtor, release resources
+     */
+    void close();
+}

core/src/main/java/com/taobao/arthas/core/distribution/impl/TermResultDistributorImpl.java

@@ -0,0 +1,36 @@
+package com.taobao.arthas.core.distribution.impl;
+
+import com.taobao.arthas.core.command.model.ResultModel;
+import com.taobao.arthas.core.command.view.ResultView;
+import com.taobao.arthas.core.command.view.ResultViewResolver;
+import com.taobao.arthas.core.distribution.ResultDistributor;
+import com.taobao.arthas.core.shell.command.CommandProcess;
+
+/**
+ * Term/Tty Result Distributor
+ *
+ * @author gongdewei 2020-03-26
+ */
+public class TermResultDistributorImpl implements ResultDistributor {
+
+    private final CommandProcess commandProcess;
+    private final ResultViewResolver resultViewResolver;
+
+    public TermResultDistributorImpl(CommandProcess commandProcess) {
+        this.commandProcess = commandProcess;
+        this.resultViewResolver = ResultViewResolver.getInstance();
+    }
+
+    @Override
+    public void appendResult(ResultModel result) {
+        ResultView resultView = resultViewResolver.getResultView(result.getType());
+        if (resultView != null) {
+            resultView.draw(commandProcess, result);
+        }
+    }
+
+    @Override
+    public void close() {
+    }
+
+}

core/src/main/java/com/taobao/arthas/core/shell/ShellServer.java

@@ -5,6 +5,8 @@ import com.taobao.arthas.core.shell.future.Future;
 import com.taobao.arthas.core.shell.handlers.Handler;
 import com.taobao.arthas.core.shell.handlers.NoOpHandler;
 import com.taobao.arthas.core.shell.impl.ShellServerImpl;
+import com.taobao.arthas.core.shell.system.impl.InternalCommandManager;
+import com.taobao.arthas.core.shell.system.impl.JobControllerImpl;
 import com.taobao.arthas.core.shell.term.Term;
 import com.taobao.arthas.core.shell.term.TermServer;
 
@@ -102,4 +104,14 @@ public abstract class ShellServer {
      * @param completionHandler handler for getting notified when service is stopped
      */
     public abstract void close(Handler<Future<Void>> completionHandler);
+
+    /**
+     * @return global job controller instance
+     */
+    public abstract JobControllerImpl getJobController();
+
+    /**
+     * @return get command manager instance
+     */
+    public abstract InternalCommandManager getCommandManager();
 }

core/src/main/java/com/taobao/arthas/core/shell/command/CommandProcess.java

@@ -1,6 +1,7 @@
 package com.taobao.arthas.core.shell.command;
 
 import com.taobao.arthas.core.advisor.AdviceListener;
+import com.taobao.arthas.core.command.model.ResultModel;
 import com.taobao.arthas.core.shell.cli.CliToken;
 import com.taobao.arthas.core.shell.handlers.Handler;
 import com.taobao.arthas.core.shell.session.Session;
@@ -119,6 +120,13 @@ public interface CommandProcess extends Tty {
      */
     void end(int status);
 
+    /**
+     * End the process.
+     *
+     * @param status the exit status.
+     */
+    void end(int status, String message);
+
 
     /**
      * Register listener
@@ -167,4 +175,11 @@ public interface CommandProcess extends Tty {
      * Whether the process is running
      */
     boolean isRunning();
+
+    /**
+     * Append the phased result to queue
+     * @param result a phased result of the command
+     */
+    void appendResult(ResultModel result);
+
 }

core/src/main/java/com/taobao/arthas/core/shell/impl/ShellImpl.java

@@ -7,22 +7,21 @@ import com.taobao.arthas.core.shell.ShellServer;
 import com.taobao.arthas.core.shell.cli.CliToken;
 import com.taobao.arthas.core.shell.cli.CliTokens;
 import com.taobao.arthas.core.shell.future.Future;
-import com.taobao.arthas.core.shell.handlers.shell.CloseHandler;
-import com.taobao.arthas.core.shell.handlers.shell.CommandManagerCompletionHandler;
-import com.taobao.arthas.core.shell.handlers.shell.FutureHandler;
-import com.taobao.arthas.core.shell.handlers.shell.InterruptHandler;
-import com.taobao.arthas.core.shell.handlers.shell.ShellLineHandler;
-import com.taobao.arthas.core.shell.handlers.shell.SuspendHandler;
+import com.taobao.arthas.core.shell.handlers.shell.*;
 import com.taobao.arthas.core.shell.session.Session;
 import com.taobao.arthas.core.shell.session.impl.SessionImpl;
 import com.taobao.arthas.core.shell.system.ExecStatus;
 import com.taobao.arthas.core.shell.system.Job;
 import com.taobao.arthas.core.shell.system.JobController;
+import com.taobao.arthas.core.shell.system.JobListener;
 import com.taobao.arthas.core.shell.system.impl.InternalCommandManager;
 import com.taobao.arthas.core.shell.system.impl.JobControllerImpl;
 import com.taobao.arthas.core.shell.term.Term;
-import com.taobao.arthas.core.shell.term.impl.httptelnet.HttpTelnetTermServer;
+import com.taobao.arthas.core.shell.term.impl.TermImpl;
+import com.taobao.arthas.core.util.Constants;
+import com.taobao.arthas.core.util.FileUtils;
 
+import java.io.File;
 import java.lang.instrument.Instrumentation;
 import java.util.Date;
 import java.util.List;
@@ -79,7 +78,7 @@ public class ShellImpl implements Shell {
 
     @Override
     public synchronized Job createJob(List<CliToken> args) {
-        Job job = jobController.createJob(commandManager, args, this);
+        Job job = jobController.createJob(commandManager, args, session, new ShellJobHandler(this), term, null);
         return job;
     }
 
@@ -165,7 +164,7 @@ public class ShellImpl implements Shell {
                 // sometimes an NPE will be thrown during shutdown via web-socket,
                 // this ensures the shutdown process is finished properly
                 // https://github.com/alibaba/arthas/issues/320
-                logger.error("ARTHAS", "Error writing data:", t);
+                logger.error("Error writing data:", t);
             }
             term.close();
         } else {
@@ -180,4 +179,53 @@ public class ShellImpl implements Shell {
     public Job getForegroundJob() {
         return currentForegroundJob;
     }
+
+    private class ShellJobHandler implements JobListener {
+        ShellImpl shell;
+
+        public ShellJobHandler(ShellImpl shell) {
+            this.shell = shell;
+        }
+
+        @Override
+        public void onForeground(Job job) {
+            shell.setForegroundJob(job);
+            //reset stdin handler to job's origin handler
+            //shell.term().stdinHandler(job.process().getStdinHandler());
+        }
+
+        @Override
+        public void onBackground(Job job) {
+            resetAndReadLine();
+        }
+
+        @Override
+        public void onTerminated(Job job) {
+            if (!job.isRunInBackground()){
+                resetAndReadLine();
+            }
+
+            // save command history
+            Term term = shell.term();
+            if (term instanceof TermImpl) {
+                List<int[]> history = ((TermImpl) term).getReadline().getHistory();
+                FileUtils.saveCommandHistory(history, new File(Constants.CMD_HISTORY_FILE));
+            }
+        }
+
+        @Override
+        public void onSuspend(Job job) {
+            if (!job.isRunInBackground()){
+                resetAndReadLine();
+            }
+        }
+
+        private void resetAndReadLine() {
+            //reset stdin handler to echo handler
+            //shell.term().stdinHandler(null);
+            shell.setForegroundJob(null);
+            shell.readline();
+        }
+    }
+
 }

core/src/main/java/com/taobao/arthas/core/shell/impl/ShellServerImpl.java

@@ -242,4 +242,12 @@ public class ShellServerImpl extends ShellServer {
             bootstrap.destroy();
         }
     }
+
+    public JobControllerImpl getJobController() {
+        return jobController;
+    }
+
+    public InternalCommandManager getCommandManager() {
+        return commandManager;
+    }
 }

core/src/main/java/com/taobao/arthas/core/shell/system/Job.java

@@ -57,6 +57,11 @@ public interface Job {
      */
     Job resume();
 
+    /**
+     * @return true if the job is running in background
+     */
+    boolean isRunInBackground();
+
     /**
      * Send the job to background.
      *

core/src/main/java/com/taobao/arthas/core/shell/system/JobController.java

@@ -1,9 +1,11 @@
 package com.taobao.arthas.core.shell.system;
 
+import com.taobao.arthas.core.distribution.ResultDistributor;
 import com.taobao.arthas.core.shell.cli.CliToken;
 import com.taobao.arthas.core.shell.handlers.Handler;
-import com.taobao.arthas.core.shell.impl.ShellImpl;
+import com.taobao.arthas.core.shell.session.Session;
 import com.taobao.arthas.core.shell.system.impl.InternalCommandManager;
+import com.taobao.arthas.core.shell.term.Term;
 
 import java.util.List;
 import java.util.Set;
@@ -33,10 +35,13 @@ public interface JobController {
      *
      * @param commandManager command manager
      * @param tokens    the command tokens
-     * @param shell     the current shell
+     * @param session     the current session
+     * @param jobHandler  job event handler
+     * @param term     telnet term
+     * @param resultDistributor
      * @return the created job
      */
-    Job createJob(InternalCommandManager commandManager, List<CliToken> tokens, ShellImpl shell);
+    Job createJob(InternalCommandManager commandManager, List<CliToken> tokens, Session session, JobListener jobHandler, Term term, ResultDistributor resultDistributor);
 
     /**
      * Close the controller and terminate all the underlying jobs, a closed controller does not accept anymore jobs.

core/src/main/java/com/taobao/arthas/core/shell/system/JobListener.java

@@ -0,0 +1,16 @@
+package com.taobao.arthas.core.shell.system;
+
+/**
+ * Job listener
+ * @author gongdewei 2020-03-23
+ */
+public interface JobListener {
+
+    void onForeground(Job job);
+
+    void onBackground(Job job);
+
+    void onTerminated(Job job);
+
+    void onSuspend(Job job);
+}

core/src/main/java/com/taobao/arthas/core/shell/system/impl/GlobalJobControllerImpl.java

@@ -9,11 +9,15 @@ import java.util.concurrent.TimeUnit;
 import com.alibaba.arthas.deps.org.slf4j.Logger;
 import com.alibaba.arthas.deps.org.slf4j.LoggerFactory;
 import com.taobao.arthas.core.GlobalOptions;
+import com.taobao.arthas.core.distribution.ResultDistributor;
 import com.taobao.arthas.core.server.ArthasBootstrap;
 import com.taobao.arthas.core.shell.cli.CliToken;
 import com.taobao.arthas.core.shell.handlers.Handler;
-import com.taobao.arthas.core.shell.impl.ShellImpl;
+import com.taobao.arthas.core.shell.session.Session;
 import com.taobao.arthas.core.shell.system.Job;
+import com.taobao.arthas.core.shell.system.JobListener;
+import com.taobao.arthas.core.shell.term.Term;
+
 
 /**
  * 全局的Job Controller，不应该存在启停的概念，不需要在连接的断开时关闭，
@@ -49,8 +53,8 @@ public class GlobalJobControllerImpl extends JobControllerImpl {
     }
 
     @Override
-    public Job createJob(InternalCommandManager commandManager, List<CliToken> tokens, ShellImpl shell) {
-        final Job job = super.createJob(commandManager, tokens, shell);
+    public Job createJob(InternalCommandManager commandManager, List<CliToken> tokens, Session session, JobListener jobHandler, Term term, ResultDistributor resultDistributor) {
+        final Job job = super.createJob(commandManager, tokens, session, jobHandler, term, resultDistributor);
 
         /*
          * 达到超时时间将会停止job

core/src/main/java/com/taobao/arthas/core/shell/system/impl/JobControllerImpl.java

@@ -1,6 +1,7 @@
 package com.taobao.arthas.core.shell.system.impl;
 
 import com.taobao.arthas.core.GlobalOptions;
+import com.taobao.arthas.core.distribution.ResultDistributor;
 import com.taobao.arthas.core.shell.cli.CliToken;
 import com.taobao.arthas.core.shell.command.Command;
 import com.taobao.arthas.core.shell.command.internal.RedirectHandler;
@@ -8,33 +9,27 @@ import com.taobao.arthas.core.shell.command.internal.StdoutHandler;
 import com.taobao.arthas.core.shell.command.internal.TermHandler;
 import com.taobao.arthas.core.shell.future.Future;
 import com.taobao.arthas.core.shell.handlers.Handler;
-import com.taobao.arthas.core.shell.impl.ShellImpl;
+import com.taobao.arthas.core.shell.session.Session;
 import com.taobao.arthas.core.shell.system.Job;
 import com.taobao.arthas.core.shell.system.JobController;
+import com.taobao.arthas.core.shell.system.JobListener;
 import com.taobao.arthas.core.shell.system.Process;
 import com.taobao.arthas.core.shell.system.impl.ProcessImpl.ProcessOutput;
 import com.taobao.arthas.core.shell.term.Term;
 import com.taobao.arthas.core.util.Constants;
 import com.taobao.arthas.core.util.LogUtil;
 import com.taobao.arthas.core.util.TokenUtils;
-
 import io.termd.core.function.Function;
 
 import java.io.File;
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.List;
-import java.util.ListIterator;
-import java.util.Set;
-import java.util.SortedMap;
-import java.util.TreeMap;
+import java.util.*;
 import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
  * @author hengyunabc 2019-05-14
+ * @author gongdewei 2020-03-23
  */
 public class JobControllerImpl implements JobController {
 
@@ -58,16 +53,16 @@ public class JobControllerImpl implements JobController {
     }
 
     @Override
-    public Job createJob(InternalCommandManager commandManager, List<CliToken> tokens, ShellImpl shell) {
+    public Job createJob(InternalCommandManager commandManager, List<CliToken> tokens, Session session, JobListener jobHandler, Term term, ResultDistributor resultDistributor) {
         int jobId = idGenerator.incrementAndGet();
         StringBuilder line = new StringBuilder();
         for (CliToken arg : tokens) {
             line.append(arg.raw());
         }
         boolean runInBackground = runInBackground(tokens);
-        Process process = createProcess(tokens, commandManager, jobId, shell.term());
+        Process process = createProcess(tokens, commandManager, jobId, term, resultDistributor);
         process.setJobId(jobId);
-        JobImpl job = new JobImpl(jobId, this, process, line.toString(), runInBackground, shell);
+        JobImpl job = new JobImpl(jobId, this, process, line.toString(), runInBackground, session, jobHandler);
         jobs.put(jobId, job);
         return job;
     }
@@ -120,9 +115,10 @@ public class JobControllerImpl implements JobController {
      * @param commandManager command manager
      * @param jobId job id
      * @param term term
+     * @param resultDistributor
      * @return the created process
      */
-    private Process createProcess(List<CliToken> line, InternalCommandManager commandManager, int jobId, Term term) {
+    private Process createProcess(List<CliToken> line, InternalCommandManager commandManager, int jobId, Term term, ResultDistributor resultDistributor) {
         try {
             ListIterator<CliToken> tokens = line.listIterator();
             while (tokens.hasNext()) {
@@ -130,7 +126,7 @@ public class JobControllerImpl implements JobController {
                 if (token.isText()) {
                     Command command = commandManager.getCommand(token.value());
                     if (command != null) {
-                        return createCommandProcess(command, tokens, jobId, term);
+                        return createCommandProcess(command, tokens, jobId, term, resultDistributor);
                     } else {
                         throw new IllegalArgumentException(token.value() + ": command not found");
                     }
@@ -152,7 +148,7 @@ public class JobControllerImpl implements JobController {
         return runInBackground;
     }
 
-    private Process createCommandProcess(Command command, ListIterator<CliToken> tokens, int jobId, Term term) throws IOException {
+    private Process createCommandProcess(Command command, ListIterator<CliToken> tokens, int jobId, Term term, ResultDistributor resultDistributor) throws IOException {
         List<CliToken> remaining = new ArrayList<CliToken>();
         List<CliToken> pipelineTokens = new ArrayList<CliToken>();
         boolean isPipeline = false;
@@ -199,7 +195,9 @@ public class JobControllerImpl implements JobController {
             }
         }
         ProcessOutput ProcessOutput = new ProcessOutput(stdoutHandlerChain, cacheLocation, term);
-        return new ProcessImpl(command, remaining, command.processHandler(), ProcessOutput);
+        ProcessImpl process = new ProcessImpl(command, remaining, command.processHandler(), ProcessOutput, resultDistributor);
+        process.setTty(term);
+        return process;
     }
 
     private String getRedirectFileName(ListIterator<CliToken> tokens) {
@@ -226,4 +224,5 @@ public class JobControllerImpl implements JobController {
     public void close() {
         close(null);
     }
+
 }

core/src/main/java/com/taobao/arthas/core/shell/system/impl/JobImpl.java

@@ -7,20 +7,16 @@ import java.util.concurrent.atomic.AtomicBoolean;
 
 import com.taobao.arthas.core.shell.future.Future;
 import com.taobao.arthas.core.shell.handlers.Handler;
-import com.taobao.arthas.core.shell.handlers.shell.ShellForegroundUpdateHandler;
-import com.taobao.arthas.core.shell.impl.ShellImpl;
 import com.taobao.arthas.core.shell.session.Session;
 import com.taobao.arthas.core.shell.system.ExecStatus;
 import com.taobao.arthas.core.shell.system.Job;
+import com.taobao.arthas.core.shell.system.JobListener;
 import com.taobao.arthas.core.shell.system.Process;
-import com.taobao.arthas.core.shell.term.Term;
-import com.taobao.arthas.core.shell.term.impl.TermImpl;
-import com.taobao.arthas.core.util.Constants;
-import com.taobao.arthas.core.util.FileUtils;
 
 /**
  * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
  * @author hengyunabc 2019-05-14
+ * @author gongdewei 2020-03-23
  */
 public class JobImpl implements Job {
 
@@ -28,25 +24,30 @@ public class JobImpl implements Job {
     final JobControllerImpl controller;
     final Process process;
     final String line;
+    private volatile Session session;
     private volatile ExecStatus actualStatus; // Used internally for testing only
     volatile long lastStopped; // When the job was last stopped
-    volatile ShellImpl shell;
+    volatile JobListener jobHandler;
     volatile Handler<ExecStatus> statusUpdateHandler;
     volatile Date timeoutDate;
     final Future<Void> terminateFuture;
     final AtomicBoolean runInBackground;
-    final Handler<Job> foregroundUpdatedHandler;
+    //final Handler<Job> foregroundUpdatedHandler;
 
     JobImpl(int id, final JobControllerImpl controller, Process process, String line, boolean runInBackground,
-            ShellImpl shell) {
+            Session session, JobListener jobHandler) {
         this.id = id;
         this.controller = controller;
         this.process = process;
         this.line = line;
+        this.session = session;
         this.terminateFuture = Future.future();
         this.runInBackground = new AtomicBoolean(runInBackground);
-        this.shell = shell;
-        this.foregroundUpdatedHandler = new ShellForegroundUpdateHandler(shell);
+        this.jobHandler = jobHandler;
+        if (jobHandler == null) {
+            throw new IllegalArgumentException("JobListener is required");
+        }
+        //this.foregroundUpdatedHandler = new ShellForegroundUpdateHandler(shell);
         process.terminatedHandler(new TerminatedHandler(controller));
     }
 
@@ -76,7 +77,7 @@ public class JobImpl implements Job {
 
     @Override
     public Session getSession() {
-        return shell.session();
+        return session;
     }
 
     @Override
@@ -89,19 +90,21 @@ public class JobImpl implements Job {
 
         runInBackground.set(!foreground);
 
-        if (foreground) {
-            if (foregroundUpdatedHandler != null) {
-                foregroundUpdatedHandler.handle(this);
-            }
-        }
+//        if (foreground) {
+//            if (foregroundUpdatedHandler != null) {
+//                foregroundUpdatedHandler.handle(this);
+//            }
+//        }
         if (statusUpdateHandler != null) {
             statusUpdateHandler.handle(process.status());
         }
 
-        if (foreground) {
-            shell.setForegroundJob(this);
-        } else {
-            shell.setForegroundJob(null);
+        if (this.status() == ExecStatus.RUNNING) {
+            if (foreground) {
+                jobHandler.onForeground(this);
+            } else {
+                jobHandler.onBackground(this);
+            }
         }
         return this;
     }
@@ -113,14 +116,15 @@ public class JobImpl implements Job {
         } catch (IllegalStateException ignore) {
             return this;
         }
-        if (!runInBackground.get() && foregroundUpdatedHandler != null) {
-            foregroundUpdatedHandler.handle(null);
-        }
+//        if (!runInBackground.get() && foregroundUpdatedHandler != null) {
+//            foregroundUpdatedHandler.handle(null);
+//        }
         if (statusUpdateHandler != null) {
             statusUpdateHandler.handle(process.status());
         }
 
-        shell.setForegroundJob(null);
+//        shell.setForegroundJob(null);
+        jobHandler.onSuspend(this);
         return this;
     }
 
@@ -147,6 +151,11 @@ public class JobImpl implements Job {
         return line;
     }
 
+    @Override
+    public boolean isRunInBackground() {
+        return runInBackground.get();
+    }
+
     @Override
     public Job toBackground() {
         if (!this.runInBackground.get()) {
@@ -156,10 +165,12 @@ public class JobImpl implements Job {
                 if (statusUpdateHandler != null) {
                     statusUpdateHandler.handle(process.status());
                 }
+                jobHandler.onBackground(this);
             }
         }
 
-        shell.setForegroundJob(null);
+//        shell.setForegroundJob(null);
+//        jobHandler.onBackground(this);
         return this;
     }
 
@@ -167,15 +178,16 @@ public class JobImpl implements Job {
     public Job toForeground() {
         if (this.runInBackground.get()) {
             if (runInBackground.compareAndSet(true, false)) {
-                if (foregroundUpdatedHandler != null) {
-                    foregroundUpdatedHandler.handle(this);
-                }
+//                if (foregroundUpdatedHandler != null) {
+//                    foregroundUpdatedHandler.handle(this);
+//                }
                 process.toForeground();
                 if (statusUpdateHandler != null) {
                     statusUpdateHandler.handle(process.status());
                 }
 
-                shell.setForegroundJob(this);
+//                shell.setForegroundJob(this);
+                jobHandler.onForeground(this);
             }
         }
 
@@ -194,26 +206,34 @@ public class JobImpl implements Job {
 
     @Override
     public Job run(boolean foreground) {
-        if (foreground && foregroundUpdatedHandler != null) {
-            foregroundUpdatedHandler.handle(this);
-        }
+//        if (foreground && foregroundUpdatedHandler != null) {
+//            foregroundUpdatedHandler.handle(this);
+//        }
 
         actualStatus = ExecStatus.RUNNING;
         if (statusUpdateHandler != null) {
             statusUpdateHandler.handle(ExecStatus.RUNNING);
         }
-        process.setTty(shell.term());
-        process.setSession(shell.session());
+        //set process's tty in JobControllerImpl.createCommandProcess
+        //process.setTty(shell.term());
+        process.setSession(this.session);
         process.run(foreground);
 
-        if (!foreground && foregroundUpdatedHandler != null) {
-            foregroundUpdatedHandler.handle(null);
-        }
-
-        if (foreground) {
-            shell.setForegroundJob(this);
-        } else {
-            shell.setForegroundJob(null);
+//        if (!foreground && foregroundUpdatedHandler != null) {
+//            foregroundUpdatedHandler.handle(null);
+//        }
+//
+//        if (foreground) {
+//            shell.setForegroundJob(this);
+//        } else {
+//            shell.setForegroundJob(null);
+//        }
+        if (this.status() == ExecStatus.RUNNING) {
+            if (foreground) {
+                jobHandler.onForeground(this);
+            } else {
+                jobHandler.onBackground(this);
+            }
         }
         return this;
     }
@@ -228,24 +248,25 @@ public class JobImpl implements Job {
 
         @Override
         public void handle(Integer exitCode) {
-            if (!runInBackground.get() && actualStatus.equals(ExecStatus.RUNNING)) {
+//            if (!runInBackground.get() && actualStatus.equals(ExecStatus.RUNNING)) {
                 // 只有前台在运行的任务，才需要调用foregroundUpdateHandler
-                if (foregroundUpdatedHandler != null) {
-                    foregroundUpdatedHandler.handle(null);
-                }
-            }
+//                if (foregroundUpdatedHandler != null) {
+//                    foregroundUpdatedHandler.handle(null);
+//                }
+//            }
+            jobHandler.onTerminated(JobImpl.this);
             controller.removeJob(JobImpl.this.id);
             if (statusUpdateHandler != null) {
                 statusUpdateHandler.handle(ExecStatus.TERMINATED);
             }
             terminateFuture.complete();
 
-            // save command history
-            Term term = shell.term();
-            if (term instanceof TermImpl) {
-                List<int[]> history = ((TermImpl) term).getReadline().getHistory();
-                FileUtils.saveCommandHistory(history, new File(Constants.CMD_HISTORY_FILE));
-            }
+            // save command history (move to JobControllerImpl.ShellJobHandler.onTerminated)
+//            Term term = shell.term();
+//            if (term instanceof TermImpl) {
+//                List<int[]> history = ((TermImpl) term).getReadline().getHistory();
+//                FileUtils.saveCommandHistory(history, new File(Constants.CMD_HISTORY_FILE));
+//            }
         }
     }
 

core/src/main/java/com/taobao/arthas/core/shell/system/impl/ProcessImpl.java

@@ -4,6 +4,11 @@ import com.alibaba.arthas.deps.org.slf4j.Logger;
 import com.alibaba.arthas.deps.org.slf4j.LoggerFactory;
 import com.taobao.arthas.core.advisor.AdviceListener;
 import com.taobao.arthas.core.advisor.AdviceWeaver;
+import com.taobao.arthas.core.command.basic1000.HelpCommand;
+import com.taobao.arthas.core.command.model.ResultModel;
+import com.taobao.arthas.core.command.model.StatusModel;
+import com.taobao.arthas.core.distribution.ResultDistributor;
+import com.taobao.arthas.core.distribution.impl.TermResultDistributorImpl;
 import com.taobao.arthas.core.server.ArthasBootstrap;
 import com.taobao.arthas.core.shell.cli.CliToken;
 import com.taobao.arthas.core.shell.command.Command;
@@ -16,12 +21,8 @@ import com.taobao.arthas.core.shell.system.ExecStatus;
 import com.taobao.arthas.core.shell.system.Process;
 import com.taobao.arthas.core.shell.system.ProcessAware;
 import com.taobao.arthas.core.shell.term.Tty;
-import com.taobao.arthas.core.util.usage.StyledUsageFormatter;
 import com.taobao.middleware.cli.CLIException;
 import com.taobao.middleware.cli.CommandLine;
-import com.taobao.middleware.cli.UsageMessageFormatter;
-import com.taobao.text.Color;
-
 import io.termd.core.function.Function;
 
 import java.lang.instrument.ClassFileTransformer;
@@ -32,6 +33,7 @@ import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * @author beiwei30 on 10/11/2016.
+ * @author gongdewei 2020-03-26
  */
 public class ProcessImpl implements Process {
 
@@ -55,16 +57,18 @@ public class ProcessImpl implements Process {
     private Handler<String> stdinHandler;
     private Handler<Void> resizeHandler;
     private Integer exitCode;
-    private CommandProcess process;
+    private CommandProcessImpl process;
     private Date startTime;
     private ProcessOutput processOutput;
     private int jobId;
+    private ResultDistributor resultDistributor;
 
     public ProcessImpl(Command commandContext, List<CliToken> args, Handler<CommandProcess> handler,
-                       ProcessOutput processOutput) {
+                       ProcessOutput processOutput, ResultDistributor resultDistributor) {
         this.commandContext = commandContext;
         this.handler = handler;
         this.args = args;
+        this.resultDistributor = resultDistributor;
         this.processStatus = ExecStatus.READY;
         this.processOutput = processOutput;
     }
@@ -236,13 +240,15 @@ public class ProcessImpl implements Process {
 
     @Override
     public void terminate(Handler<Void> completionHandler) {
-        if (!terminate(-10, completionHandler)) {
+        if (!terminate(-10, completionHandler, null)) {
             throw new IllegalStateException("Cannot terminate terminated process");
         }
     }
 
-    private synchronized boolean terminate(int exitCode, Handler<Void> completionHandler) {
+    private synchronized boolean terminate(int exitCode, Handler<Void> completionHandler, String message) {
         if (processStatus != ExecStatus.TERMINATED) {
+            //add status message
+            this.appendResult(new StatusModel(exitCode, message));
             if (process != null) {
                 processOutput.close();
             }
@@ -256,6 +262,13 @@ public class ProcessImpl implements Process {
         }
     }
 
+    private void appendResult(ResultModel result) {
+        result.setJobId(jobId);
+        if (resultDistributor != null) {
+            resultDistributor.appendResult(result);
+        }
+    }
+
     private void updateStatus(ExecStatus statusUpdate, Integer exitCodeUpdate, boolean foregroundUpdate,
                               Handler<Void> handler, Handler<Integer> terminatedHandler,
                               Handler<Void> completionHandler) {
@@ -320,6 +333,11 @@ public class ProcessImpl implements Process {
             throw new IllegalStateException("Cannot execute process without a TTY set");
         }
 
+        process = new CommandProcessImpl(this, tty);
+        if (resultDistributor == null) {
+            resultDistributor = new TermResultDistributorImpl(process);
+        }
+
         final List<String> args2 = new LinkedList<String>();
         for (CliToken arg : args) {
             if (arg.isText()) {
@@ -331,25 +349,20 @@ public class ProcessImpl implements Process {
         try {
             if (commandContext.cli() != null) {
                 if (commandContext.cli().parse(args2, false).isAskingForHelp()) {
-                    UsageMessageFormatter formatter = new StyledUsageFormatter(Color.green);
-                    formatter.setWidth(tty.width());
-                    StringBuilder usage = new StringBuilder();
-                    commandContext.cli().usage(usage, formatter);
-                    usage.append('\n');
-                    tty.write(usage.toString());
+                    appendResult(new HelpCommand().createHelpDetailModel(commandContext));
                     terminate();
                     return;
                 }
 
                 cl = commandContext.cli().parse(args2);
+                process.setArgs2(args2);
+                process.setCommandLine(cl);
             }
         } catch (CLIException e) {
-            tty.write(e.getMessage() + "\n");
-            terminate();
+            terminate(-10, null, e.getMessage());
             return;
         }
 
-        process = new CommandProcessImpl(this, args2, tty, cl);
         if (cacheLocation() != null) {
             process.echoTips("job id  : " + this.jobId + "\n");
             process.echoTips("cache location  : " + cacheLocation() + "\n");
@@ -372,27 +385,25 @@ public class ProcessImpl implements Process {
                 handler.handle(process);
             } catch (Throwable t) {
                 logger.error("Error during processing the command:", t);
-                process.write("Error during processing the command, exception type: " + t.getClass().getName() + ", message:" + t.getMessage()
-                        + ", please check $HOME/logs/arthas/arthas.log for more details. \n");
-                terminate(1, null);
+                process.end(1, "Error during processing the command: " + t.getClass().getName() + ", message:" + t.getMessage()
+                        + ", please check $HOME/logs/arthas/arthas.log for more details." );
             }
         }
     }
 
     private class CommandProcessImpl implements CommandProcess {
+
         private final Process process;
-        private final List<String> args2;
         private final Tty tty;
-        private final CommandLine commandLine;
+        private List<String> args2;
+        private CommandLine commandLine;
         private AtomicInteger times = new AtomicInteger();
         private AdviceListener listener = null;
         private ClassFileTransformer transformer;
 
-        public CommandProcessImpl(Process process, List<String> args2, Tty tty, CommandLine commandLine) {
+        public CommandProcessImpl(Process process, Tty tty) {
             this.process = process;
-            this.args2 = args2;
             this.tty = tty;
-            this.commandLine = commandLine;
         }
 
         @Override
@@ -440,6 +451,14 @@ public class ProcessImpl implements Process {
             return times;
         }
 
+        public void setArgs2(List<String> args2) {
+            this.args2 = args2;
+        }
+
+        public void setCommandLine(CommandLine commandLine) {
+            this.commandLine = commandLine;
+        }
+
         @Override
         public CommandProcess stdinHandler(Handler<String> handler) {
             stdinHandler = handler;
@@ -578,13 +597,27 @@ public class ProcessImpl implements Process {
 
         @Override
         public void end(int statusCode) {
-            terminate(statusCode, null);
+            end(statusCode, null);
+        }
+
+        @Override
+        public void end(int statusCode, String message) {
+            terminate(statusCode, null, message);
         }
 
         @Override
         public boolean isRunning() {
             return processStatus == ExecStatus.RUNNING;
         }
+
+        @Override
+        public void appendResult(ResultModel result) {
+            if (processStatus != ExecStatus.RUNNING) {
+                throw new IllegalStateException(
+                        "Cannot write to standard output when " + status().name().toLowerCase());
+            }
+            ProcessImpl.this.appendResult(result);
+        }
     }
 
     static class ProcessOutput {
@@ -620,7 +653,10 @@ public class ProcessImpl implements Process {
 
         private void write(String data) {
             if (stdoutHandlerChain != null) {
-                for (Function<String, String> function : stdoutHandlerChain) {
+                //hotspot, reduce memory fragment (foreach/iterator)
+                int size = stdoutHandlerChain.size();
+                for (int i = 0; i < size; i++) {
+                    Function<String, String> function = stdoutHandlerChain.get(i);
                     data = function.apply(data);
                 }
             }