microsoft · testforstephen · Apr 12, 2019 · Mar 13, 2019 · Mar 18, 2019 · Apr 3, 2019
diff --git a/...t.java.debug.core/src/main/java/com/microsoft/java/debug/core/adapter/ProcessConsole.java b/...t.java.debug.core/src/main/java/com/microsoft/java/debug/core/adapter/ProcessConsole.java
@@ -1,5 +1,5 @@
 /*******************************************************************************
-* Copyright (c) 2017 Microsoft Corporation and others.
+* Copyright (c) 2017-2019 Microsoft Corporation and others.
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * which accompanies this distribution, and is available at
@@ -17,18 +17,18 @@
 import java.io.InputStreamReader;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
+import java.util.stream.Stream;
 
-import io.reactivex.functions.Consumer;
+import com.microsoft.java.debug.core.protocol.Events.OutputEvent.Category;
+
+import io.reactivex.Observable;
+import io.reactivex.schedulers.Schedulers;
 import io.reactivex.subjects.PublishSubject;
 
 public class ProcessConsole {
-    private Process process;
-    private String name;
-    private Charset encoding;
-    private PublishSubject<String> stdoutSubject = PublishSubject.<String>create();
-    private PublishSubject<String> stderrSubject = PublishSubject.<String>create();
-    private Thread stdoutThread = null;
-    private Thread stderrThread = null;
+    private InputStreamObservable stdoutStream;
+    private InputStreamObservable stderrStream;
+    private Observable<ConsoleMessage> observable = null;
 
     public ProcessConsole(Process process) {
         this(process, "Process", StandardCharsets.UTF_8);
@@ -44,76 +44,126 @@ public ProcessConsole(Process process) {
      *              the process encoding format
      */
     public ProcessConsole(Process process, String name, Charset encoding) {
-        this.process = process;
-        this.name = name;
-        this.encoding = encoding;
+        this.stdoutStream = new InputStreamObservable(name + " Stdout Handler", process.getInputStream(), encoding);
+        this.stderrStream = new InputStreamObservable(name + " Stderr Handler", process.getErrorStream(), encoding);
+        Observable<ConsoleMessage> stdout = this.stdoutStream.messages().map((message) -> new ConsoleMessage(message, Category.stdout));
+        Observable<ConsoleMessage> stderr = this.stderrStream.messages().map((message) -> new ConsoleMessage(message, Category.stderr));
+        this.observable = Observable.mergeArrayDelayError(stdout, stderr).observeOn(Schedulers.newThread());
     }
 
     /**
-     * Start two separate threads to monitor the messages from stdout and stderr streams of the target process.
+     * Start monitoring the stdout/stderr streams of the target process.
      */
     public void start() {
-        this.stdoutThread = new Thread(this.name + " Stdout Handler") {
-            public void run() {
-                monitor(process.getInputStream(), stdoutSubject);
-            }
-        };
-        stdoutThread.setDaemon(true);
-        stdoutThread.start();
-
-        this.stderrThread = new Thread(this.name + " Stderr Handler") {
-            public void run() {
-                monitor(process.getErrorStream(), stderrSubject);
-            }
-        };
-        stderrThread.setDaemon(true);
-        stderrThread.start();
+        stdoutStream.start();
+        stderrStream.start();
     }
 
     /**
-     * Stop the process console handlers.
+     * Stop monitoring the process console.
      */
     public void stop() {
-        if (this.stdoutThread != null) {
-            this.stdoutThread.interrupt();
-            this.stdoutThread = null;
-        }
+        stdoutStream.stop();
+        stderrStream.stop();
+    }
 
-        if (this.stderrThread != null) {
-            this.stderrThread.interrupt();
-            this.stderrThread = null;
-        }
+    public Observable<ConsoleMessage> messages() {
+        return observable;
     }
 
-    public void onStdout(Consumer<String> callback) {
-        stdoutSubject.subscribe(callback);
+    public Observable<ConsoleMessage> stdoutMessages() {
+        return this.messages().filter((message) -> message.category == Category.stdout);
     }
 
-    public void onStderr(Consumer<String> callback) {
-        stderrSubject.subscribe(callback);
+    public Observable<ConsoleMessage> stderrMessages() {
+        return this.messages().filter((message) -> message.category == Category.stderr);
     }
 
-    private void monitor(InputStream input, PublishSubject<String> subject) {
-        BufferedReader reader = new BufferedReader(new InputStreamReader(input, encoding));
-        final int BUFFERSIZE = 4096;
-        char[] buffer = new char[BUFFERSIZE];
-        while (true) {
-            try {
-                if (Thread.interrupted()) {
-                    subject.onComplete();
-                    return;
+    /**
+     * Split the stdio message to lines, and return them as a new Observable.
+     */
+    public Observable<ConsoleMessage> lineMessages() {
+        return this.messages().map((message) -> {
+            String[] lines = message.output.split("(?<=\n)");
+            return Stream.of(lines).map((line) -> new ConsoleMessage(line, message.category)).toArray(ConsoleMessage[]::new);
+        }).concatMap((lines) -> Observable.fromArray(lines));
+    }
+
+    public static class InputStreamObservable {
+        private PublishSubject<String> rxSubject = PublishSubject.<String>create();
+        private String name;
+        private InputStream inputStream;
+        private Charset encoding;
+        private Thread loopingThread;
+
+        /**
+         * Constructor.
+         */
+        public InputStreamObservable(String name, InputStream inputStream, Charset encoding) {
+            this.name = name;
+            this.inputStream = inputStream;
+            this.encoding = encoding;
+        }
+
+        /**
+         * Starts the stream.
+         */
+        public void start() {
+            loopingThread = new Thread(name) {
+                public void run() {
+                    monitor(inputStream, rxSubject);
                 }
-                int read = reader.read(buffer, 0, BUFFERSIZE);
-                if (read == -1) {
-                    subject.onComplete();
+            };
+            loopingThread.setDaemon(true);
+            loopingThread.start();
+        }
+
+        /**
+         * Stops the stream.
+         */
+        public void stop() {
+            if (loopingThread != null) {
+                loopingThread.interrupt();
+                loopingThread = null;
+            }
+        }
+
+        private void monitor(InputStream input, PublishSubject<String> subject) {
+            BufferedReader reader = new BufferedReader(new InputStreamReader(input, encoding));
+            final int BUFFERSIZE = 4096;
+            char[] buffer = new char[BUFFERSIZE];
+            while (true) {
+                try {
+                    if (Thread.interrupted()) {
+                        subject.onComplete();
+                        return;
+                    }
+                    int read = reader.read(buffer, 0, BUFFERSIZE);
+                    if (read == -1) {
+                        subject.onComplete();
+                        return;
+                    }
+
+                    subject.onNext(new String(buffer, 0, read));
+                } catch (IOException e) {
+                    subject.onError(e);
                     return;
                 }
-
-                subject.onNext(new String(buffer, 0, read));
-            } catch (IOException e) {
-                subject.onError(e);
-                return;
             }
         }
+
+        public Observable<String> messages() {
+            return rxSubject;
+        }
+    }
+
+    public static class ConsoleMessage {
+        public String output;
+        public Category category;
+
+        public ConsoleMessage(String message, Category category) {
+            this.output = message;
+            this.category = category;
+        }
     }
 }
diff --git a/...n/java/com/microsoft/java/debug/core/adapter/handler/ConfigurationDoneRequestHandler.java b/...n/java/com/microsoft/java/debug/core/adapter/handler/ConfigurationDoneRequestHandler.java
@@ -79,14 +79,7 @@ private void handleDebugEvent(DebugEvent debugEvent, IDebugSession debugSession,
             context.setVmTerminated();
             context.getProtocolServer().sendEvent(new Events.ExitedEvent(0));
         } else if (event instanceof VMDisconnectEvent) {
-            context.setVmTerminated();
-            context.getProtocolServer().sendEvent(new Events.TerminatedEvent());
-            // Terminate eventHub thread.
-            try {
-                debugSession.getEventHub().close();
-            } catch (Exception e) {
-                // do nothing.
-            }
+            // ignore since LaunchRequestHandler has already handled.
         } else if (event instanceof ThreadStartEvent) {
             ThreadReference startThread = ((ThreadStartEvent) event).thread();
             Events.ThreadEvent threadEvent = new Events.ThreadEvent("started", startThread.uniqueID());

diff --git a/...bug.core/src/main/java/com/microsoft/java/debug/core/adapter/handler/ILaunchDelegate.java b/...bug.core/src/main/java/com/microsoft/java/debug/core/adapter/handler/ILaunchDelegate.java
@@ -11,11 +11,14 @@
 
 package com.microsoft.java.debug.core.adapter.handler;
 
+import java.io.IOException;
 import java.util.concurrent.CompletableFuture;
 
 import com.microsoft.java.debug.core.adapter.IDebugAdapterContext;
 import com.microsoft.java.debug.core.protocol.Messages.Response;
 import com.microsoft.java.debug.core.protocol.Requests.LaunchArguments;
+import com.sun.jdi.connect.IllegalConnectorArgumentsException;
+import com.sun.jdi.connect.VMStartException;
 
 public interface ILaunchDelegate {
     void postLaunch(LaunchArguments launchArguments, IDebugAdapterContext context);
@@ -24,6 +27,7 @@ public interface ILaunchDelegate {

     CompletableFuture<Response> launchInTerminal(LaunchArguments launchArguments, Response response, IDebugAdapterContext context);
 
-    CompletableFuture<Response> launchInternally(LaunchArguments launchArguments, Response response, IDebugAdapterContext context);
+    Process launch(LaunchArguments launchArguments, IDebugAdapterContext context)
+            throws IOException, IllegalConnectorArgumentsException, VMStartException;
 
 }