From 2adc572628f9382ace5fbd791325dc64f7c968d3 Mon Sep 17 00:00:00 2001 From: Ian Wetherbee Date: Thu, 7 Jun 2012 16:10:09 -0700 Subject: [PATCH] Refactor TransportHttp for long-polling Split Service into MultiRequestService (fetch, push) and LongPollService (upcoming publish-subscribe). Change-Id: Ice373d3dee63c395490d2707473ccf20a022e5cf --- .../eclipse/jgit/transport/TransportHttp.java | 241 +++++++++++------- 1 file changed, 144 insertions(+), 97 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 7adeeca50..2f68eb9d8 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 @@ void close() { } class SmartHttpFetchConnection extends BasePackFetchConnection { - private Service svc; + private MultiRequestService svc; SmartHttpFetchConnection(final InputStream advertisement) throws TransportException { @@ -734,8 +734,8 @@ protected void doFetch(final ProgressMonitor monitor, final Collection want, final Set 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,12 +762,128 @@ class SmartHttpPushConnection extends BasePackPushConnection { protected void doPush(final ProgressMonitor monitor, final Map 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); } } + /** Basic service for sending and receiving HTTP requests. */ + abstract class Service { + protected final String serviceName; + + protected final String requestType; + + protected final String responseType; + + protected HttpURLConnection conn; + + protected HttpOutputStream out; + + protected final HttpExecuteStream execute; + + final UnionInputStream in; + + 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); + } + + void openStream() throws IOException { + conn = httpOpen(METHOD_POST, new URL(baseUrl, serviceName)); + conn.setInstanceFollowRedirects(false); + conn.setDoOutput(true); + conn.setRequestProperty(HDR_CONTENT_TYPE, requestType); + conn.setRequestProperty(HDR_ACCEPT, responseType); + } + + 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; + } catch (IOException err) { + // Most likely caused by overflowing the buffer, meaning + // its larger if it were compressed. Don't compress. + 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(); + } + } + + void openResponse() throws IOException { + final int status = HttpSupport.response(conn); + if (status != HttpURLConnection.HTTP_OK) { + throw new TransportException(uri, status + " " //$NON-NLS-1$ + + conn.getResponseMessage()); + } + + final String contentType = conn.getContentType(); + if (!responseType.equals(contentType)) { + conn.getInputStream().close(); + throw wrongContentType(responseType, contentType); + } + } + + HttpOutputStream getOutputStream() { + return out; + } + + InputStream getInputStream() { + return in; + } + + abstract void execute() throws IOException; + + class HttpExecuteStream extends InputStream { + public int read() throws IOException { + execute(); + return -1; + } + + public int read(byte[] b, int off, int len) throws IOException { + execute(); + return -1; + } + + public long skip(long n) throws IOException { + execute(); + 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. *

@@ -788,41 +904,15 @@ protected void doPush(final ProgressMonitor monitor, * be preserved between requests, it is left up to the JVM's implementation * of the HTTP client. */ - class Service { - private final String serviceName; - - private final String requestType; - - private final String responseType; - - private final HttpExecuteStream execute; - + class MultiRequestService extends Service { boolean finalRequest; - final UnionInputStream in; - - final HttpOutputStream out; - - HttpURLConnection conn; - - Service(final 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.execute = new HttpExecuteStream(); - this.in = new UnionInputStream(execute); - this.out = new HttpOutputStream(); - } - - void openStream() throws IOException { - conn = httpOpen(METHOD_POST, new URL(baseUrl, serviceName)); - conn.setInstanceFollowRedirects(false); - conn.setDoOutput(true); - conn.setRequestProperty(HDR_CONTENT_TYPE, requestType); - conn.setRequestProperty(HDR_ACCEPT, responseType); + MultiRequestService(final String serviceName) { + super(serviceName); } + /** Keep opening send-receive pairs to the given URI. */ + @Override void execute() throws IOException { out.close(); @@ -839,80 +929,37 @@ void execute() throws IOException { 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. - 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(); - } + sendRequest(); } out.reset(); - final int status = HttpSupport.response(conn); - if (status != HttpURLConnection.HTTP_OK) { - throw new TransportException(uri, status + " " //$NON-NLS-1$ - + conn.getResponseMessage()); - } - - final String contentType = conn.getContentType(); - if (!responseType.equals(contentType)) { - conn.getInputStream().close(); - throw wrongContentType(responseType, contentType); - } + openResponse(); in.add(openInputStream(conn)); if (!finalRequest) in.add(execute); conn = null; } + } - class HttpOutputStream extends TemporaryBuffer { - HttpOutputStream() { - super(http.postBuffer); - } - - @Override - protected OutputStream overflow() throws IOException { - openStream(); - conn.setChunkedStreamingMode(0); - return conn.getOutputStream(); - } + /** Service for maintaining a single long-poll connection. */ + class LongPollService extends Service { + /** + * @param serviceName + */ + LongPollService(String serviceName) { + super(serviceName); } - class HttpExecuteStream extends InputStream { - public int read() throws IOException { - execute(); - return -1; - } - - public int read(byte[] b, int off, int len) throws IOException { - execute(); - return -1; - } - - public long skip(long n) throws IOException { - execute(); - return 0; - } + /** Only open one send-receive request. */ + @Override + void execute() throws IOException { + out.close(); + if (conn == null) + sendRequest(); + openResponse(); + in.add(openInputStream(conn)); } }