Bug 546641 - [console] ProcessConsole InputReadJob is not cancelableI20190425-0030I20190424-1800

Change-Id: I453380da668bc5d04e2f90469a16d2ce37d1e90d
Signed-off-by: Paul Pazderski <paul-eclipse@ppazderski.de>
Signed-off-by: Andrey Loskutov <loskutov@gmx.de>

4 files changed, 327 insertions, 5 deletions

diff --git a/org.eclipse.debug.tests/src/org/eclipse/debug/tests/AutomatedSuite.java b/org.eclipse.debug.tests/src/org/eclipse/debug/tests/AutomatedSuite.java
index b8b816930..7d873562c 100644
--- a/org.eclipse.debug.tests/src/org/eclipse/debug/tests/AutomatedSuite.java
+++ b/org.eclipse.debug.tests/src/org/eclipse/debug/tests/AutomatedSuite.java
@@ -19,6 +19,7 @@ import org.eclipse.debug.tests.breakpoint.BreakpointOrderingTests;
 import org.eclipse.debug.tests.console.ConsoleDocumentAdapterTests;
 import org.eclipse.debug.tests.console.ConsoleManagerTests;
 import org.eclipse.debug.tests.console.ConsoleTests;
+import org.eclipse.debug.tests.console.ProcessConsoleTests;
 import org.eclipse.debug.tests.launching.AcceleratorSubstitutionTests;
 import org.eclipse.debug.tests.launching.ArgumentParsingTests;
 import org.eclipse.debug.tests.launching.LaunchConfigurationTests;
@@ -111,6 +112,7 @@ public class AutomatedSuite extends TestSuite {
 		addTest(new TestSuite(ConsoleDocumentAdapterTests.class));
 		addTest(new TestSuite(ConsoleManagerTests.class));
 		addTest(new TestSuite(ConsoleTests.class));
+		addTest(new TestSuite(ProcessConsoleTests.class));
 
 		// Launch Groups
 		addTest(new TestSuite(LaunchGroupTests.class));
diff --git a/org.eclipse.debug.tests/src/org/eclipse/debug/tests/console/MockProcess.java b/org.eclipse.debug.tests/src/org/eclipse/debug/tests/console/MockProcess.java
new file mode 100644
index 000000000..a07a5059c
--- /dev/null
+++ b/org.eclipse.debug.tests/src/org/eclipse/debug/tests/console/MockProcess.java
@@ -0,0 +1,197 @@
+/*******************************************************************************
+ * Copyright (c) 2019 Paul Pazderski and others.
+ *
+ * This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License 2.0
+ * which accompanies this distribution, and is available at
+ * https://www.eclipse.org/legal/epl-2.0/
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ *
+ * Contributors:
+ *     Paul Pazderski - initial API and implementation
+ *******************************************************************************/
+package org.eclipse.debug.tests.console;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * A mockup process which can either simulate generation of output or wait for
+ * input to read.
+ */
+public class MockProcess extends Process {
+	/**
+	 * Use as run time parameter if mockup process should not terminate until
+	 * {@link #destroy()} is used.
+	 */
+	public static final int RUN_FOREVER = -1;
+
+	/** Mockup processe's standard streams. */
+	private final ByteArrayOutputStream stdin = new ByteArrayOutputStream();
+	private final InputStream stdout;
+	private final InputStream stderr;
+
+	/** Lock used in {@link #waitFor()}. */
+	private final Object waitForTerminationLock = new Object();
+	/**
+	 * Store number of bytes received which are not buffered anymore (i.e. those
+	 * input was already passed through {@link #getReceivedInput()}).
+	 */
+	private AtomicInteger receivedInput = new AtomicInteger(0);
+	/**
+	 * The time (in epoch milliseconds) when the mockup process terminates.
+	 * <p>
+	 * If this value is in the future it is the processes timeout. If it is in
+	 * the past it is the processes termination time. If it is <code>-1</code>
+	 * the process does not terminate and must be stopped using
+	 * {@link #destroy()}.
+	 * </p>
+	 */
+	private long endTime;
+
+	/**
+	 * Create new silent mockup process which runs for a given amount of time.
+	 * Does not read input or produce any output.
+	 *
+	 * @param runTimeMs runtime of the mockup process in milliseconds. If
+	 *            <code>0</code> the process terminates immediately. A
+	 *            <i>negative</i> value means the mockup process never
+	 *            terminates and must stopped with {@link #destroy()}.
+	 */
+	public MockProcess(long runTimeMs) {
+		this(null, null, runTimeMs);
+	}
+
+	/**
+	 * Create new mockup process and feed standard output streams with given
+	 * content.
+	 *
+	 * @param stdout mockup process standard output stream. May be
+	 *            <code>null</code>.
+	 * @param stderr mockup process standard error stream. May be
+	 *            <code>null</code>.
+	 * @param runTimeMs runtime of the mockup process in milliseconds. If
+	 *            <code>0</code> the process terminates immediately. A
+	 *            <i>negative</i> value means the mockup process never
+	 *            terminates and must stopped with {@link #destroy()}.
+	 */
+	public MockProcess(InputStream stdout, InputStream stderr, long runTimeMs) {
+		super();
+		this.stdout = (stdout != null ? stdout : new ByteArrayInputStream(new byte[0]));
+		this.stderr = (stderr != null ? stderr : new ByteArrayInputStream(new byte[0]));
+		this.endTime = runTimeMs < 0 ? RUN_FOREVER : System.currentTimeMillis() + runTimeMs;
+	}
+
+	/**
+	 * Create new mockup process and wait for input on standard input stream.
+	 * The mockup process terminates after receiving the given amount of data or
+	 * after it's timeout.
+	 *
+	 * @param expectedInputSize number of bytes to receive before termination
+	 * @param timeoutMs mockup process will be stopped after given amount of
+	 *            milliseconds. If <i>negative</i> timeout is disabled.
+	 */
+	public MockProcess(final int expectedInputSize, long timeoutMs) {
+		super();
+		this.stdout = new ByteArrayInputStream(new byte[0]);
+		this.stderr = new ByteArrayInputStream(new byte[0]);
+		this.endTime = (timeoutMs > 0 ? System.currentTimeMillis() + timeoutMs : RUN_FOREVER);
+
+		final Thread inputMonitor = new Thread(() -> {
+			while (!MockProcess.this.isTerminated()) {
+				synchronized (waitForTerminationLock) {
+					if (receivedInput.get() + stdin.size() >= expectedInputSize) {
+						endTime = System.currentTimeMillis();
+						waitForTerminationLock.notifyAll();
+						break;
+					}
+				}
+				try {
+					Thread.sleep(50);
+				} catch (InterruptedException e) {
+					break;
+				}
+			}
+		}, "Mockup Process Input Monitor");
+		inputMonitor.setDaemon(true);
+		inputMonitor.start();
+	}
+
+	/**
+	 * Get bytes received through stdin since last invocation of this method.
+	 * <p>
+	 * Not thread safe. It may miss some input if new content is written while
+	 * this method is executed.
+	 * </p>
+	 *
+	 * @return standard input since last invocation
+	 */
+	public synchronized byte[] getReceivedInput() {
+		final byte[] content = stdin.toByteArray();
+		stdin.reset();
+		receivedInput.addAndGet(content.length);
+		return content;
+	}
+
+	@Override
+	public OutputStream getOutputStream() {
+		return stdin;
+	}
+
+	@Override
+	public InputStream getInputStream() {
+		return stdout;
+	}
+
+	@Override
+	public InputStream getErrorStream() {
+		return stderr;
+	}
+
+	@Override
+	public int waitFor() throws InterruptedException {
+		synchronized (waitForTerminationLock) {
+			while (!isTerminated()) {
+				if (endTime == RUN_FOREVER) {
+					waitForTerminationLock.wait();
+				} else {
+					final long waitTime = endTime - System.currentTimeMillis();
+					if (waitTime > 0) {
+						waitForTerminationLock.wait(waitTime);
+					}
+				}
+			}
+		}
+		return 0;
+	}
+
+	@Override
+	public int exitValue() {
+		if (!isTerminated()) {
+			final String end = (endTime == RUN_FOREVER ? "never." : "in " + (endTime - System.currentTimeMillis()) + " ms.");
+			throw new IllegalThreadStateException("Mockup process terminates " + end);
+		}
+		return 0;
+	}
+
+	@Override
+	public void destroy() {
+		synchronized (waitForTerminationLock) {
+			endTime = System.currentTimeMillis();
+			waitForTerminationLock.notifyAll();
+		}
+	}
+
+	/**
+	 * Check if this process is already terminated.
+	 *
+	 * @return <code>true</code> if process is terminated
+	 */
+	private boolean isTerminated() {
+		return endTime != RUN_FOREVER && System.currentTimeMillis() > endTime;
+	}
+}
diff --git a/org.eclipse.debug.tests/src/org/eclipse/debug/tests/console/ProcessConsoleTests.java b/org.eclipse.debug.tests/src/org/eclipse/debug/tests/console/ProcessConsoleTests.java
new file mode 100644
index 000000000..4468db1ac
--- /dev/null
+++ b/org.eclipse.debug.tests/src/org/eclipse/debug/tests/console/ProcessConsoleTests.java
@@ -0,0 +1,101 @@
+/*******************************************************************************
+ * Copyright (c) 2019 Paul Pazderski and others.
+ *
+ * This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License 2.0
+ * which accompanies this distribution, and is available at
+ * https://www.eclipse.org/legal/epl-2.0/
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ *
+ * Contributors:
+ *     Paul Pazderski - initial API and implementation
+ *******************************************************************************/
+package org.eclipse.debug.tests.console;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.eclipse.core.runtime.ILogListener;
+import org.eclipse.core.runtime.IStatus;
+import org.eclipse.core.runtime.Platform;
+import org.eclipse.core.runtime.jobs.Job;
+import org.eclipse.debug.core.DebugPlugin;
+import org.eclipse.debug.core.ILaunchManager;
+import org.eclipse.debug.core.Launch;
+import org.eclipse.debug.core.model.IProcess;
+import org.eclipse.debug.tests.AbstractDebugTest;
+import org.eclipse.debug.tests.TestUtil;
+import org.eclipse.debug.ui.console.ConsoleColorProvider;
+import org.eclipse.ui.console.ConsolePlugin;
+
+/**
+ * Tests the ProcessConsole.
+ */
+public class ProcessConsoleTests extends AbstractDebugTest {
+	/**
+	 * Number of received log messages with severity error while running a
+	 * single test method.
+	 */
+	private final AtomicInteger loggedErrors = new AtomicInteger(0);
+
+	/** Listener to count error messages in {@link ConsolePlugin} log. */
+	private final ILogListener errorLogListener = new ILogListener() {
+		@Override
+		public void logging(IStatus status, String plugin) {
+			if (status.matches(IStatus.ERROR)) {
+				loggedErrors.incrementAndGet();
+			}
+		}
+	};
+
+	public ProcessConsoleTests() {
+		super(ProcessConsoleTests.class.getSimpleName());
+	}
+
+	public ProcessConsoleTests(String name) {
+		super(name);
+	}
+
+	@Override
+	protected void setUp() throws Exception {
+		super.setUp();
+		loggedErrors.set(0);
+		Platform.addLogListener(errorLogListener);
+	}
+
+	@Override
+	protected void tearDown() throws Exception {
+		assertEquals("Test triggered errors.", 0, loggedErrors.get());
+		Platform.removeLogListener(errorLogListener);
+		super.tearDown();
+	}
+
+	/**
+	 * Test if InputReadJob can be canceled.
+	 * <p>
+	 * Actually tests cancellation for all jobs of
+	 * <code>ProcessConsole.class</code> family.
+	 * </p>
+	 */
+	public void testInputReadJobCancel() throws Exception {
+		final MockProcess mockProcess = new MockProcess(MockProcess.RUN_FOREVER);
+		try {
+			final IProcess process = DebugPlugin.newProcess(new Launch(null, ILaunchManager.RUN_MODE, null), mockProcess, "testInputReadJobCancel");
+			@SuppressWarnings("restriction")
+			final org.eclipse.debug.internal.ui.views.console.ProcessConsole console = new org.eclipse.debug.internal.ui.views.console.ProcessConsole(process, new ConsoleColorProvider());
+			try {
+				console.initialize();
+				@SuppressWarnings("restriction")
+				final Class<?> jobFamily = org.eclipse.debug.internal.ui.views.console.ProcessConsole.class;
+				assertTrue("Input read job not started.", Job.getJobManager().find(jobFamily).length > 0);
+				Job.getJobManager().cancel(jobFamily);
+				TestUtil.waitForJobs(getName(), 0, 1000);
+				assertEquals("Input read job not canceled.", 0, Job.getJobManager().find(jobFamily).length);
+			} finally {
+				console.destroy();
+			}
+		} finally {
+			mockProcess.destroy();
+		}
+	}
+}
diff --git a/org.eclipse.debug.ui/ui/org/eclipse/debug/internal/ui/views/console/ProcessConsole.java b/org.eclipse.debug.ui/ui/org/eclipse/debug/internal/ui/views/console/ProcessConsole.java
index 0af92d15a..085cb0f80 100644
--- a/org.eclipse.debug.ui/ui/org/eclipse/debug/internal/ui/views/console/ProcessConsole.java
+++ b/org.eclipse.debug.ui/ui/org/eclipse/debug/internal/ui/views/console/ProcessConsole.java
@@ -707,11 +707,32 @@ public class ProcessConsole extends IOConsole implements IConsole, IDebugEventSe
 
         private IStreamsProxy streamsProxy;
 
+		/**
+		 * The {@link InputStream} this job is currently reading from or maybe blocking
+		 * on. May be <code>null</code>.
+		 */
+		private InputStream readingStream;
+
         InputReadJob(IStreamsProxy streamsProxy) {
             super("Process Console Input Job"); //$NON-NLS-1$
             this.streamsProxy = streamsProxy;
         }
 
+		@Override
+		protected void canceling() {
+			super.canceling();
+			if (readingStream != null) {
+				// Close stream or job may not be able to cancel.
+				// This is primary for IOConsoleInputStream because there is no guarantee an
+				// arbitrary InputStream will release a blocked read() on close.
+				try {
+					readingStream.close();
+				} catch (IOException e) {
+					DebugUIPlugin.log(e);
+				}
+			}
+		}
+
         @Override
 		public boolean belongsTo(Object family) {
 			return ProcessConsole.class == family;
@@ -723,12 +744,12 @@ public class ProcessConsole extends IOConsole implements IConsole, IDebugEventSe
             try {
                 byte[] b = new byte[1024];
                 int read = 0;
-				while (read >= 0) {
-					InputStream input = fInput;
-					if (input == null) {
+				while (read >= 0 && !monitor.isCanceled()) {
+					readingStream = fInput;
+					if (readingStream == null) {
 						break;
 					}
-					read = input.read(b);
+					read = readingStream.read(b);
                     if (read > 0) {
                         String s;
                         if (encoding != null) {
@@ -742,7 +763,8 @@ public class ProcessConsole extends IOConsole implements IConsole, IDebugEventSe
             } catch (IOException e) {
                 DebugUIPlugin.log(e);
             }
-            return Status.OK_STATUS;
+			readingStream = null;
+			return monitor.isCanceled() ? Status.CANCEL_STATUS : Status.OK_STATUS;
         }
     }