From 243b0d64a68990b10a3e6e0d5612cd5bb25626f8 Mon Sep 17 00:00:00 2001
From: "Shawn O. Pearce" <spearce@spearce.org>
Date: Wed, 10 Feb 2010 19:54:07 -0800
Subject: [PATCH] Wait for EOF on stderr before finishing SSH channel

JSch will allow us to close the connection and then just drop
any late messages coming over the stderr stream for the command.
This makes it easy to lose final output on a command, like from
Gerrit Code Review's post receive hook.

Instead spawn a background thread to copy data from JSch's pipe
into our own buffer, and wait for that thread to receive EOF on the
pipe before we declare the connection closed. This way we don't
have a race condition between the stderr data arriving and JSch
just tearing down the channel.

Change-Id: Ica1ba40ed2b4b6efb7d5e4ea240efc0a56fb71f6
Signed-off-by: Shawn O. Pearce <spearce@spearce.org>
---
 .../jgit/transport/TransportGitSsh.java       | 58 ++++++++++++++-----
 1 file changed, 45 insertions(+), 13 deletions(-)

diff --git a/org.eclipse.jgit/src/org/eclipse/jgit/transport/TransportGitSsh.java b/org.eclipse.jgit/src/org/eclipse/jgit/transport/TransportGitSsh.java
index c69304243..d4d4f5412 100644
--- a/org.eclipse.jgit/src/org/eclipse/jgit/transport/TransportGitSsh.java
+++ b/org.eclipse.jgit/src/org/eclipse/jgit/transport/TransportGitSsh.java
@@ -1,4 +1,5 @@
 /*
+ * Copyright (C) 2008-2010, Google Inc.
  * Copyright (C) 2008, Marek Zawirski <marek.zawirski@gmail.com>
  * Copyright (C) 2008, Robin Rosenberg <robin.rosenberg@dewire.com>
  * Copyright (C) 2008, Shawn O. Pearce <spearce@spearce.org>
@@ -46,6 +47,7 @@
 package org.eclipse.jgit.transport;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.io.OutputStream;
 import java.io.PipedInputStream;
 import java.io.PipedOutputStream;
@@ -142,15 +144,13 @@ private String commandFor(final String exe) {
 		return cmd.toString();
 	}
 
-	ChannelExec exec(final String exe, final OutputStream err)
-			throws TransportException {
+	ChannelExec exec(final String exe) throws TransportException {
 		initSession();
 
 		final int tms = getTimeout() > 0 ? getTimeout() * 1000 : 0;
 		try {
 			final ChannelExec channel = (ChannelExec) sock.openChannel("exec");
 			channel.setCommand(commandFor(exe));
-			channel.setErrStream(err);
 			channel.connect(tms);
 			return channel;
 		} catch (JSchException je) {
@@ -224,6 +224,8 @@ public void close() throws IOException {
 	class SshFetchConnection extends BasePackFetchConnection {
 		private ChannelExec channel;
 
+		private Thread errorThread;
+
 		private int exitStatus;
 
 		SshFetchConnection() throws TransportException {
@@ -231,12 +233,16 @@ class SshFetchConnection extends BasePackFetchConnection {
 			try {
 				final MessageWriter msg = new MessageWriter();
 				setMessageWriter(msg);
-				channel = exec(getOptionUploadPack(), msg.getRawStream());
 
-				if (channel.isConnected())
-					init(channel.getInputStream(), outputStream(channel));
-				else
-					throw new TransportException(uri, getMessages());
+				channel = exec(getOptionUploadPack());
+				if (!channel.isConnected())
+					throw new TransportException(uri, "connection failed");
+
+				final InputStream upErr = channel.getErrStream();
+				errorThread = new StreamCopyThread(upErr, msg.getRawStream());
+				errorThread.start();
+
+				init(channel.getInputStream(), outputStream(channel));
 
 			} catch (TransportException err) {
 				close();
@@ -258,6 +264,16 @@ class SshFetchConnection extends BasePackFetchConnection {
 
 		@Override
 		public void close() {
+			if (errorThread != null) {
+				try {
+					errorThread.join();
+				} catch (InterruptedException e) {
+					// Stop waiting and return anyway.
+				} finally {
+					errorThread = null;
+				}
+			}
+
 			super.close();
 
 			if (channel != null) {
@@ -275,6 +291,8 @@ public void close() {
 	class SshPushConnection extends BasePackPushConnection {
 		private ChannelExec channel;
 
+		private Thread errorThread;
+
 		private int exitStatus;
 
 		SshPushConnection() throws TransportException {
@@ -282,12 +300,16 @@ class SshPushConnection extends BasePackPushConnection {
 			try {
 				final MessageWriter msg = new MessageWriter();
 				setMessageWriter(msg);
-				channel = exec(getOptionReceivePack(), msg.getRawStream());
 
-				if (channel.isConnected())
-					init(channel.getInputStream(), outputStream(channel));
-				else
-					throw new TransportException(uri, getMessages());
+				channel = exec(getOptionReceivePack());
+				if (!channel.isConnected())
+					throw new TransportException(uri, "connection failed");
+
+				final InputStream rpErr = channel.getErrStream();
+				errorThread = new StreamCopyThread(rpErr, msg.getRawStream());
+				errorThread.start();
+
+				init(channel.getInputStream(), outputStream(channel));
 
 			} catch (TransportException err) {
 				close();
@@ -309,6 +331,16 @@ class SshPushConnection extends BasePackPushConnection {
 
 		@Override
 		public void close() {
+			if (errorThread != null) {
+				try {
+					errorThread.join();
+				} catch (InterruptedException e) {
+					// Stop waiting and return anyway.
+				} finally {
+					errorThread = null;
+				}
+			}
+
 			super.close();
 
 			if (channel != null) {