do recording through SessionOutputBuffer implementation like we have for SessionInputBuffer; refactor helper classes into their own file RecordingHttpClient.java

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);
-        }
-    }
 }

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);
+                }
+            };
+        }
+    }
+
+
+}

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

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;
+    }
+}