Refactor TransportHttp for long-polling

Split Service into MultiRequestService (fetch, push) and
LongPollService (upcoming publish-subscribe).

Change-Id: Ice373d3dee63c395490d2707473ccf20a022e5cf

1 files changed, 139 insertions, 92 deletions

diff --git a/org.eclipse.jgit/src/org/eclipse/jgit/transport/TransportHttp.java b/org.eclipse.jgit/src/org/eclipse/jgit/transport/TransportHttp.java
index 7adeeca50c..2f68eb9d8d 100644
--- a/org.eclipse.jgit/src/org/eclipse/jgit/transport/TransportHttp.java
+++ b/org.eclipse.jgit/src/org/eclipse/jgit/transport/TransportHttp.java
@@ -717,7 +717,7 @@ public class TransportHttp extends HttpTransport implements WalkTransport,
 	}
 
 	class SmartHttpFetchConnection extends BasePackFetchConnection {
-		private Service svc;
+		private MultiRequestService svc;
 
 		SmartHttpFetchConnection(final InputStream advertisement)
 				throws TransportException {
@@ -734,8 +734,8 @@ public class TransportHttp extends HttpTransport implements WalkTransport,
 				final Collection<Ref> want, final Set<ObjectId> have)
 				throws TransportException {
 			try {
-				svc = new Service(SVC_UPLOAD_PACK);
-				init(svc.in, svc.out);
+				svc = new MultiRequestService(SVC_UPLOAD_PACK);
+				init(svc.getInputStream(), svc.getOutputStream());
 				super.doFetch(monitor, want, have);
 			} finally {
 				svc = null;
@@ -762,57 +762,36 @@ public class TransportHttp extends HttpTransport implements WalkTransport,
 		protected void doPush(final ProgressMonitor monitor,
 				final Map<String, RemoteRefUpdate> refUpdates)
 				throws TransportException {
-			final Service svc = new Service(SVC_RECEIVE_PACK);
-			init(svc.in, svc.out);
+			final Service svc = new MultiRequestService(SVC_RECEIVE_PACK);
+			init(svc.getInputStream(), svc.getOutputStream());
 			super.doPush(monitor, refUpdates);
 		}
 	}
 
-	/**
-	 * State required to speak multiple HTTP requests with the remote.
-	 * <p>
-	 * A service wrapper provides a normal looking InputStream and OutputStream
-	 * pair which are connected via HTTP to the named remote service. Writing to
-	 * the OutputStream is buffered until either the buffer overflows, or
-	 * reading from the InputStream occurs. If overflow occurs HTTP/1.1 and its
-	 * chunked transfer encoding is used to stream the request data to the
-	 * remote service. If the entire request fits in the memory buffer, the
-	 * older HTTP/1.0 standard and a fixed content length is used instead.
-	 * <p>
-	 * It is an error to attempt to read without there being outstanding data
-	 * ready for transmission on the OutputStream.
-	 * <p>
-	 * No state is preserved between write-read request pairs. The caller is
-	 * responsible for replaying state vector information as part of the request
-	 * data written to the OutputStream. Any session HTTP cookies may or may not
-	 * be preserved between requests, it is left up to the JVM's implementation
-	 * of the HTTP client.
-	 */
-	class Service {
-		private final String serviceName;
+	/** Basic service for sending and receiving HTTP requests. */
+	abstract class Service {
+		protected final String serviceName;
 
-		private final String requestType;
+		protected final String requestType;
 
-		private final String responseType;
+		protected final String responseType;
 
-		private final HttpExecuteStream execute;
+		protected HttpURLConnection conn;
 
-		boolean finalRequest;
+		protected HttpOutputStream out;
 
-		final UnionInputStream in;
+		protected final HttpExecuteStream execute;
 
-		final HttpOutputStream out;
-
-		HttpURLConnection conn;
+		final UnionInputStream in;
 
-		Service(final String serviceName) {
+		Service(String serviceName) {
 			this.serviceName = serviceName;
 			this.requestType = "application/x-" + serviceName + "-request"; //$NON-NLS-1$ //$NON-NLS-2$
 			this.responseType = "application/x-" + serviceName + "-result"; //$NON-NLS-1$ //$NON-NLS-2$
 
+			this.out = new HttpOutputStream();
 			this.execute = new HttpExecuteStream();
 			this.in = new UnionInputStream(execute);
-			this.out = new HttpOutputStream();
 		}
 
 		void openStream() throws IOException {
@@ -823,50 +802,34 @@ public class TransportHttp extends HttpTransport implements WalkTransport,
 			conn.setRequestProperty(HDR_ACCEPT, responseType);
 		}
 
-		void execute() throws IOException {
-			out.close();
-
-			if (conn == null) {
-				if (out.length() == 0) {
-					// Request output hasn't started yet, but more data is being
-					// requested. If there is no request data buffered and the
-					// final request was already sent, do nothing to ensure the
-					// caller is shown EOF on the InputStream; otherwise an
-					// programming error has occurred within this module.
-					if (finalRequest)
-						return;
-					throw new TransportException(uri,
-							JGitText.get().startingReadStageWithoutWrittenRequestDataPendingIsNotSupported);
-				}
-
-				// Try to compress the content, but only if that is smaller.
-				TemporaryBuffer buf = new TemporaryBuffer.Heap(http.postBuffer);
-				try {
-					GZIPOutputStream gzip = new GZIPOutputStream(buf);
-					out.writeTo(gzip, null);
-					gzip.close();
-					if (out.length() < buf.length())
-						buf = out;
-				} catch (IOException err) {
-					// Most likely caused by overflowing the buffer, meaning
-					// its larger if it were compressed. Don't compress.
+		void sendRequest() throws IOException {
+			// Try to compress the content, but only if that is smaller.
+			TemporaryBuffer buf = new TemporaryBuffer.Heap(http.postBuffer);
+			try {
+				GZIPOutputStream gzip = new GZIPOutputStream(buf);
+				out.writeTo(gzip, null);
+				gzip.close();
+				if (out.length() < buf.length())
 					buf = out;
-				}
-
-				openStream();
-				if (buf != out)
-					conn.setRequestProperty(HDR_CONTENT_ENCODING, ENCODING_GZIP);
-				conn.setFixedLengthStreamingMode((int) buf.length());
-				final OutputStream httpOut = conn.getOutputStream();
-				try {
-					buf.writeTo(httpOut, null);
-				} finally {
-					httpOut.close();
-				}
+			} catch (IOException err) {
+				// Most likely caused by overflowing the buffer, meaning
+				// its larger if it were compressed. Don't compress.
+				buf = out;
 			}
 
-			out.reset();
+			openStream();
+			if (buf != out)
+				conn.setRequestProperty(HDR_CONTENT_ENCODING, ENCODING_GZIP);
+			conn.setFixedLengthStreamingMode((int) buf.length());
+			final OutputStream httpOut = conn.getOutputStream();
+			try {
+				buf.writeTo(httpOut, null);
+			} finally {
+				httpOut.close();
+			}
+		}
 
+		void openResponse() throws IOException {
 			final int status = HttpSupport.response(conn);
 			if (status != HttpURLConnection.HTTP_OK) {
 				throw new TransportException(uri, status + " " //$NON-NLS-1$
@@ -878,26 +841,18 @@ public class TransportHttp extends HttpTransport implements WalkTransport,
 				conn.getInputStream().close();
 				throw wrongContentType(responseType, contentType);
 			}
-
-			in.add(openInputStream(conn));
-			if (!finalRequest)
-				in.add(execute);
-			conn = null;
 		}
 
-		class HttpOutputStream extends TemporaryBuffer {
-			HttpOutputStream() {
-				super(http.postBuffer);
-			}
+		HttpOutputStream getOutputStream() {
+			return out;
+		}
 
-			@Override
-			protected OutputStream overflow() throws IOException {
-				openStream();
-				conn.setChunkedStreamingMode(0);
-				return conn.getOutputStream();
-			}
+		InputStream getInputStream() {
+			return in;
 		}
 
+		abstract void execute() throws IOException;
+
 		class HttpExecuteStream extends InputStream {
 			public int read() throws IOException {
 				execute();
@@ -914,6 +869,98 @@ public class TransportHttp extends HttpTransport implements WalkTransport,
 				return 0;
 			}
 		}
+
+		class HttpOutputStream extends TemporaryBuffer {
+			HttpOutputStream() {
+				super(http.postBuffer);
+			}
+
+			@Override
+			protected OutputStream overflow() throws IOException {
+				openStream();
+				conn.setChunkedStreamingMode(0);
+				return conn.getOutputStream();
+			}
+		}
+	}
+
+	/**
+	 * State required to speak multiple HTTP requests with the remote.
+	 * <p>
+	 * A service wrapper provides a normal looking InputStream and OutputStream
+	 * pair which are connected via HTTP to the named remote service. Writing to
+	 * the OutputStream is buffered until either the buffer overflows, or
+	 * reading from the InputStream occurs. If overflow occurs HTTP/1.1 and its
+	 * chunked transfer encoding is used to stream the request data to the
+	 * remote service. If the entire request fits in the memory buffer, the
+	 * older HTTP/1.0 standard and a fixed content length is used instead.
+	 * <p>
+	 * It is an error to attempt to read without there being outstanding data
+	 * ready for transmission on the OutputStream.
+	 * <p>
+	 * No state is preserved between write-read request pairs. The caller is
+	 * responsible for replaying state vector information as part of the request
+	 * data written to the OutputStream. Any session HTTP cookies may or may not
+	 * be preserved between requests, it is left up to the JVM's implementation
+	 * of the HTTP client.
+	 */
+	class MultiRequestService extends Service {
+		boolean finalRequest;
+
+		MultiRequestService(final String serviceName) {
+			super(serviceName);
+		}
+
+		/** Keep opening send-receive pairs to the given URI. */
+		@Override
+		void execute() throws IOException {
+			out.close();
+
+			if (conn == null) {
+				if (out.length() == 0) {
+					// Request output hasn't started yet, but more data is being
+					// requested. If there is no request data buffered and the
+					// final request was already sent, do nothing to ensure the
+					// caller is shown EOF on the InputStream; otherwise an
+					// programming error has occurred within this module.
+					if (finalRequest)
+						return;
+					throw new TransportException(uri,
+							JGitText.get().startingReadStageWithoutWrittenRequestDataPendingIsNotSupported);
+				}
+
+				sendRequest();
+			}
+
+			out.reset();
+
+			openResponse();
+
+			in.add(openInputStream(conn));
+			if (!finalRequest)
+				in.add(execute);
+			conn = null;
+		}
+	}
+
+	/** Service for maintaining a single long-poll connection. */
+	class LongPollService extends Service {
+		/**
+		 * @param serviceName
+		 */
+		LongPollService(String serviceName) {
+			super(serviceName);
+		}
+
+		/** Only open one send-receive request. */
+		@Override
+		void execute() throws IOException {
+			out.close();
+			if (conn == null)
+				sendRequest();
+			openResponse();
+			in.add(openInputStream(conn));
+		}
 	}
 
 	private static class DummyX509TrustManager implements X509TrustManager {