From c5eb3ffd538efde455c7f72444094a890fe318a5 Mon Sep 17 00:00:00 2001 From: gojomo Date: Wed, 6 Apr 2011 20:42:43 +0000 Subject: [PATCH] [HER-1053] compressed HTTP fetch: "Accept-encoding: gzip" [HER-728] Offer replay stream that has been un-chunked (whether because response was HTTP/1.1 or used chunked in HTTP/1.0 against spec) [HER-1876] Offer HTTP/1.1 option - for chunked transfer-encoding (but not persistent connections) * FetchHTTP add 'acceptCompression' and 'useHTTP11' properties, both default false * Recorder track whether recorded-input is transfer-encoded (chunked) or content-encoded (gzip etc) offer alternate replay streams for (1) raw 'messageBody'; (2) entity (un-chunked if necessary) (3) content (decompressed if necessary) always content & GenericReplayCharSequence for CharSequence replays * GenericReplayCharSequence always use a stream (rather than random-access buffer) always decode to prefix buffer first, so short content never touches disk no matter the encoding * InMemoryReplayCharSequence deleted; 'Generic' now works similarly for small content and anyway random-access for single-byte-encodings is now rarely possible (given deconding streams) * RecordingInputStream, RecordingOutputStream adjust for changed stream names, functionality moved to Recorder * ReplayCharSequence use Charset instances rather than names * ReplayInputStream add convenience constructor (and tmp-file-destroy) for copying any other inputStream into a seekable ReplayInputStream --- .../archive/io/GenericReplayCharSequence.java | 225 ++++++++---------- .../io/InMemoryReplayCharSequence.java | 165 ------------- .../org/archive/io/RecordingInputStream.java | 38 +-- .../org/archive/io/RecordingOutputStream.java | 85 ++----- .../org/archive/io/ReplayCharSequence.java | 6 +- .../org/archive/io/ReplayInputStream.java | 46 +++- .../main/java/org/archive/util/Recorder.java | 198 ++++++++++++++- .../archive/modules/fetcher/FetchHTTP.java | 68 +++++- 8 files changed, 427 insertions(+), 404 deletions(-) delete mode 100644 commons/src/main/java/org/archive/io/InMemoryReplayCharSequence.java diff --git a/commons/src/main/java/org/archive/io/GenericReplayCharSequence.java b/commons/src/main/java/org/archive/io/GenericReplayCharSequence.java index d61d729b..9294e08a 100644 --- a/commons/src/main/java/org/archive/io/GenericReplayCharSequence.java +++ b/commons/src/main/java/org/archive/io/GenericReplayCharSequence.java @@ -20,26 +20,29 @@ package org.archive.io; import java.io.BufferedReader; -import java.io.BufferedWriter; import java.io.File; import java.io.FileInputStream; import java.io.FileNotFoundException; import java.io.FileOutputStream; import java.io.IOException; +import java.io.InputStream; import java.io.InputStreamReader; import java.io.OutputStreamWriter; -import java.nio.ByteBuffer; +import java.io.Writer; import java.nio.CharBuffer; import java.nio.channels.FileChannel; import java.nio.charset.CharacterCodingException; import java.nio.charset.Charset; -import java.nio.charset.CharsetDecoder; import java.text.NumberFormat; import java.util.logging.Level; import java.util.logging.Logger; +import org.apache.commons.io.IOUtils; import org.archive.util.DevUtils; +import com.google.common.base.Charsets; +import com.google.common.primitives.Ints; + /** * (Replay)CharSequence view on recorded streams. * @@ -47,8 +50,8 @@ import org.archive.util.DevUtils; * *

Call {@link close()} on this class when done to clean up resources. * - * @author stack - * @author nlevitt + * @contributor stack + * @contributor nlevitt * @version $Revision$, $Date$ */ public class GenericReplayCharSequence implements ReplayCharSequence { @@ -66,7 +69,7 @@ public class GenericReplayCharSequence implements ReplayCharSequence { * *

See Encoding. */ - private static final String WRITE_ENCODING = "UTF-16BE"; + public static final Charset WRITE_ENCODING = Charsets.UTF_16BE; private static final long MAP_MAX_BYTES = 64 * 1024 * 1024; // 64M @@ -78,7 +81,7 @@ public class GenericReplayCharSequence implements ReplayCharSequence { * MAP_MAX_BYTES - MAP_TARGET_LEFT_PADDING * bytes to the right of the target. */ - private static final long MAP_TARGET_LEFT_PADDING_BYTES = (long) (MAP_MAX_BYTES * 0.2); + private static final long MAP_TARGET_LEFT_PADDING_BYTES = (long) (MAP_MAX_BYTES * 0.01); /** * Total length of character stream to replay minus the HTTP headers @@ -108,11 +111,7 @@ public class GenericReplayCharSequence implements ReplayCharSequence { private long bytesPerChar; - private ByteBuffer mappedBuffer = null; - - private CharsetDecoder decoder = null; - - private ByteBuffer tempBuf = null; + private CharBuffer mappedBuffer = null; /** * File that has decoded content. @@ -133,90 +132,53 @@ public class GenericReplayCharSequence implements ReplayCharSequence { /** * Constructor. * - * @param buffer In-memory buffer of recordings prefix. We read from - * here first and will only go to the backing file if size - * requested is greater than buffer.length. - * @param size Total size of stream to replay in bytes. Used to find - * EOS. This is total length of content including HTTP headers if - * present. * @param contentReplayInputStream inputStream of content - * @param charsetName Encoding to use reading the passed prefix - * buffer and backing file. For now, should be java canonical name for the - * encoding. Must not be null. + * @param charset Encoding to use reading the passed prefix + * buffer and backing file. Must not be null. * @param backingFilename Path to backing file with content in excess of * whats in buffer. * * @throws IOException */ - public GenericReplayCharSequence( - ReplayInputStream contentReplayInputStream, String backingFilename, - String charsetName) throws IOException { + public GenericReplayCharSequence(InputStream contentReplayInputStream, + int prefixMax, + String backingFilename, + Charset charset) throws IOException { super(); logger.fine("new GenericReplayCharSequence() characterEncoding=" - + charsetName + " backingFilename=" + backingFilename); - Charset charset; - try { - charset = Charset.forName(charsetName); - } catch (IllegalArgumentException e) { - logger.log(Level.WARNING,"charset problem: "+charsetName,e); - // TODO: better detection or default - charset = Charset.forName(FALLBACK_CHARSET_NAME); + + charset + " backingFilename=" + backingFilename); + + if(charset==null) { + charset = ReplayCharSequence.FALLBACK_CHARSET; } - if (charset.newEncoder().maxBytesPerChar() == 1.0) { - logger.fine("charset=" + charsetName - + ": supports random access, using backing file directly"); - this.bytesPerChar = 1; - this.backingFileIn = new FileInputStream(backingFilename); - this.decoder = charset.newDecoder(); - this.prefixBuffer = this.decoder.decode( - ByteBuffer.wrap(contentReplayInputStream.getBuffer())); - } else { - logger.fine("charset=" + charsetName - + ": may not support random access, decoding to separate file"); + // decodes only up to Integer.MAX_VALUE characters + decode(contentReplayInputStream, prefixMax, backingFilename, charset); - // decodes only up to Integer.MAX_VALUE characters - decodeToFile(contentReplayInputStream, backingFilename, charsetName); - - this.bytesPerChar = 2; - this.backingFileIn = new FileInputStream(decodedFile); - this.decoder = Charset.forName(WRITE_ENCODING).newDecoder(); - this.prefixBuffer = CharBuffer.wrap(""); - } - - this.tempBuf = ByteBuffer.wrap(new byte[(int) this.bytesPerChar]); - this.backingFileChannel = backingFileIn.getChannel(); + this.bytesPerChar = 2; - // we only support the first Integer.MAX_VALUE characters - long wouldBeLength = prefixBuffer.limit() + backingFileChannel.size() / bytesPerChar; - if (wouldBeLength <= Integer.MAX_VALUE) { - this.length = (int) wouldBeLength; - } else { - logger.warning("input stream is longer than Integer.MAX_VALUE=" - + NumberFormat.getInstance().format(Integer.MAX_VALUE) - + " characters -- only first " - + NumberFormat.getInstance().format(Integer.MAX_VALUE) - + " are accessible through this GenericReplayCharSequence"); - this.length = Integer.MAX_VALUE; + if(length>prefixBuffer.position()) { + this.backingFileIn = new FileInputStream(decodedFile); + this.backingFileChannel = backingFileIn.getChannel(); + this.mapByteOffset = 0; + updateMemoryMappedBuffer(); } - - this.mapByteOffset = 0; - updateMemoryMappedBuffer(); } private void updateMemoryMappedBuffer() { - long fileLength = (long) this.length() - (long) prefixBuffer.limit(); // in characters - long mapSize = Math.min(fileLength * bytesPerChar - mapByteOffset, MAP_MAX_BYTES); + long charLength = (long) this.length() - (long) prefixBuffer.limit(); // in characters + long mapSize = Math.min((charLength * bytesPerChar) - mapByteOffset, MAP_MAX_BYTES); logger.fine("updateMemoryMappedBuffer: mapOffset=" + NumberFormat.getInstance().format(mapByteOffset) + " mapSize=" + NumberFormat.getInstance().format(mapSize)); try { - System.gc(); - System.runFinalization(); + // TODO: stress-test without these possibly-costly requests! +// System.gc(); +// System.runFinalization(); // TODO: Confirm the READ_ONLY works. I recall it not working. // The buffers seem to always say that the buffer is writable. mappedBuffer = backingFileChannel.map( FileChannel.MapMode.READ_ONLY, mapByteOffset, mapSize) - .asReadOnlyBuffer(); + .asReadOnlyBuffer().asCharBuffer(); } catch (IOException e) { // TODO convert this to a runtime error? DevUtils.logger.log(Level.SEVERE, @@ -237,46 +199,65 @@ public class GenericReplayCharSequence implements ReplayCharSequence { * * @throws IOException */ - private void decodeToFile(ReplayInputStream inStream, - String backingFilename, String encoding) throws IOException { + protected void decode(InputStream inStream, int prefixMax, + String backingFilename, Charset charset) throws IOException { + // TODO: consider if BufferedReader is helping any + // TODO: consider adding TBW 'LimitReader' to stop reading at + // Integer.MAX_VALUE characters because of charAt(int) limit BufferedReader reader = new BufferedReader(new InputStreamReader( - inStream, encoding)); - - this.decodedFile = new File(backingFilename + "." + WRITE_ENCODING); + inStream, charset)); logger.fine("decodeToFile: backingFilename=" + backingFilename - + " encoding=" + encoding + " decodedFile=" + decodedFile); + + " encoding=" + charset + " decodedFile=" + decodedFile); - FileOutputStream fos; - try { - fos = new FileOutputStream(this.decodedFile); - } catch (FileNotFoundException e) { - // Windows workaround attempt - System.gc(); - System.runFinalization(); - this.decodedFile = new File(decodedFile.getAbsolutePath()+".win"); - logger.info("Windows 'file with a user-mapped section open' " - + "workaround gc/finalization/name-extension performed."); - // try again - fos = new FileOutputStream(this.decodedFile); - } - BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(fos, - WRITE_ENCODING)); - - int c; + this.prefixBuffer = CharBuffer.allocate(prefixMax); + long count = 0; - while ((c = reader.read()) >= 0 && count < Integer.MAX_VALUE) { - writer.write(c); - count++; - if (count % 100000000 == 0) { - logger.fine("wrote " + count + " characters so far..."); + while(count < prefixMax) { + int read = reader.read(prefixBuffer); + if(read<0) { + break; } + count += read; } - writer.close(); + + if(reader.ready()) { + // more to decode to file overflow + this.decodedFile = new File(backingFilename + "." + WRITE_ENCODING); - logger.fine("decodeToFile: wrote " + count + " characters to " - + decodedFile); + FileOutputStream fos; + try { + fos = new FileOutputStream(this.decodedFile); + } catch (FileNotFoundException e) { + // Windows workaround attempt + System.gc(); + System.runFinalization(); + this.decodedFile = new File(decodedFile.getAbsolutePath()+".win"); + logger.info("Windows 'file with a user-mapped section open' " + + "workaround gc/finalization/name-extension performed."); + // try again + fos = new FileOutputStream(this.decodedFile); + } + + Writer writer = new OutputStreamWriter(fos,WRITE_ENCODING); + count += IOUtils.copyLarge(reader, writer); + writer.close(); + reader.close(); + } + + this.length = Ints.saturatedCast(count); + if(count>Integer.MAX_VALUE) { + logger.warning("input stream is longer than Integer.MAX_VALUE=" + + NumberFormat.getInstance().format(Integer.MAX_VALUE) + + " characters -- only first " + + NumberFormat.getInstance().format(Integer.MAX_VALUE) + + " are accessible through this GenericReplayCharSequence"); + } + + logger.fine("decode: decoded " + count + " characters" + + ((decodedFile==null) ? "" + : " ("+(count-prefixBuffer.length())+" to "+decodedFile+")")); } /** @@ -296,10 +277,13 @@ public class GenericReplayCharSequence implements ReplayCharSequence { } // otherwise we gotta get it from disk via memory map - long fileIndex = (long) index - (long) prefixBuffer.limit(); - long fileLength = (long) this.length() - (long) prefixBuffer.limit(); // in characters - if (fileIndex * bytesPerChar < mapByteOffset - || fileIndex * bytesPerChar - mapByteOffset >= mappedBuffer.limit()) { + long charFileIndex = (long) index - (long) prefixBuffer.limit(); + long charFileLength = (long) this.length() - (long) prefixBuffer.limit(); // in characters + if (charFileIndex * bytesPerChar < mapByteOffset) { + logger.log(Level.WARNING,"left-fault; probably don't want to use CharSequence that far backward"); + } + if (charFileIndex * bytesPerChar < mapByteOffset + || charFileIndex - (mapByteOffset / bytesPerChar) >= mappedBuffer.limit()) { // fault /* * mapByteOffset is bounded by 0 and file size +/- size of the map, @@ -307,30 +291,13 @@ public class GenericReplayCharSequence implements ReplayCharSequence { * MAP_TARGET_LEFT_PADDING_BYTES as it can while also not * being smaller than it needs to be. */ - mapByteOffset = Math.max(0, fileIndex * bytesPerChar - MAP_TARGET_LEFT_PADDING_BYTES); - mapByteOffset = Math.min(mapByteOffset, fileLength * bytesPerChar - MAP_MAX_BYTES); + mapByteOffset = Math.min(charFileIndex * bytesPerChar - MAP_TARGET_LEFT_PADDING_BYTES, + charFileLength * bytesPerChar - MAP_MAX_BYTES); + mapByteOffset = Math.max(0, mapByteOffset); updateMemoryMappedBuffer(); } - // CharsetDecoder always decodes up to the end of the ByteBuffer, so we - // create a new ByteBuffer with only the bytes we're interested in. - mappedBuffer.position((int) (fileIndex * bytesPerChar - mapByteOffset)); - mappedBuffer.get(tempBuf.array()); - tempBuf.position(0); // decoder starts at this position - - try { - CharBuffer cbuf = decoder.decode(tempBuf); - return cbuf.get(); - } catch (CharacterCodingException e) { - logger.log(Level.FINE,"unable to get character at index=" + index + " (fileIndex=" + fileIndex + "): " + e, e); - decodingExceptions++; - if(codingException==null) { - codingException = e; - } - // U+FFFD REPLACEMENT CHARACTER -- - // "used to replace an incoming character whose value is unknown or unrepresentable in Unicode" - return (char) 0xfffd; - } + return mappedBuffer.get((int)(charFileIndex-(mapByteOffset/bytesPerChar))); } public CharSequence subSequence(int start, int end) { diff --git a/commons/src/main/java/org/archive/io/InMemoryReplayCharSequence.java b/commons/src/main/java/org/archive/io/InMemoryReplayCharSequence.java deleted file mode 100644 index 41f80509..00000000 --- a/commons/src/main/java/org/archive/io/InMemoryReplayCharSequence.java +++ /dev/null @@ -1,165 +0,0 @@ -/* - * 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.io; - -import java.io.IOException; -import java.nio.ByteBuffer; -import java.nio.CharBuffer; -import java.nio.charset.CharacterCodingException; -import java.nio.charset.Charset; -import java.nio.charset.CodingErrorAction; -import java.util.logging.Level; -import java.util.logging.Logger; - -public class InMemoryReplayCharSequence implements ReplayCharSequence { - - protected static Logger logger = - Logger.getLogger(InMemoryReplayCharSequence.class.getName()); - - /** - * CharBuffer of decoded content. - * - * Content of this buffer is unicode. - */ - private CharBuffer charBuffer = null; - - protected long decodingExceptionsCount = 0; - protected CharacterCodingException codingException = null; - - /** - * Constructor for all in-memory operation. - * - * @param buffer - * @param size Total size of stream to replay in bytes. Used to find - * EOS. This is total length of content including HTTP headers if - * present. - * @param responseBodyStart Where the response body starts in bytes. - * Used to skip over the HTTP headers if present. - * @param encoding Encoding to use reading the passed prefix buffer and - * backing file. For now, should be java canonical name for the - * encoding. Must not be null. - * - * @throws IOException - */ - public InMemoryReplayCharSequence(byte[] buffer, long size, - long responseBodyStart, String encoding) throws IOException { - super(); - this.charBuffer = decodeInMemory(buffer, size, responseBodyStart, - encoding); - } - - /** - * Decode passed buffer into a CharBuffer. - * - * This method decodes a memory buffer returning a memory buffer. - * - * @param buffer - * @param size Total size of stream to replay in bytes. Used to find - * EOS. This is total length of content including HTTP headers if - * present. - * @param responseBodyStart Where the response body starts in bytes. - * Used to skip over the HTTP headers if present. - * @param encoding Encoding to use reading the passed prefix buffer and - * backing file. For now, should be java canonical name for the - * encoding. Must not be null. - * - * @return A CharBuffer view on decodings of the contents of passed - * buffer. - */ - private CharBuffer decodeInMemory(byte[] buffer, long size, - long responseBodyStart, String encoding) { - ByteBuffer bb = ByteBuffer.wrap(buffer); - // Move past the HTTP header if present. - bb.position((int) responseBodyStart); - bb.mark(); - // Set the end-of-buffer to be end-of-content. - bb.limit((int) size); - Charset charset; - try { - charset = Charset.forName(encoding); - } catch (IllegalArgumentException e) { - logger.log(Level.WARNING,"charset problem: "+encoding,e); - // TODO: better detection or default - charset = Charset.forName(FALLBACK_CHARSET_NAME); - } - try { - return charset.newDecoder() - .onMalformedInput(CodingErrorAction.REPORT) - .onUnmappableCharacter(CodingErrorAction.REPORT) - .decode(bb).asReadOnlyBuffer(); - } catch (CharacterCodingException cce) { - bb.reset(); - decodingExceptionsCount++; - codingException = cce; - return charset.decode(bb).asReadOnlyBuffer(); - } - } - - public void close() { - this.charBuffer = null; - } - - protected void finalize() throws Throwable { - super.finalize(); - // Maybe TODO: eliminate close here, requiring explicit close instead - close(); - } - - public int length() { - return this.charBuffer.limit(); - } - - public char charAt(int index) { - return this.charBuffer.get(index); - } - - public CharSequence subSequence(int start, int end) { - return new CharSubSequence(this, start, end); - } - - public String toString() { - StringBuffer sb = new StringBuffer(length()); - sb.append(this); - return sb.toString(); - } - - /** - * Return 1 if there were decoding problems (a full count isn't possible). - * - * @see org.archive.io.ReplayCharSequence#getDecodeExceptionCount() - */ - @Override - public long getDecodeExceptionCount() { - return decodingExceptionsCount; - } - - /* (non-Javadoc) - * @see org.archive.io.ReplayCharSequence#getCodingException() - */ - @Override - public CharacterCodingException getCodingException() { - return codingException; - } - - @Override - public boolean isOpen() { - return this.charBuffer != null; - } -} diff --git a/commons/src/main/java/org/archive/io/RecordingInputStream.java b/commons/src/main/java/org/archive/io/RecordingInputStream.java index 9f3991cc..f28f3d27 100644 --- a/commons/src/main/java/org/archive/io/RecordingInputStream.java +++ b/commons/src/main/java/org/archive/io/RecordingInputStream.java @@ -18,8 +18,6 @@ */ package org.archive.io; -import java.io.File; -import java.io.FileOutputStream; import java.io.IOException; import java.io.InputStream; import java.net.SocketException; @@ -149,8 +147,8 @@ public class RecordingInputStream return this.recordingOutputStream.getReplayInputStream(); } - public ReplayInputStream getContentReplayInputStream() throws IOException { - return this.recordingOutputStream.getContentReplayInputStream(); + public ReplayInputStream getMessageBodyReplayInputStream() throws IOException { + return this.recordingOutputStream.getMessageBodyReplayInputStream(); } public long readFully() throws IOException { @@ -249,11 +247,11 @@ public class RecordingInputStream } public void markContentBegin() { - this.recordingOutputStream.markContentBegin(); + this.recordingOutputStream.markMessageBodyBegin(); } public long getContentBegin() { - return this.recordingOutputStream.getContentBegin(); + return this.recordingOutputStream.getMessageBodyBegin(); } public void startDigest() { @@ -302,22 +300,6 @@ public class RecordingInputStream return this.recordingOutputStream.getDigestValue(); } - public ReplayCharSequence getReplayCharSequence() throws IOException { - return getReplayCharSequence(null); - } - - /** - * @param characterEncoding Encoding of recorded stream. - * @return A ReplayCharSequence Will return null if an IOException. Call - * close on returned RCS when done. - * @throws IOException - */ - public ReplayCharSequence getReplayCharSequence(String characterEncoding) - throws IOException { - return this.recordingOutputStream. - getReplayCharSequence(characterEncoding); - } - public long getResponseContentLength() { return this.recordingOutputStream.getResponseContentLength(); } @@ -326,18 +308,6 @@ public class RecordingInputStream this.recordingOutputStream.closeRecorder(); } - /** - * @param tempFile - * @throws IOException - */ - public void copyContentBodyTo(File tempFile) throws IOException { - FileOutputStream fos = new FileOutputStream(tempFile); - ReplayInputStream ris = getContentReplayInputStream(); - ris.readFullyTo(fos); - fos.close(); - ris.close(); - } - /** * @return True if we've been opened. */ diff --git a/commons/src/main/java/org/archive/io/RecordingOutputStream.java b/commons/src/main/java/org/archive/io/RecordingOutputStream.java index ab211e69..d3756da0 100644 --- a/commons/src/main/java/org/archive/io/RecordingOutputStream.java +++ b/commons/src/main/java/org/archive/io/RecordingOutputStream.java @@ -70,10 +70,10 @@ public class RecordingOutputStream extends OutputStream { * Later passed to ReplayInputStream on creation. It uses it to know when * EOS. */ - private long size = 0; + protected long size = 0; - private String backingFilename; - private OutputStream diskStream = null; + protected String backingFilename; + protected OutputStream diskStream = null; /** * Buffer we write recordings to. @@ -106,7 +106,7 @@ public class RecordingOutputStream extends OutputStream { private MessageDigest digest = null; /** - * Define for SHA1 alogarithm. + * Define for SHA1 algarithm. */ private static final String SHA1 = "SHA1"; @@ -130,7 +130,7 @@ public class RecordingOutputStream extends OutputStream { /** * When recording HTTP, where the content-body starts. */ - private long contentBeginMark; + protected long messageBodyBeginMark; /** * Stream to record. @@ -186,7 +186,7 @@ public class RecordingOutputStream extends OutputStream { this.markPosition = 0; this.maxPosition = 0; this.size = 0; - this.contentBeginMark = -1; + this.messageBodyBeginMark = -1; // ensure recording turned on this.recording = true; // Always begins false; must use startDigest() to begin @@ -245,7 +245,7 @@ public class RecordingOutputStream extends OutputStream { */ protected void checkLimits() throws RecorderIOException { // too much material before finding end of headers? - if (contentBeginMark<0) { + if (messageBodyBeginMark<0) { // no mark yet if(position>MAX_HEADER_MATERIAL) { throw new RecorderTooMuchHeaderException(); @@ -344,10 +344,10 @@ public class RecordingOutputStream extends OutputStream { } public void close() throws IOException { - if(contentBeginMark<0) { + if(messageBodyBeginMark<0) { // if unset, consider 0 posn as content-start // (so that a -1 never survives to replay step) - contentBeginMark = 0; + messageBodyBeginMark = 0; } if (this.out != null) { this.out.close(); @@ -396,7 +396,7 @@ public class RecordingOutputStream extends OutputStream { // -- the size will zero so any attempt at a read will get back EOF. assert this.out == null: "Stream is still open."; ReplayInputStream replay = new ReplayInputStream(this.buffer, - this.size, this.contentBeginMark, this.backingFilename); + this.size, this.messageBodyBeginMark, this.backingFilename); replay.skip(skip); return replay; } @@ -407,8 +407,8 @@ public class RecordingOutputStream extends OutputStream { * @throws IOException * @return An RIS. */ - public ReplayInputStream getContentReplayInputStream() throws IOException { - return getReplayInputStream(this.contentBeginMark); + public ReplayInputStream getMessageBodyReplayInputStream() throws IOException { + return getReplayInputStream(this.messageBodyBeginMark); } public long getSize() { @@ -416,20 +416,20 @@ public class RecordingOutputStream extends OutputStream { } /** - * Remember the current position as the start of the "response + * Remember the current position as the start of the "message * body". Useful when recording HTTP traffic as a way to start * replays after the headers. */ - public void markContentBegin() { - this.contentBeginMark = this.position; + public void markMessageBodyBegin() { + this.messageBodyBeginMark = this.position; startDigest(); } /** - * Return stored content-begin-mark (which is also end-of-headers) + * Return stored message-body-begin-mark (which is also end-of-headers) */ - public long getContentBegin() { - return this.contentBeginMark; + public long getMessageBodyBegin() { + return this.messageBodyBeginMark; } /** @@ -499,51 +499,8 @@ public class RecordingOutputStream extends OutputStream { return this.digest.digest(); } - public ReplayCharSequence getReplayCharSequence() throws IOException { - return getReplayCharSequence(null); - } - - public ReplayCharSequence getReplayCharSequence(String characterEncoding) - throws IOException { - return getReplayCharSequence(characterEncoding, this.contentBeginMark); - } - - /** - * @param characterEncoding Encoding of recorded stream. - * @return A ReplayCharSequence Will return null if an IOException. Call - * close on returned RCS when done. - * @throws IOException - */ - public ReplayCharSequence getReplayCharSequence(String characterEncoding, - long startOffset) throws IOException { - if (characterEncoding == null) { - characterEncoding = "UTF-8"; - } - logger.fine("this.size=" + this.size + " this.buffer.length=" + this.buffer.length); - if (this.size <= this.buffer.length) { - logger.fine("using InMemoryReplayCharSequence"); - // raw data is all in memory; do in memory - return new InMemoryReplayCharSequence( - this.buffer, - this.size, - startOffset, - characterEncoding); - } - else { - logger.fine("using GenericReplayCharSequence"); - // raw data overflows to disk; use temp file - ReplayInputStream ris = getReplayInputStream(startOffset); - ReplayCharSequence rcs = new GenericReplayCharSequence( - ris, - this.backingFilename, - characterEncoding); - ris.close(); - return rcs; - } - } - public long getResponseContentLength() { - return this.size - this.contentBeginMark; + return this.size - this.messageBodyBeginMark; } /** @@ -553,6 +510,10 @@ public class RecordingOutputStream extends OutputStream { return this.out != null; } + public int getBufferLength() { + return this.buffer.length; + } + /** * When used alongside a mark-supporting RecordingInputStream, remember * a position reachable by a future reset(). diff --git a/commons/src/main/java/org/archive/io/ReplayCharSequence.java b/commons/src/main/java/org/archive/io/ReplayCharSequence.java index f5f15176..99a04ec8 100644 --- a/commons/src/main/java/org/archive/io/ReplayCharSequence.java +++ b/commons/src/main/java/org/archive/io/ReplayCharSequence.java @@ -22,6 +22,9 @@ package org.archive.io; import java.io.Closeable; import java.io.IOException; import java.nio.charset.CharacterCodingException; +import java.nio.charset.Charset; + +import com.google.common.base.Charsets; /** @@ -37,8 +40,7 @@ public interface ReplayCharSequence extends CharSequence, Closeable { /** charset to use in replay when declared value * is absent/illegal/unavailable */ -// String FALLBACK_CHARSET_NAME = "UTF-8"; - String FALLBACK_CHARSET_NAME = "ISO8859_1"; + public Charset FALLBACK_CHARSET = Charsets.ISO_8859_1; // TODO: should this be UTF-8? /** * Call this method when done so implementation has chance to clean up diff --git a/commons/src/main/java/org/archive/io/ReplayInputStream.java b/commons/src/main/java/org/archive/io/ReplayInputStream.java index 04656d4b..d50d0067 100644 --- a/commons/src/main/java/org/archive/io/ReplayInputStream.java +++ b/commons/src/main/java/org/archive/io/ReplayInputStream.java @@ -21,8 +21,13 @@ package org.archive.io; import java.io.File; import java.io.IOException; +import java.io.InputStream; import java.io.OutputStream; +import org.apache.commons.io.IOUtils; +import org.archive.util.ArchiveUtils; +import org.archive.util.FileUtils; + /** * Replays the bytes recorded from a RecordingInputStream or @@ -34,6 +39,7 @@ import java.io.OutputStream; */ public class ReplayInputStream extends SeekInputStream { + private static final int DEFAULT_BUFFER_SIZE = 256*1024; // 256KiB private BufferedSeekInputStream diskStream; private byte[] buffer; private long position; @@ -89,9 +95,43 @@ public class ReplayInputStream extends SeekInputStream this.buffer = buffer; this.size = size; if (size > buffer.length) { - RandomAccessInputStream rais = new RandomAccessInputStream( - new File(backingFilename)); - diskStream = new BufferedSeekInputStream(rais, 4096); + setupDiskStream(new File(backingFilename)); + } + } + + protected void setupDiskStream(File backingFile) throws IOException { + RandomAccessInputStream rais = new RandomAccessInputStream(backingFile); + diskStream = new BufferedSeekInputStream(rais, 4096); + } + + File backingFile; + + /** + * Create a ReplayInputStream from the given source stream. Requires + * reading the entire stream (and possibly overflowing to a temporary + * file). Primary reason for doing so would be to have a repositionable + * version of the original stream's contents. + * @param fillStream + * @throws IOException + */ + public ReplayInputStream(InputStream fillStream) throws IOException { + this.buffer = new byte[DEFAULT_BUFFER_SIZE]; + long count = ArchiveUtils.readFully(fillStream, buffer); + if(fillStream.available()>0) { + this.backingFile = File.createTempFile("tid"+Thread.currentThread().getId(), "ris"); + count += FileUtils.readFullyToFile(fillStream, backingFile); + setupDiskStream(backingFile); + } + this.size = count; + } + + /** + * Close & destroy any internally-generated temporary files. + */ + public void destroy() { + IOUtils.closeQuietly(this); + if(backingFile!=null) { + FileUtils.deleteSoonerOrLater(backingFile); } } diff --git a/commons/src/main/java/org/archive/util/Recorder.java b/commons/src/main/java/org/archive/util/Recorder.java index 6247ba83..59ceeec2 100644 --- a/commons/src/main/java/org/archive/util/Recorder.java +++ b/commons/src/main/java/org/archive/util/Recorder.java @@ -22,15 +22,26 @@ import java.io.BufferedInputStream; import java.io.File; import java.io.IOException; import java.io.InputStream; +import java.io.InputStreamReader; import java.io.OutputStream; +import java.nio.charset.Charset; import java.util.logging.Level; import java.util.logging.Logger; +import java.util.zip.DeflaterInputStream; +import java.util.zip.GZIPInputStream; +import org.apache.commons.httpclient.ChunkedInputStream; +import org.apache.commons.io.FileUtils; +import org.apache.commons.io.IOUtils; +import org.apache.commons.lang.StringUtils; +import org.archive.io.GenericReplayCharSequence; import org.archive.io.RecordingInputStream; import org.archive.io.RecordingOutputStream; import org.archive.io.ReplayCharSequence; import org.archive.io.ReplayInputStream; +import com.google.common.base.Charsets; + /** * Pairs together a RecordingInputStream and RecordingOutputStream @@ -47,8 +58,8 @@ public class Recorder { protected static Logger logger = Logger.getLogger("org.archive.util.HttpRecorder"); - private static final int DEFAULT_OUTPUT_BUFFER_SIZE = 4096; - private static final int DEFAULT_INPUT_BUFFER_SIZE = 65536; + private static final int DEFAULT_OUTPUT_BUFFER_SIZE = 16384; + private static final int DEFAULT_INPUT_BUFFER_SIZE = 524288; private RecordingInputStream ris = null; private RecordingOutputStream ros = null; @@ -71,10 +82,16 @@ public class Recorder { private static final String RECORDING_INPUT_STREAM_SUFFIX = ".ris"; /** - * Response character encoding. + * recording-input (ris) content character encoding. */ private String characterEncoding = null; + + /** whether recording-input (ris) message-body is chunked */ + protected boolean inputIsChunked = false; + /** recording-input (ris) entity content-encoding (eg gzip, deflate), if any */ + protected String contentEncoding = null; + private ReplayCharSequence replayCharSequence; @@ -143,6 +160,12 @@ public class Recorder { public InputStream inputWrap(InputStream is) throws IOException { logger.fine(Thread.currentThread().getName() + " wrapping input"); + + // discard any state from previously-recorded input + this.characterEncoding = null; + this.inputIsChunked = false; + this.contentEncoding = null; + this.ris.open(is); return this.ris; } @@ -278,12 +301,43 @@ public class Recorder { } /** - * @return Returns the characterEncoding. + * @return Returns the characterEncoding of input recording. */ public String getCharacterEncoding() { return this.characterEncoding; } + + /** + * @param characterEncoding Character encoding of input recording. + */ + public void setInputIsChunked(boolean chunked) { + this.inputIsChunked = chunked; + } + + /** + * @param contentEncoding declared content-encoding of input recording. + */ + public void setContentEncoding(String contentEncoding) { + this.contentEncoding = contentEncoding; + } + /** + * @return Returns the characterEncoding. + */ + public String getContentEncoding() { + return this.contentEncoding; + } + + + /** + * @return + * @throws IOException + * @deprecated use getContentReplayCharSequence + */ + public ReplayCharSequence getReplayCharSequence() throws IOException { + return getContentReplayCharSequence(); + } + /** * @return A ReplayCharSequence. Caller may call * {@link ReplayCharSequence#close()} when finished. However, in @@ -293,15 +347,50 @@ public class Recorder { * @throws IOException * @see {@link #endReplays()} */ - public ReplayCharSequence getReplayCharSequence() throws IOException { + public ReplayCharSequence getContentReplayCharSequence() throws IOException { if (replayCharSequence == null || !replayCharSequence.isOpen()) { - replayCharSequence = getRecordedInput().getReplayCharSequence(this.characterEncoding); + replayCharSequence = getContentReplayCharSequence(this.characterEncoding); } return replayCharSequence; } + /** + * @param characterEncoding Encoding of recorded stream. + * @return A ReplayCharSequence Will return null if an IOException. Call + * close on returned RCS when done. + * @throws IOException + */ + public ReplayCharSequence getContentReplayCharSequence(String encoding) throws IOException { + Charset charset = Charsets.UTF_8; + if (encoding != null) { + try { + charset = Charset.forName(encoding); + } catch (IllegalArgumentException e) { + logger.log(Level.WARNING,"charset problem: "+encoding,e); + // TODO: better detection or default + charset = ReplayCharSequence.FALLBACK_CHARSET; + } + } + + logger.fine("using GenericReplayCharSequence"); + // raw data overflows to disk; use temp file + InputStream ris = getContentReplayInputStream(); + ReplayCharSequence rcs = new GenericReplayCharSequence( + ris, + this.ros.getBufferLength()/2, + this.backingFileBasename + RECORDING_OUTPUT_STREAM_SUFFIX, + charset); + ris.close(); + return rcs; + + } + + /** + * Get a raw replay of all recorded data (including, for example, HTTP + * protocol headers) + * * @return A replay input stream. * @throws IOException */ @@ -309,6 +398,103 @@ public class Recorder { return getRecordedInput().getReplayInputStream(); } + /** + * Get a raw replay of the 'message-body'. For the common case of + * HTTP, this is the raw, possibly chunked-transfer-encoded message + * contents not including the leading headers. + * + * @return A replay input stream. + * @throws IOException + */ + public ReplayInputStream getMessageBodyReplayInputStream() throws IOException { + return getRecordedInput().getMessageBodyReplayInputStream(); + } + + /** + * Get a raw replay of the 'entity'. For the common case of + * HTTP, this is the message-body after any (usually-unnecessary) + * transfer-decoding but before any content-encoding (eg gzip) decoding + * + * @return A replay input stream. + * @throws IOException + */ + public InputStream getEntityReplayInputStream() throws IOException { + if(inputIsChunked) { + return new ChunkedInputStream(getRecordedInput().getMessageBodyReplayInputStream()); + } else { + return getRecordedInput().getMessageBodyReplayInputStream(); + } + } + + /** + * Get a replay cued up for the 'content' (after all leading headers) + * + * TODO: handle chunking + * TODO: handle decompression, either here or in a parallel method + * + * @return A replay input stream. + * @throws IOException + */ + public InputStream getContentReplayInputStream() throws IOException { + InputStream entityStream = getEntityReplayInputStream(); + if(StringUtils.isEmpty(contentEncoding)) { + return entityStream; + } else if ("gzip".equalsIgnoreCase(contentEncoding) || "x-gzip".equalsIgnoreCase(contentEncoding)) { + try { + return new GZIPInputStream(entityStream); + } catch (IOException ioe) { + logger.log(Level.WARNING,"gzip problem; using raw entity instead",ioe); + IOUtils.closeQuietly(entityStream); // close partially-read stream + return getEntityReplayInputStream(); + } + } else if ("deflate".equalsIgnoreCase(contentEncoding)) { + return new DeflaterInputStream(entityStream); + } else if ("identity".equalsIgnoreCase(contentEncoding)) { + return entityStream; + } else { + logger.log(Level.WARNING,"Unknown content-encoding '"+contentEncoding+"' declared; using raw entity instead"); + return entityStream; + } + } + + /** + * Return a short prefix of the presumed-textual content as a String. + * + * @param size max length of String to return + * @return String prefix, or empty String (with logged exception) on any error + */ + public String getContentReplayPrefixString(int size) { + try { + InputStreamReader isr = (characterEncoding == null) + ? new InputStreamReader(getContentReplayInputStream(), Charsets.ISO_8859_1) + : new InputStreamReader(getContentReplayInputStream(), characterEncoding); + char[] chars = new char[size]; + int count = isr.read(chars); + isr.close(); + return new String(chars,0,count); + } catch (IOException e) { + logger.log(Level.SEVERE,"unable to get replay prefix string", e); + return ""; + } + } + + /** + * @param tempFile + * @throws IOException + */ + public void copyContentBodyTo(File tempFile) throws IOException { + InputStream inStream = null; + OutputStream outStream = null; + try { + inStream = getContentReplayInputStream(); + outStream = FileUtils.openOutputStream(tempFile); + IOUtils.copy(inStream, outStream); + } finally { + IOUtils.closeQuietly(inStream); + IOUtils.closeQuietly(outStream); + } + } + /** * Record the input stream for later playback by an extractor, etc. * This is convenience method used to setup an artificial HttpRecorder diff --git a/modules/src/main/java/org/archive/modules/fetcher/FetchHTTP.java b/modules/src/main/java/org/archive/modules/fetcher/FetchHTTP.java index 28a53bbd..9e769c29 100644 --- a/modules/src/main/java/org/archive/modules/fetcher/FetchHTTP.java +++ b/modules/src/main/java/org/archive/modules/fetcher/FetchHTTP.java @@ -195,7 +195,8 @@ public class FetchHTTP extends Processor implements Lifecycle { /** * Accept Headers to include in each request. Each must be the complete - * header, e.g., 'Accept-Language: en'. + * header, e.g., 'Accept-Language: en'. (Thus, this can also be used to + * other headers not beginning 'Accept-' as well.) */ { setAcceptHeaders(new LinkedList()); @@ -322,6 +323,34 @@ public class FetchHTTP extends Processor implements Lifecycle { */ private static final String MIDFETCH_ABORT_LOG = "midFetchAbort"; + /** + * Use HTTP/1.1. Note: even when offering an HTTP/1.1 request, + * Heritrix may not properly handle persistent/keep-alive connections, + * so the sendConnectionClose parameter should remain 'true'. + */ + { + setUseHTTP11(false); + } + public boolean getUseHTTP11() { + return (Boolean) kp.get("useHTTP11"); + } + public void setUseHTTP11(boolean useHTTP11) { + kp.put("useHTTP11",useHTTP11); + } + + /** + * Set headers to accept compressed responses. + */ + { + setAcceptCompression(false); + } + public boolean getAcceptCompression() { + return (Boolean) kp.get("acceptCompression"); + } + public void setAcceptCompression(boolean acceptCompression) { + kp.put("acceptCompression",acceptCompression); + } + /** * Send 'Connection: close' header with every request. */ @@ -622,6 +651,7 @@ public class FetchHTTP extends Processor implements Lifecycle { // Set the response charset into the HttpRecord if available. setCharacterEncoding(curi, rec, method); setSizes(curi, rec); + setOtherCodings(curi, rec, method); } if (digestContent) { @@ -765,6 +795,32 @@ public class FetchHTTP extends Processor implements Lifecycle { } rec.setCharacterEncoding(encoding); } + + /** + * Set the transfer, content encodings based on headers (if necessary). + * + * @param rec + * Recorder for this request. + * @param method + * Method used for the request. + */ + private void setOtherCodings(CrawlURI uri, final Recorder rec, + final HttpMethod method) { + Header transferCodingHeader = ((HttpMethodBase) method).getResponseHeader("Transfer-Encoding"); + if (transferCodingHeader !=null) { + String te = transferCodingHeader.getValue().trim(); + if(te.equalsIgnoreCase("chunked")) { + rec.setInputIsChunked(true); + } else { + logger.log(Level.WARNING,"Unknown transfer-encoding '"+te+"' for "+uri.getURI()); + } + } + Header contentEncodingHeader = ((HttpMethodBase) method).getResponseHeader("Content-Encoding"); + if (contentEncodingHeader!=null) { + String ce = contentEncodingHeader.getValue().trim(); + rec.setContentEncoding(ce); + } + } /** * Cleanup after a failed method execute. @@ -862,8 +918,9 @@ public class FetchHTTP extends Processor implements Lifecycle { ignoreCookies ? CookiePolicy.IGNORE_COOKIES : CookiePolicy.BROWSER_COMPATIBILITY); - // Use only HTTP/1.0 (to avoid receiving chunked responses) - method.getParams().setVersion(HttpVersion.HTTP_1_0); + method.getParams().setVersion(getUseHTTP11() + ? HttpVersion.HTTP_1_1 + : HttpVersion.HTTP_1_0); UserAgentProvider uap = getUserAgentProvider(); String from = uap.getFrom(); @@ -1398,6 +1455,11 @@ public class FetchHTTP extends Processor implements Lifecycle { private void setAcceptHeaders(CrawlURI curi, HttpMethod get) { + if(getAcceptCompression()) { + // we match the Firefox header exactly (ordering and whitespace) + // as a favor to caches + get.setRequestHeader("Accept-Encoding","gzip,deflate"); + } List acceptHeaders = getAcceptHeaders(); if (acceptHeaders.isEmpty()) { return;