Target Explorer: Fix reading from process streams until EOF

3 files changed, 89 insertions, 25 deletions

diff --git a/target_explorer/plugins/org.eclipse.tcf.te.tcf.processes.core/src/org/eclipse/tcf/te/tcf/processes/core/launcher/ProcessLauncher.java b/target_explorer/plugins/org.eclipse.tcf.te.tcf.processes.core/src/org/eclipse/tcf/te/tcf/processes/core/launcher/ProcessLauncher.java
index 6c9b6f17a..3079b0132 100644
--- a/target_explorer/plugins/org.eclipse.tcf.te.tcf.processes.core/src/org/eclipse/tcf/te/tcf/processes/core/launcher/ProcessLauncher.java
+++ b/target_explorer/plugins/org.eclipse.tcf.te.tcf.processes.core/src/org/eclipse/tcf/te/tcf/processes/core/launcher/ProcessLauncher.java
@@ -107,6 +107,9 @@ public class ProcessLauncher extends PlatformObject implements IProcessLauncher
 	// The streams proxy instance
 	private IProcessStreamsProxy streamsProxy = null;
 
+	// The output stream connected to process stdin
+	private OutputStream stdinStream;
+
 	// The active token.
 	IToken activeToken = null;
 
@@ -162,7 +165,12 @@ public class ProcessLauncher extends PlatformObject implements IProcessLauncher
 		if (streamsListener != null) {
 			// Dispose the streams listener
 			if (streamsListener instanceof ProcessStreamsListener) {
-				((ProcessStreamsListener)streamsListener).dispose(new AsyncCallbackCollector.SimpleCollectorCallback(collector));
+				AsyncCallbackCollector.SimpleCollectorCallback cb = new AsyncCallbackCollector.SimpleCollectorCallback(collector);
+				// if process was terminated explicitly dispose immediately, else wait for EOF
+				if (sigTermSent)
+					((ProcessStreamsListener)streamsListener).dispose(cb);
+				else
+					((ProcessStreamsListener)streamsListener).disposeOnEOF(cb);
 			}
 			streamsListener = null;
 		}
@@ -185,6 +193,18 @@ public class ProcessLauncher extends PlatformObject implements IProcessLauncher
 			streamsProxy.dispose(new AsyncCallbackCollector.SimpleCollectorCallback(collector));
 			streamsProxy = null;
 		}
+
+		// Close stdin
+		if (stdinStream != null) {
+			try {
+				stdinStream.close();
+			}
+			catch (IOException e) {
+				// unlikely: ignore exception on close
+			}
+			stdinStream = null;
+		}
+
 		// Mark the collector initialization as done
 		collector.initDone();
 
@@ -512,7 +532,7 @@ public class ProcessLauncher extends PlatformObject implements IProcessLauncher
 		// The streams got subscribed, check what we need to do with them
 		if (streamsProxy != null) {
 			// Publish the streams to the supplied proxy
-			streamsProxy.connectInputStreamMonitor(connectRemoteOutputStream(getStreamsListener(), new String[] { IProcesses.PROP_STDIN_ID }));
+			streamsProxy.connectInputStreamMonitor(stdinStream = connectRemoteOutputStream(getStreamsListener(), new String[] { IProcesses.PROP_STDIN_ID }));
 			// Create and store the streams the terminal will see as stdout
 			streamsProxy.connectOutputStreamMonitor(connectRemoteInputStream(getStreamsListener(), new String[] { IProcesses.PROP_STDOUT_ID }, null));
 			// Create and store the streams the terminal will see as stderr
@@ -1100,34 +1120,32 @@ public class ProcessLauncher extends PlatformObject implements IProcessLauncher
 		return processesListener;
 	}
 
-	/* (non-Javadoc)
-	 * @see org.eclipse.core.runtime.PlatformObject#getAdapter(java.lang.Class)
-	 */
+	@SuppressWarnings("unchecked")
 	@Override
-	public Object getAdapter(Class adapter) {
+	public <T> T getAdapter(Class<T> adapter) {
 		if (adapter.isAssignableFrom(IProcesses.ProcessesListener.class)) {
-			return processesListener;
+			return (T) processesListener;
 		}
 		else if (adapter.isAssignableFrom(IStreams.StreamsListener.class)) {
-			return streamsListener;
+			return (T) streamsListener;
 		}
 		else if (adapter.isAssignableFrom(IStreams.class)) {
-			return svcStreams;
+			return (T) svcStreams;
 		}
 		else if (adapter.isAssignableFrom(IProcesses.class)) {
-			return svcProcesses;
+			return (T) svcProcesses;
 		}
 		else if (adapter.isAssignableFrom(IChannel.class)) {
-			return channel;
+			return (T) channel;
 		}
 		else if (adapter.isAssignableFrom(IPropertiesContainer.class)) {
-			return properties;
+			return (T) properties;
 		}
 		else if (adapter.isAssignableFrom(IProcesses.ProcessContext.class)) {
-			return processContext;
+			return (T) processContext;
 		}
 		else if (adapter.isAssignableFrom(this.getClass())) {
-			return this;
+			return (T) this;
 		}
 
 
diff --git a/target_explorer/plugins/org.eclipse.tcf.te.tcf.processes.core/src/org/eclipse/tcf/te/tcf/processes/core/launcher/ProcessStreamsListener.java b/target_explorer/plugins/org.eclipse.tcf.te.tcf.processes.core/src/org/eclipse/tcf/te/tcf/processes/core/launcher/ProcessStreamsListener.java
index 1c8a78ec7..bf2d8631e 100644
--- a/target_explorer/plugins/org.eclipse.tcf.te.tcf.processes.core/src/org/eclipse/tcf/te/tcf/processes/core/launcher/ProcessStreamsListener.java
+++ b/target_explorer/plugins/org.eclipse.tcf.te.tcf.processes.core/src/org/eclipse/tcf/te/tcf/processes/core/launcher/ProcessStreamsListener.java
@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (c) 2011, 2014 Wind River Systems, Inc. and others. All rights reserved.
+ * Copyright (c) 2011, 2015 Wind River Systems, Inc. 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 http://www.eclipse.org/legal/epl-v10.html
@@ -162,6 +162,17 @@ public class ProcessStreamsListener implements IChannelManager.IStreamsListener,
 		 * @param callback The callback to invoke if the runnable stopped.
 		 */
 		public final synchronized void stop(ICallback callback) {
+			onEOF(callback);
+			// Mark the runnable as stopped
+			stopped = true;
+		}
+
+		/**
+		 * Notify callback on EOF.
+		 *
+		 * @param callback The callback to invoke on EOF
+		 */
+		public final synchronized void onEOF(ICallback callback) {
 			// If the runnable is stopped already, invoke the callback directly
 			if (stopped) {
 				if (callback != null) callback.done(this, Status.OK_STATUS);
@@ -170,8 +181,6 @@ public class ProcessStreamsListener implements IChannelManager.IStreamsListener,
 
 			// Store the callback instance
 			this.callback = callback;
-			// Mark the runnable as stopped
-			stopped = true;
 		}
 
 		/**
@@ -408,6 +417,17 @@ public class ProcessStreamsListener implements IChannelManager.IStreamsListener,
 		 * @param callback The callback to invoke if the runnable stopped.
 		 */
 		public final synchronized void stop(ICallback callback) {
+			onEOF(callback);
+			// Mark the runnable as stopped
+			stopped = true;
+		}
+
+		/**
+		 * Notify callback on EOF.
+		 *
+		 * @param callback The callback to invoke on EOF
+		 */
+		public final synchronized void onEOF(ICallback callback) {
 			// If the runnable is stopped already, invoke the callback directly
 			if (stopped) {
 				if (callback != null) callback.done(this, Status.OK_STATUS);
@@ -416,8 +436,6 @@ public class ProcessStreamsListener implements IChannelManager.IStreamsListener,
 
 			// Store the callback instance
 			this.callback = callback;
-			// Mark the runnable as stopped
-			stopped = true;
 		}
 
 		/**
@@ -593,7 +611,7 @@ public class ProcessStreamsListener implements IChannelManager.IStreamsListener,
 	 */
 	@Override
 	public void dispose() {
-		dispose(null);
+		dispose(false, null);
 	}
 
 	/**
@@ -602,6 +620,25 @@ public class ProcessStreamsListener implements IChannelManager.IStreamsListener,
 	 * @param callback The callback to invoke if the dispose finished or <code>null</code>.
 	 */
 	public void dispose(final ICallback callback) {
+		dispose(false, callback);
+	}
+
+	/**
+	 * Dispose the streams listener instance only on EOF.
+	 * This allows to keep reading streams after the process has exited.
+	 *
+	 * @param callback The callback to invoke if the dispose finished or <code>null</code>.
+	 */
+	void disposeOnEOF(final ICallback callback) {
+		dispose(true, callback);
+	}
+
+	/**
+	 * Dispose the streams listener instance.
+	 * @param onEof dispose as soon as all streams have seen EOF
+	 * @param callback The callback to invoke if the dispose finished or <code>null</code>.
+	 */
+	private void dispose(boolean onEof, final ICallback callback) {
 		// Store a final reference to the streams listener instance
 		final IChannelManager.IStreamsListener finStreamsListener = this;
 
@@ -636,14 +673,21 @@ public class ProcessStreamsListener implements IChannelManager.IStreamsListener,
 			}
 		}, new CallbackInvocationDelegate());
 
-		// Loop all runnable's and force them to stop
+		// Loop all runnable's and attach our callback
 		synchronized (runnables) {
 			for (Runnable runnable : runnables) {
+				AsyncCallbackCollector.SimpleCollectorCallback cb = new AsyncCallbackCollector.SimpleCollectorCallback(collector);
 				if (runnable instanceof StreamReaderRunnable) {
-					((StreamReaderRunnable)runnable).stop(new AsyncCallbackCollector.SimpleCollectorCallback(collector));
+					if (onEof)
+						((StreamReaderRunnable)runnable).onEOF(cb);
+					else
+						((StreamReaderRunnable)runnable).stop(cb);
 				}
-				if (runnable instanceof StreamWriterRunnable) {
-					((StreamWriterRunnable)runnable).stop(new AsyncCallbackCollector.SimpleCollectorCallback(collector));
+				else if (runnable instanceof StreamWriterRunnable) {
+					if (onEof)
+						((StreamWriterRunnable)runnable).onEOF(cb);
+					else
+						((StreamWriterRunnable)runnable).stop(cb);
 				}
 			}
 			runnables.clear();
diff --git a/target_explorer/plugins/org.eclipse.tcf.te.tcf.remote.core/src/org/eclipse/tcf/te/tcf/remote/core/TCFProcessStreams.java b/target_explorer/plugins/org.eclipse.tcf.te.tcf.remote.core/src/org/eclipse/tcf/te/tcf/remote/core/TCFProcessStreams.java
index 48d8cd419..0ff3f2771 100644
--- a/target_explorer/plugins/org.eclipse.tcf.te.tcf.remote.core/src/org/eclipse/tcf/te/tcf/remote/core/TCFProcessStreams.java
+++ b/target_explorer/plugins/org.eclipse.tcf.te.tcf.remote.core/src/org/eclipse/tcf/te/tcf/remote/core/TCFProcessStreams.java
@@ -14,6 +14,7 @@ package org.eclipse.tcf.te.tcf.remote.core;
 import java.io.InputStream;
 import java.io.OutputStream;
 
+import org.eclipse.core.runtime.Status;
 import org.eclipse.tcf.te.runtime.interfaces.callback.ICallback;
 import org.eclipse.tcf.te.tcf.processes.core.interfaces.launcher.IProcessStreamsProxy;
 
@@ -26,7 +27,6 @@ public class TCFProcessStreams implements IProcessStreamsProxy {
 	@Override
 	public void connectOutputStreamMonitor(InputStream stream) {
 		fRemoteStdout = stream;
-
 	}
 
 	@Override
@@ -41,6 +41,8 @@ public class TCFProcessStreams implements IProcessStreamsProxy {
 
 	@Override
 	public void dispose(ICallback callback) {
+		if (callback != null)
+			callback.done(this, Status.OK_STATUS);
 	}
 
 	public InputStream getStdout() {