From 7f0efbdac8884584fc1ee662abe3702fef725602 Mon Sep 17 00:00:00 2001 From: Noah Levitt Date: Mon, 18 Jun 2012 15:42:50 -0700 Subject: [PATCH] do recording through SessionOutputBuffer implementation like we have for SessionInputBuffer; refactor helper classes into their own file RecordingHttpClient.java --- .../archive/modules/fetcher/FetchHTTP2.java | 94 +-------- .../modules/fetcher/RecordingHttpClient.java | 103 ++++++++++ .../fetcher/RecordingSocketInputBuffer.java | 23 ++- .../fetcher/RecordingSocketOutputBuffer.java | 184 ++++++++++++++++++ 4 files changed, 313 insertions(+), 91 deletions(-) create mode 100644 modules/src/main/java/org/archive/modules/fetcher/RecordingHttpClient.java create mode 100644 modules/src/main/java/org/archive/modules/fetcher/RecordingSocketOutputBuffer.java diff --git a/modules/src/main/java/org/archive/modules/fetcher/FetchHTTP2.java b/modules/src/main/java/org/archive/modules/fetcher/FetchHTTP2.java index b258fc5b..8a86539e 100644 --- a/modules/src/main/java/org/archive/modules/fetcher/FetchHTTP2.java +++ b/modules/src/main/java/org/archive/modules/fetcher/FetchHTTP2.java @@ -27,38 +27,18 @@ import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_FETCH_HIST import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_REFERENCE_LENGTH; import java.io.IOException; -import java.io.InputStream; -import java.io.OutputStream; -import java.net.Socket; import java.security.MessageDigest; import java.util.Map; import org.apache.commons.httpclient.HttpStatus; import org.apache.http.Header; -import org.apache.http.HttpException; import org.apache.http.HttpResponse; import org.apache.http.client.ClientProtocolException; import org.apache.http.client.HttpClient; -import org.apache.http.client.HttpRequestRetryHandler; import org.apache.http.client.methods.HttpGet; import org.apache.http.client.methods.HttpPost; import org.apache.http.client.methods.HttpRequestBase; -import org.apache.http.conn.ClientConnectionManager; -import org.apache.http.conn.ClientConnectionOperator; -import org.apache.http.conn.OperatedClientConnection; -import org.apache.http.conn.scheme.SchemeRegistry; import org.apache.http.impl.client.DefaultHttpClient; -import org.apache.http.impl.conn.BasicClientConnectionManager; -import org.apache.http.impl.conn.DefaultClientConnection; -import org.apache.http.impl.conn.DefaultClientConnectionOperator; -import org.apache.http.impl.conn.SchemeRegistryFactory; -import org.apache.http.impl.io.AbstractSessionInputBuffer; -import org.apache.http.impl.io.AbstractSessionOutputBuffer; -import org.apache.http.impl.io.IdentityOutputStream; -import org.apache.http.io.SessionInputBuffer; -import org.apache.http.io.SessionOutputBuffer; -import org.apache.http.params.HttpParams; -import org.apache.http.protocol.HttpContext; import org.archive.io.RecorderLengthExceededException; import org.archive.io.RecorderTimeoutException; import org.archive.modules.CrawlURI; @@ -72,6 +52,8 @@ import org.springframework.context.Lifecycle; public class FetchHTTP2 extends Processor implements Lifecycle { + protected DefaultHttpClient httpClient; + protected ServerCache serverCache; public ServerCache getServerCache() { return this.serverCache; @@ -98,7 +80,7 @@ public class FetchHTTP2 extends Processor implements Lifecycle { kp.put("digestContent",digest); } - protected String digestAlgorithm = "sha1"; + protected String digestAlgorithm = "sha1"; public String getDigestAlgorithm() { return digestAlgorithm; } @@ -164,10 +146,9 @@ public class FetchHTTP2 extends Processor implements Lifecycle { request = new HttpGet(curiString); curi.setFetchType(FetchType.HTTP_GET); } - HttpClient httpClient = obtainHttpClient(rec); HttpResponse response = null; try { - response = httpClient.execute(request); + response = getHttpClient().execute(request); addResponseContent(response, curi); } catch (ClientProtocolException e) { failedExecuteCleanup(request, curi, e); @@ -225,29 +206,10 @@ public class FetchHTTP2 extends Processor implements Lifecycle { } } - protected HttpClient obtainHttpClient(final Recorder rec) { - DefaultHttpClient httpClient = new DefaultHttpClient() { - @Override - protected ClientConnectionManager createClientConnectionManager() { - return new BasicClientConnectionManager( - SchemeRegistryFactory.createDefault()) { - @Override - protected ClientConnectionOperator createConnectionOperator( - SchemeRegistry schreg) { - return new RecordingClientConnectionOperator(schreg, rec); - } - }; - } - }; - - // never retry (heritrix handles this elsewhere) - httpClient.setHttpRequestRetryHandler(new HttpRequestRetryHandler() { - @Override - public boolean retryRequest(IOException exception, int executionCount, - HttpContext context) { - return false; - } - }); + protected HttpClient getHttpClient() { + if (httpClient == null) { + httpClient = new RecordingHttpClient(); + } return httpClient; } @@ -342,44 +304,4 @@ public class FetchHTTP2 extends Processor implements Lifecycle { curi.setFetchStatus(status); curi.getRecorder().close(); } - - protected static class RecordingClientConnectionOperator extends DefaultClientConnectionOperator { - - protected final Recorder rec; - - protected RecordingClientConnectionOperator(SchemeRegistry schemes, Recorder rec) { - super(schemes); - this.rec = rec; - } - - @Override - public OperatedClientConnection createConnection() { - return new DefaultClientConnection() { - @Override - protected SessionInputBuffer createSessionInputBuffer(Socket socket, int buffersize, HttpParams params) throws IOException { - return new RecordingSocketInputBuffer(rec, socket, params); - } - - @Override - protected SessionOutputBuffer createSessionOutputBuffer(Socket socket, int buffersize, HttpParams params) throws IOException { - SessionOutputBuffer sob = super.createSessionOutputBuffer(socket, buffersize, params); - OutputStream ros = rec.outputWrap(new IdentityOutputStream(sob)); - return new HcOutputWrapper(ros, buffersize, params); - } - - @Override - public void receiveResponseEntity(HttpResponse response) - throws HttpException, IOException { - rec.markContentBegin(); - super.receiveResponseEntity(response); - } - }; - } - } - - protected static class HcOutputWrapper extends AbstractSessionOutputBuffer { - public HcOutputWrapper(OutputStream out, int buffersize, HttpParams params) { - this.init(out, buffersize, params); - } - } } diff --git a/modules/src/main/java/org/archive/modules/fetcher/RecordingHttpClient.java b/modules/src/main/java/org/archive/modules/fetcher/RecordingHttpClient.java new file mode 100644 index 00000000..30cfbdd0 --- /dev/null +++ b/modules/src/main/java/org/archive/modules/fetcher/RecordingHttpClient.java @@ -0,0 +1,103 @@ +/* + * This file is part of the Heritrix web crawler (crawler.archive.org). + * + * Licensed to the Internet Archive (IA) by one or more individual + * contributors. + * + * The IA licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.archive.modules.fetcher; + +import java.io.IOException; +import java.net.Socket; + +import org.apache.http.HttpException; +import org.apache.http.HttpResponse; +import org.apache.http.client.HttpRequestRetryHandler; +import org.apache.http.conn.ClientConnectionManager; +import org.apache.http.conn.ClientConnectionOperator; +import org.apache.http.conn.OperatedClientConnection; +import org.apache.http.conn.scheme.SchemeRegistry; +import org.apache.http.impl.client.DefaultHttpClient; +import org.apache.http.impl.conn.BasicClientConnectionManager; +import org.apache.http.impl.conn.DefaultClientConnection; +import org.apache.http.impl.conn.DefaultClientConnectionOperator; +import org.apache.http.impl.conn.SchemeRegistryFactory; +import org.apache.http.io.SessionInputBuffer; +import org.apache.http.io.SessionOutputBuffer; +import org.apache.http.params.HttpParams; +import org.apache.http.protocol.HttpContext; +import org.archive.util.Recorder; + +public class RecordingHttpClient extends DefaultHttpClient { + public RecordingHttpClient() { + super(); + + // never retry (heritrix handles this elsewhere) + setHttpRequestRetryHandler(new HttpRequestRetryHandler() { + @Override + public boolean retryRequest(IOException exception, int executionCount, + HttpContext context) { + return false; + } + }); + } + + @Override + protected ClientConnectionManager createClientConnectionManager() { + return new BasicClientConnectionManager( + SchemeRegistryFactory.createDefault()) { + @Override + protected ClientConnectionOperator createConnectionOperator( + SchemeRegistry schreg) { + return new RecordingHttpClient.RecordingClientConnectionOperator(schreg); + } + }; + } + + protected static class RecordingClientConnectionOperator extends DefaultClientConnectionOperator { + protected RecordingClientConnectionOperator(SchemeRegistry schemes) { + super(schemes); + } + + @Override + public OperatedClientConnection createConnection() { + return new DefaultClientConnection() { + @Override + protected SessionInputBuffer createSessionInputBuffer(Socket socket, int buffersize, HttpParams params) throws IOException { + return new RecordingSocketInputBuffer(socket, buffersize, params); + } + + @Override + protected SessionOutputBuffer createSessionOutputBuffer(Socket socket, int buffersize, HttpParams params) throws IOException { + return new RecordingSocketOutputBuffer(socket, buffersize, params); + } + + @Override + public void receiveResponseEntity(HttpResponse response) + throws HttpException, IOException { + // XXX is this null check necessary? what happens if proxied, etc? + Recorder recorder = Recorder.getHttpRecorder(); + if (recorder != null) { + recorder.markContentBegin(); + } + + super.receiveResponseEntity(response); + } + }; + } + } + + +} \ No newline at end of file diff --git a/modules/src/main/java/org/archive/modules/fetcher/RecordingSocketInputBuffer.java b/modules/src/main/java/org/archive/modules/fetcher/RecordingSocketInputBuffer.java index 91eebcc5..341943b8 100644 --- a/modules/src/main/java/org/archive/modules/fetcher/RecordingSocketInputBuffer.java +++ b/modules/src/main/java/org/archive/modules/fetcher/RecordingSocketInputBuffer.java @@ -18,6 +18,7 @@ */ package org.archive.modules.fetcher; +import java.io.BufferedInputStream; import java.io.IOException; import java.io.InputStream; import java.net.Socket; @@ -33,19 +34,31 @@ import org.archive.util.Recorder; public class RecordingSocketInputBuffer implements SessionInputBuffer { protected Socket socket; - protected Recorder recorder; protected InputStream in; protected HttpTransportMetricsImpl metrics; - public RecordingSocketInputBuffer(Recorder rec, Socket socket, HttpParams params) throws IOException { + /** + * Gets recorder from current thread. + * @param socket + * @param buffersize + * @param params + * @throws IOException + */ + public RecordingSocketInputBuffer(Socket socket, int buffersize, HttpParams params) throws IOException { if (socket == null) { throw new IllegalArgumentException("Socket may not be null"); } - this.recorder = rec; this.socket = socket; - this.in = recorder.inputWrap(socket.getInputStream()); - this.metrics = new HttpTransportMetricsImpl(); + + Recorder recorder = Recorder.getHttpRecorder(); + Recorder httpRecorder = Recorder.getHttpRecorder(); + if (httpRecorder == null) { // XXX || (isSecure() && isProxied())) { + // no recorder, OR defer recording for pre-tunnel leg + this.in = new BufferedInputStream(socket.getInputStream(), buffersize); + } else { + this.in = recorder.inputWrap(new BufferedInputStream(socket.getInputStream(), buffersize)); + } } @Override diff --git a/modules/src/main/java/org/archive/modules/fetcher/RecordingSocketOutputBuffer.java b/modules/src/main/java/org/archive/modules/fetcher/RecordingSocketOutputBuffer.java new file mode 100644 index 00000000..2cde825c --- /dev/null +++ b/modules/src/main/java/org/archive/modules/fetcher/RecordingSocketOutputBuffer.java @@ -0,0 +1,184 @@ +/* + * This file is part of the Heritrix web crawler (crawler.archive.org). + * + * Licensed to the Internet Archive (IA) by one or more individual + * contributors. + * + * The IA licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.archive.modules.fetcher; + +import java.io.BufferedOutputStream; +import java.io.IOException; +import java.io.OutputStream; +import java.net.Socket; +import java.nio.ByteBuffer; +import java.nio.CharBuffer; +import java.nio.charset.Charset; +import java.nio.charset.CharsetEncoder; +import java.nio.charset.CoderResult; +import java.nio.charset.CodingErrorAction; + +import org.apache.http.impl.io.HttpTransportMetricsImpl; +import org.apache.http.io.HttpTransportMetrics; +import org.apache.http.io.SessionOutputBuffer; +import org.apache.http.params.HttpParams; +import org.apache.http.params.HttpProtocolParams; +import org.apache.http.protocol.HTTP; +import org.apache.http.util.CharArrayBuffer; +import org.archive.util.Recorder; + +public class RecordingSocketOutputBuffer implements SessionOutputBuffer { + + protected static final byte[] CRLF = new byte[] {HTTP.CR, HTTP.LF}; + protected static final Charset ASCII = Charset.forName("US-ASCII"); + + protected Socket socket; + protected HttpTransportMetricsImpl metrics; + protected OutputStream out; + protected Charset charset; + protected boolean ascii; + protected CharsetEncoder encoder; + protected CodingErrorAction onMalformedInputAction; + protected CodingErrorAction onUnMappableInputAction; + protected ByteBuffer bbuf; + + public RecordingSocketOutputBuffer(Socket socket, int buffersize, HttpParams params) throws IOException { + if (socket == null) { + throw new IllegalArgumentException("Socket may not be null"); + } + this.socket = socket; + this.metrics = new HttpTransportMetricsImpl(); + + this.charset = Charset.forName(HttpProtocolParams.getHttpElementCharset(params)); + this.ascii = this.charset.equals(ASCII); + this.onMalformedInputAction = HttpProtocolParams.getMalformedInputAction(params); + this.onUnMappableInputAction = HttpProtocolParams.getUnmappableInputAction(params); + + Recorder recorder = Recorder.getHttpRecorder(); + Recorder httpRecorder = Recorder.getHttpRecorder(); + if (httpRecorder == null) { // XXX || (isSecure() && isProxied())) { + // no recorder, OR defer recording for pre-tunnel leg + this.out = new BufferedOutputStream(socket.getOutputStream(), buffersize); + } else { + this.out = recorder.outputWrap(new BufferedOutputStream(socket.getOutputStream(), buffersize)); + } + } + + @Override + public void write(byte[] b, int off, int len) throws IOException { + if (b == null) { + return; + } + out.write(b, off, len); + metrics.incrementBytesTransferred(len - off); + } + + @Override + public void write(byte[] b) throws IOException { + if (b == null) { + return; + } + out.write(b); + metrics.incrementBytesTransferred(b.length); + } + + @Override + public void write(int b) throws IOException { + out.write(b); + metrics.incrementBytesTransferred(1); + } + + @Override + public void writeLine(String s) throws IOException { + if (s == null) { + return; + } + if (s.length() > 0) { + if (this.ascii) { + for (int i = 0; i < s.length(); i++) { + write(s.charAt(i)); + } + } else { + CharBuffer cbuf = CharBuffer.wrap(s); + writeEncoded(cbuf); + } + } + write(CRLF); + + } + + // copied verbatim from org.apache.http.impl.io.AbstractSessionOutputBuffer + protected void writeEncoded(final CharBuffer cbuf) throws IOException { + if (!cbuf.hasRemaining()) { + return; + } + if (this.encoder == null) { + this.encoder = this.charset.newEncoder(); + this.encoder.onMalformedInput(this.onMalformedInputAction); + this.encoder.onUnmappableCharacter(this.onUnMappableInputAction); + } + if (this.bbuf == null) { + this.bbuf = ByteBuffer.allocate(1024); + } + this.encoder.reset(); + while (cbuf.hasRemaining()) { + CoderResult result = this.encoder.encode(cbuf, this.bbuf, true); + handleEncodingResult(result); + } + CoderResult result = this.encoder.flush(this.bbuf); + handleEncodingResult(result); + this.bbuf.clear(); + } + + // copied verbatim from org.apache.http.impl.io.AbstractSessionOutputBuffer + protected void handleEncodingResult(final CoderResult result) throws IOException { + if (result.isError()) { + result.throwException(); + } + this.bbuf.flip(); + while (this.bbuf.hasRemaining()) { + write(this.bbuf.get()); + } + this.bbuf.compact(); + } + + + @Override + public void writeLine(CharArrayBuffer charbuffer) throws IOException { + if (charbuffer == null) { + return; + } + if (this.ascii) { + for (int i = 0; i < charbuffer.length(); i++) { + write(charbuffer.charAt(i)); + } + } else { + // XXX why is this wrapped in AbstractSessionOutputBuffer? copied from there + CharBuffer cbuf = CharBuffer.wrap(charbuffer.buffer(), 0, charbuffer.length()); + writeEncoded(cbuf); + } + write(CRLF); + } + + @Override + public void flush() throws IOException { + out.flush(); + } + + @Override + public HttpTransportMetrics getMetrics() { + return metrics; + } +}