Introduce the process visiable scheduler for IDEA rxjava.

Signed-off-by: Wei Zhang <[email protected]>

Author: wezhang

PluginsAndFeatures/azure-toolkit-for-intellij/src/com/microsoft/azure/hdinsight/spark/run/SparkBatchJobRemoteProcess.java

@@ -156,23 +156,17 @@ public Optional<Subscription> getJobSubscription() {
     public void start() {
         // Build, deploy and wait for the job done.
         jobSubscription = SparkSubmitHelper.getInstance().buildArtifact(project, submitModel.isLocalArtifact(), submitModel.getArtifact())
-                .flatMap(artifact -> {
-                    logInfo("Deploy the jar file into cluster...");
-
-                    return JobUtils.deployArtifact(
-                                submitModel.getArtifactPath(artifact.getName())
-                                        .orElseThrow(() -> propagate(new SparkJobException("Can't find jar path to upload"))),
-                                submitModel.getSubmissionParameter().getClusterName(),
-                                ctrlSubject)
-                        .subscribeOn(IdeaSchedulers.processBarVisable());
-//                        .subscribeOn(Schedulers.io());
-                })
+                .flatMap(artifact -> JobUtils.deployArtifact(
+                            submitModel.getArtifactPath(artifact.getName())
+                                    .orElseThrow(() -> propagate(new SparkJobException("Can't find jar path to upload"))),
+                            submitModel.getSubmissionParameter().getClusterName(),
+                            ctrlSubject)
+                    .subscribeOn(IdeaSchedulers.processBarVisibleAsync(project, "Deploy the jar file into cluster")))
                 .flatMap(clusterArtifactUriPair -> {
-                    logInfo("The Spark job is submitting ...");
-
                     IClusterDetail cluster = clusterArtifactUriPair.getKey();
                     submitModel.getSubmissionParameter().setFilePath(clusterArtifactUriPair.getValue());
-                    return JobUtils.submit(cluster, submitModel.getSubmissionParameter()).subscribeOn(Schedulers.io());
+                    return JobUtils.submit(cluster, submitModel.getSubmissionParameter())
+                            .subscribeOn(IdeaSchedulers.processBarVisibleAsync(project, "Submit the Spark batch job"));
                 })
                 .doOnSuccess(job -> {
                     getEventSubject().onNext(new SparkBatchJobSubmissionEvent(
@@ -201,15 +195,15 @@ public void start() {
                             try {
                                 final String state = job.getState();
 
-                                if (state.equals("starting") || state.equals("not_started")) {
+                                if (state.equals("starting") || state.equals("not_started") || state.equals("running")) {
                                     logInfo("Job is waiting for start due to cluster busy, please wait or disconnect (The job will run when the cluster is free).");
 
                                     return Observable.timer(5, TimeUnit.SECONDS);
                                 }
                             } catch (IOException ignored) {
                             }
 
-                            return Observable.error(new SparkJobException("Spark Job Service not available, please check HDInsight cluster status."));
+                            return Observable.error(new SparkJobException("Spark Job Service not available, please check HDInsight cluster status.", err));
                         })))
                 .flatMap(runningJob -> runningJob.getJobDoneObservable().subscribeOn(Schedulers.io()))
                 .subscribe(sdPair -> {

PluginsAndFeatures/azure-toolkit-for-intellij/src/com/microsoft/intellij/rxjava/ApplicationManagerInvokeAndWaitExecutor.java

@@ -22,11 +22,29 @@
 
 package com.microsoft.intellij.rxjava;
 
+import com.intellij.openapi.application.ApplicationManager;
+import com.intellij.openapi.progress.ProgressIndicator;
+import com.intellij.openapi.progress.ProgressManager;
+import com.intellij.openapi.progress.Task;
+import com.intellij.openapi.progress.impl.BackgroundableProcessIndicator;
+import com.intellij.openapi.project.Project;
+import org.jetbrains.annotations.NotNull;
 import rx.Scheduler;
 import rx.schedulers.Schedulers;
 
 public class IdeaSchedulers {
-    public static Scheduler processBarVisable() {
-        return Schedulers.from(new ApplicationManagerInvokeAndWaitExecutor());
+    public static Scheduler processBarVisibleAsync(@NotNull Project project, @NotNull String title) {
+        return Schedulers.from(command -> ApplicationManager.getApplication().invokeLater(() -> {
+            final Task.Backgroundable task = new Task.Backgroundable(project, title) {
+                @Override
+                public void run(@NotNull ProgressIndicator progressIndicator) {
+                    command.run();
+                }
+            };
+
+            final ProgressIndicator progressIndicator = new BackgroundableProcessIndicator(task);
+
+            ProgressManager.getInstance().runProcessWithProgressAsynchronously(task, progressIndicator);
+        }));
     }
 }

PluginsAndFeatures/azure-toolkit-for-intellij/src/com/microsoft/intellij/rxjava/IdeaSchedulers.java

@@ -22,6 +22,7 @@
 
 package com.microsoft.azure.hdinsight.sdk.rest.livy.interactive.api;
 
+import com.fasterxml.jackson.annotation.JsonInclude;
 import com.microsoft.azure.hdinsight.sdk.rest.IConvertible;
 import com.microsoft.azure.hdinsight.sdk.rest.livy.interactive.SessionKind;
 
@@ -45,6 +46,7 @@
  * Response Type
  *   @see com.microsoft.azure.hdinsight.sdk.rest.livy.interactive.Session
  */
+@JsonInclude(JsonInclude.Include.NON_NULL)
 public class PostSessions implements IConvertible {
     private SessionKind         kind;                       // The session kind (required)
     private String              proxyUser;                  // User to impersonate when starting the session

Utils/hdinsight-node-common/src/com/microsoft/azure/hdinsight/sdk/rest/livy/interactive/api/PostSessions.java

@@ -532,6 +532,8 @@ public static String uploadFileToHDFS(@NotNull IClusterDetail selectedClusterDet
         String sessionName = "Helper session to upload " + destUri.toString();
         URI livyUri = getLivyBaseUri(selectedClusterDetail);
 
+        logSubject.onNext(new SimpleImmutableEntry<>(Info, "Create Spark helper interactive session..."));
+
         return Observable.using(() -> new SparkSession(sessionName, livyUri, username, password),
                                 SparkSession::create,
                                 SparkSession::close)
@@ -544,7 +546,7 @@ public static String uploadFileToHDFS(@NotNull IClusterDetail selectedClusterDet
                     try {
                         inFile = new BufferedInputStream(new FileInputStream(srcJarFile));
 
-                        logSubject.onNext(new SimpleImmutableEntry<>(Info, String.format("Uploading to %s...", destUri)));
+                        logSubject.onNext(new SimpleImmutableEntry<>(Info, String.format("Uploading %s...", srcJarFile)));
                         IOUtils.copy(inFile, base64Enc);
 
                         inFile.close();