lee/heritrix3
1
0
Fork 0
mirror of https://github.com/internetarchive/heritrix3.git synced 2026-09-23 22:26:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

New approach to overriding SessionInputBuffer seems promising.

Browse Source 
(httpcomponents.diff is the patch to httpcomponents head that is needed for this h3 commit to work, will keep up to date with each commit)
This commit is contained in:
Noah Levitt
2012-12-28 21:06:15 -08:00
parent 6300caf367
commit 6fe901d1f8
5 changed files with 1221 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
httpcomponents.diff
+1060
View File
File diff suppressed because it is too large Load Diff
modules/src/main/java/org/archive/modules/fetcher/FetchHTTP.java
+28 -36
View File
@@ -30,10 +30,7 @@ import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_REFERENCE_
import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_STATUS;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetAddress;
import java.net.Socket;
import java.net.UnknownHostException;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
@@ -99,9 +96,12 @@ import org.apache.http.impl.conn.DefaultClientConnectionFactory;
import org.apache.http.impl.conn.DefaultHttpResponseParserFactory;
import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
import org.apache.http.impl.conn.SocketClientConnectionImpl;
import org.apache.http.impl.io.HttpTransportMetricsImpl;
import org.apache.http.io.HttpMessageParserFactory;
import org.apache.http.io.HttpMessageWriterFactory;
import org.apache.http.io.SessionInputBuffer;
import org.apache.http.io.UmmSessionBufferFactory;
import org.apache.http.io.UmmSessionInputBuffer;
import org.apache.http.io.UmmSessionOutputBuffer;
import org.apache.http.message.BasicHeader;
import org.apache.http.message.BasicNameValuePair;
import org.apache.http.protocol.HTTP;
@@ -133,6 +133,27 @@ import org.springframework.context.Lifecycle;
*/
public class FetchHTTP extends Processor implements Lifecycle {
protected static class RecordingSessionBufferFactory implements UmmSessionBufferFactory {
protected static final RecordingSessionBufferFactory INSTANCE = new RecordingSessionBufferFactory();
@Override
public UmmSessionInputBuffer createInputBuffer(
HttpTransportMetricsImpl metrics, int buffersize,
int minChunkLimit, MessageConstraints constraints,
CharsetDecoder chardecoder) {
return new RecordingSessionInputBuffer(metrics, buffersize, minChunkLimit, constraints, chardecoder);
}
@Override
public UmmSessionOutputBuffer createOutputBuffer(
HttpTransportMetricsImpl metrics, int buffersize, int minChunkLimit,
CharsetEncoder charencoder) {
return new RecordingSessionOutputBuffer(metrics, buffersize, minChunkLimit, charencoder);
}
}
protected class RecordingSocketClientConnection extends
SocketClientConnectionImpl {
private final AbortableHttpRequestBase request;
@@ -146,43 +167,14 @@ public class FetchHTTP extends Processor implements Lifecycle {
ContentLengthStrategy outgoingContentStrategy,
HttpMessageWriterFactory<HttpRequest> requestWriterFactory,
HttpMessageParserFactory<HttpResponse> responseParserFactory,
AbortableHttpRequestBase request, CrawlURI curi) {
UmmSessionBufferFactory sessionBufferFactory, AbortableHttpRequestBase request, CrawlURI curi) {
super(buffersize, chardecoder, charencoder, constraints,
incomingContentStrategy, outgoingContentStrategy,
requestWriterFactory, responseParserFactory);
requestWriterFactory, responseParserFactory, sessionBufferFactory);
this.request = request;
this.curi = curi;
}
@Override
protected SessionInputBuffer getSessionInputBuffer() {
return new RecordingSessionInputBuffer(super.getSessionInputBuffer());
}
@Override
protected InputStream getSocketInputStream(Socket socket)
throws IOException {
logger.info("socket=" + socket);
Recorder recorder = Recorder.getHttpRecorder();
if (recorder != null) { // XXX || (isSecure() && isProxied())) {
return recorder.inputWrap(super.getSocketInputStream(socket));
} else {
return super.getSocketInputStream(socket);
}
}
@Override
protected OutputStream getSocketOutputStream(
Socket socket) throws IOException {
logger.info("socket=" + socket);
Recorder recorder = Recorder.getHttpRecorder();
if (recorder != null) { // XXX || (isSecure() && isProxied())) {
return recorder.outputWrap(super.getSocketOutputStream(socket));
} else {
return super.getSocketOutputStream(socket);
}
}
@Override
public void receiveResponseEntity(HttpResponse response)
throws HttpException, IOException {
@@ -895,6 +887,7 @@ public class FetchHTTP extends Processor implements Lifecycle {
return new RecordingSocketClientConnection(8 * 1024,
chardecoder, charencoder, messageConstraints, null,
null, null, DefaultHttpResponseParserFactory.INSTANCE,
RecordingSessionBufferFactory.INSTANCE,
request, curi);
}
};
@@ -907,7 +900,6 @@ public class FetchHTTP extends Processor implements Lifecycle {
// builder.setSSLSocketFactory(sslContext())
// builder.setCredentialsProvider(null)
return builder.build();
}
protected void populateHttpProxyCredential(CrawlURI curi,
modules/src/main/java/org/archive/modules/fetcher/RecordingSessionInputBuffer.java
+77 -54
View File
@@ -18,73 +18,96 @@
*/
package org.archive.modules.fetcher;
import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.CharsetDecoder;
import org.apache.http.io.HttpTransportMetrics;
import org.apache.http.io.SessionInputBuffer;
import org.apache.http.config.MessageConstraints;
import org.apache.http.impl.io.HttpTransportMetricsImpl;
import org.apache.http.impl.io.SessionInputBufferImpl;
import org.apache.http.protocol.HTTP;
import org.apache.http.util.Args;
import org.apache.http.util.CharArrayBuffer;
import org.archive.util.Recorder;
class RecordingSessionInputBuffer implements SessionInputBuffer {
class RecordingSessionInputBuffer extends SessionInputBufferImpl {
protected SessionInputBuffer wrapped;
protected int buffersize;
public RecordingSessionInputBuffer(SessionInputBuffer wrapped) {
this.wrapped = wrapped;
public RecordingSessionInputBuffer(HttpTransportMetricsImpl metrics,
int buffersize, int minChunkLimit, MessageConstraints constraints,
CharsetDecoder chardecoder) {
super(metrics, buffersize, minChunkLimit, constraints, chardecoder);
this.buffersize = buffersize;
}
@Override
public int read(byte[] b, int off, int len) throws IOException {
return wrapped.read(b, off, len);
public void bind(InputStream inputStream) throws IOException {
if (inputStream != null) {
Recorder recorder = Recorder.getHttpRecorder();
if (recorder == null) { // XXX || (isSecure() && isProxied())) {
// no recorder, OR defer recording for pre-tunnel leg
this.instream = new BufferedInputStream(inputStream, buffersize);
} else {
this.instream = recorder.inputWrap(new BufferedInputStream(inputStream, buffersize));
}
} else {
this.instream = null;
}
}
@Override
public int read(byte[] b) throws IOException {
return wrapped.read(b);
}
@Override
public int read() throws IOException {
return wrapped.read();
}
@Override
public int readLine(CharArrayBuffer buffer) throws IOException {
return wrapped.readLine(buffer);
}
@Override
public String readLine() throws IOException {
return wrapped.readLine();
}
@SuppressWarnings("deprecation")
@Override
public boolean isDataAvailable(int timeout) throws IOException {
return wrapped.isDataAvailable(timeout);
}
@Override
public HttpTransportMetrics getMetrics() {
return wrapped.getMetrics();
}
// @Override
// public boolean isBound() {
// return wrapped.isBound();
// }
//
// @Override
// public void bind(InputStream inputStream) {
// wrapped.bind(inputStream);
// }
//
// @Override
// public boolean hasBufferedData() {
// return wrapped.hasBufferedData();
// throw new RuntimeException("implement me i guess");
// if (this.instream != null) {
// try {
// return in.available() > 0;
// } catch (IOException e) {
// return false;
// }
// } else if (this.instream instanceof BufferedInputStream) {
// BufferedInputStream in = (BufferedInputStream) this.instream;
// } else if (this.instream instanceof RecordingInputStream) {
// RecordingInputStream in = (RecordingInputStream) this.instream;
// return in.
// }
// }
//
// @Override
// public int fillBuffer() throws IOException {
// return wrapped.fillBuffer();
// }
@Override
public int fillBuffer() throws IOException {
throw new RuntimeException("don't use me");
}
@Override
public int readLine(CharArrayBuffer charbuffer) throws IOException {
Args.notNull(charbuffer, "Char array buffer");
int bytesRead = 0;
int b = instream.read();
while (b >= 0 && b != HTTP.LF) {
bytesRead++;
linebuffer.append(b);
b = instream.read();
}
if (b >= 0) {
bytesRead++; // count LF
}
// if line ends with CR-LF, get rid of the CR
if (bytesRead > 0 && linebuffer.byteAt(linebuffer.length() - 1) == HTTP.CR) {
linebuffer.setLength(linebuffer.length() - 1);
}
if (bytesRead > 0) {
metrics.incrementBytesTransferred(bytesRead);
}
if (bytesRead == 0 && b == -1) {
// indicate the end of stream
return -1;
}
return lineFromLineBuffer(charbuffer);
}
}
modules/src/main/java/org/archive/modules/fetcher/RecordingSessionOutputBuffer.java
+55
View File
@@ -0,0 +1,55 @@
/*
* 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.nio.charset.CharsetEncoder;
import org.apache.http.impl.io.HttpTransportMetricsImpl;
import org.apache.http.impl.io.SessionOutputBufferImpl;
import org.archive.util.Recorder;
class RecordingSessionOutputBuffer extends SessionOutputBufferImpl {
protected int buffersize;
public RecordingSessionOutputBuffer(HttpTransportMetricsImpl metrics,
int buffersize, int minChunkLimit, CharsetEncoder charencoder) {
super(metrics, buffersize, minChunkLimit, charencoder);
this.buffersize = buffersize;
}
@Override
public void bind(OutputStream outstream) throws IOException {
if (outstream != null) {
Recorder recorder = Recorder.getHttpRecorder();
if (recorder == null) { // XXX || (isSecure() && isProxied())) {
// no recorder, OR defer recording for pre-tunnel leg
this.outstream = new BufferedOutputStream(outstream, buffersize);
} else {
this.outstream = recorder.outputWrap(new BufferedOutputStream(outstream, buffersize));
}
} else {
this.outstream = null;
}
}
}
modules/src/test/java/org/archive/modules/fetcher/FetchHTTPTests.java
+1 -1
View File
@@ -197,7 +197,7 @@ public class FetchHTTPTests extends ProcessorTestBase {
public void testDefaults() throws Exception {
CrawlURI curi = makeCrawlURI("http://localhost:7777/");
fetcher().process(curi);
logger.info('\n' + httpRequestString(curi) + contentString(curi));
logger.info('\n' + httpRequestString(curi) + rawResponseString(curi));
runDefaultChecks(curi);
}