diff --git a/commons/src/main/java/org/archive/io/Arc2Warc.java b/commons/src/main/java/org/archive/io/Arc2Warc.java index eedd848f..72ddf192 100644 --- a/commons/src/main/java/org/archive/io/Arc2Warc.java +++ b/commons/src/main/java/org/archive/io/Arc2Warc.java @@ -41,7 +41,10 @@ import org.archive.io.arc.ARCReader; import org.archive.io.arc.ARCReaderFactory; import org.archive.io.arc.ARCRecord; import org.archive.io.warc.WARCConstants; +import org.archive.io.warc.WARCConstants.WARCRecordType; +import org.archive.io.warc.WARCRecordInfo; import org.archive.io.warc.WARCWriter; +import org.archive.io.warc.WARCWriterPoolSettings; import org.archive.io.warc.WARCWriterPoolSettingsData; import org.archive.uid.RecordIDGenerator; import org.archive.uid.UUIDGenerator; @@ -101,7 +104,7 @@ public class Arc2Warc { getLength()); firstRecord.dump(baos); // Add ARC first record content as an ANVLRecord. - ANVLRecord ar = new ANVLRecord(1); + ANVLRecord ar = new ANVLRecord(); ar.addLabelValue("Filedesc", baos.toString()); List metadata = new ArrayList(1); metadata.add(ar.toString()); @@ -145,6 +148,11 @@ public class Arc2Warc { protected void write(final WARCWriter writer, final ARCRecord r) throws IOException { + WARCRecordInfo recordInfo = new WARCRecordInfo(); + recordInfo.setUrl(r.getHeader().getUrl()); + recordInfo.setContentStream(r); + recordInfo.setContentLength(r.getHeader().getLength()); + recordInfo.setEnforceLength(true); // convert ARC date to WARC-Date format String arcDateString = r.getHeader().getDate(); @@ -152,6 +160,7 @@ public class Arc2Warc { .withZone(DateTimeZone.UTC) .parseDateTime(arcDateString) .toString(ISODateTimeFormat.dateTimeNoMillis()); + recordInfo.setCreate14DigitDate(warcDateString); ANVLRecord ar = new ANVLRecord(); String ip = (String)r.getHeader() @@ -160,6 +169,7 @@ public class Arc2Warc { ar.addLabelValue(WARCConstants.NAMED_FIELD_IP_LABEL, ip); r.getMetaData(); } + recordInfo.setExtraHeaders(ar); // enable reconstruction of ARC from transformed WARC // TODO: deferred for further analysis (see HER-1750) @@ -167,18 +177,17 @@ public class Arc2Warc { // If contentBody > 0, assume http headers. Make the mimetype // be application/http. Otherwise, give it ARC mimetype. - String warcMimeTypeString; if (r.getHeader().getContentBegin() > 0) { - warcMimeTypeString = WARCConstants.HTTP_RESPONSE_MIMETYPE; - writer.writeResponseRecord(r.getHeader().getUrl(), warcDateString, - warcMimeTypeString, generator.getRecordID(), ar, r, - r.getHeader().getLength()); + recordInfo.setType(WARCRecordType.RESPONSE); + recordInfo.setMimetype(WARCConstants.HTTP_RESPONSE_MIMETYPE); + recordInfo.setRecordId(generator.getRecordID()); } else { - warcMimeTypeString = r.getHeader().getMimetype(); - writer.writeResourceRecord(r.getHeader().getUrl(), warcDateString, - warcMimeTypeString, ar, r, r.getHeader().getLength()); + recordInfo.setType(WARCRecordType.RESOURCE); + recordInfo.setMimetype(r.getHeader().getMimetype()); + recordInfo.setRecordId(((WARCWriterPoolSettings)writer.settings).getRecordIDGenerator().getRecordID()); } + writer.writeRecord(recordInfo); } /** diff --git a/commons/src/main/java/org/archive/io/warc/WARCConstants.java b/commons/src/main/java/org/archive/io/warc/WARCConstants.java index 8453238e..862f8a16 100644 --- a/commons/src/main/java/org/archive/io/warc/WARCConstants.java +++ b/commons/src/main/java/org/archive/io/warc/WARCConstants.java @@ -19,9 +19,6 @@ package org.archive.io.warc; -import java.util.Arrays; -import java.util.List; - import org.archive.io.ArchiveFileConstants; /** @@ -119,34 +116,29 @@ public interface WARCConstants extends ArchiveFileConstants { /** * WARC Record Types. */ - public static final String WARCINFO = "warcinfo"; - public static final String RESPONSE = "response"; - public static final String RESOURCE = "resource"; - public static final String REQUEST = "request"; - public static final String METADATA = "metadata"; - public static final String REVISIT = "revisit"; - public static final String CONVERSION = "conversion"; - public static final String CONTINUATION = "continuation"; + enum WARCRecordType { + WARCINFO("warcinfo"), + RESPONSE("response"), + RESOURCE("resource"), + REQUEST("request"), + METADATA("metadata"), + REVISIT("revisit"), + CONVERSION("conversion"), + CONTINUATION("continuation"); + + private String value; + private WARCRecordType(String value) { + this.value = value; + } + + @Override + public String toString() { + return value; + } + } public static final String TYPE = "type"; - // List of all WARC Record TYPES - public static final String [] TYPES = {WARCINFO, RESPONSE, RESOURCE, - REQUEST, METADATA, REVISIT, CONVERSION, CONTINUATION}; - - // Indices into TYPES array. - public static final int WARCINFO_INDEX = 0; - public static final int RESPONSE_INDEX = 1; - public static final int RESOURCE_INDEX = 2; - public static final int REQUEST_INDEX = 3; - public static final int METADATA_INDEX = 4; - public static final int REVISIT_INDEX = 5; - public static final int CONVERSION_INDEX = 6; - public static final int CONTINUATION_INDEX = 7; - - // TYPES as List. - public static final List TYPES_LIST = Arrays.asList(TYPES); - /** * WARC-ID */ @@ -201,6 +193,18 @@ public interface WARCConstants extends ArchiveFileConstants { public static final String HEADER_KEY_FILENAME = "WARC-Filename"; public static final String HEADER_KEY_ETAG = "WARC-Etag"; public static final String HEADER_KEY_LAST_MODIFIED = "WARC-Last-Modified"; + public static final String HEADER_KEY_REFERS_TO = "WARC-Refers-To"; + + /** + * These fields help a consumer of the warc to locate the warc record that + * {@value #HEADER_KEY_REFERS_TO} refers to. + * + * @see WARCWriterProcessor + */ + public static final String HEADER_KEY_REFERS_TO_TARGET_URI = "WARC-Refers-To-Target-URI"; + public static final String HEADER_KEY_REFERS_TO_DATE = "WARC-Refers-To-Date"; + public static final String HEADER_KEY_REFERS_TO_FILENAME = "WARC-Refers-To-Filename"; + public static final String HEADER_KEY_REFERS_TO_FILE_OFFSET = "WARC-Refers-To-File-Offset"; public static final String PROFILE_REVISIT_IDENTICAL_DIGEST = "http://netpreserve.org/warc/1.0/revisit/identical-payload-digest"; diff --git a/commons/src/main/java/org/archive/io/warc/WARCRecordInfo.java b/commons/src/main/java/org/archive/io/warc/WARCRecordInfo.java new file mode 100644 index 00000000..d34f00b7 --- /dev/null +++ b/commons/src/main/java/org/archive/io/warc/WARCRecordInfo.java @@ -0,0 +1,139 @@ +/* + * 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.warc; + +import java.io.InputStream; +import java.net.URI; + +import org.archive.io.warc.WARCConstants.WARCRecordType; +import org.archive.util.anvl.ANVLRecord; + +public class WARCRecordInfo { + + protected WARCRecordType type; + protected String url; + protected String create14DigitDate; + protected String mimetype; + protected URI recordId; + protected ANVLRecord extraHeaders; + protected InputStream contentStream; + protected long contentLength; + protected boolean enforceLength; + protected String warcFilename; + protected Long warcFileOffset; + + public void setType(WARCRecordType type) { + this.type = type; + } + + public void setUrl(String url) { + this.url = url; + } + + public String getCreate14DigitDate() { + return create14DigitDate; + } + + public void setCreate14DigitDate(String create14DigitDate) { + this.create14DigitDate = create14DigitDate; + } + + public String getMimetype() { + return mimetype; + } + + public void setMimetype(String mimetype) { + this.mimetype = mimetype; + } + + public URI getRecordId() { + return recordId; + } + + public void setRecordId(URI recordId) { + this.recordId = recordId; + } + + public ANVLRecord getExtraHeaders() { + return extraHeaders; + } + + public void setExtraHeaders(ANVLRecord extraHeaders) { + this.extraHeaders = extraHeaders; + } + + public InputStream getContentStream() { + return contentStream; + } + + public void setContentStream(InputStream contentStream) { + this.contentStream = contentStream; + } + + public long getContentLength() { + return contentLength; + } + + public void setContentLength(long contentLength) { + this.contentLength = contentLength; + } + + public boolean isEnforceLength() { + return enforceLength; + } + + public boolean getEnforceLength() { + return enforceLength; + } + + public void setEnforceLength(boolean enforceLength) { + this.enforceLength = enforceLength; + } + + public WARCRecordType getType() { + return type; + } + + public String getUrl() { + return url; + } + + public void addExtraHeader(String label, String value) { + if (extraHeaders == null) { + extraHeaders = new ANVLRecord(); + } + extraHeaders.addLabelValue(label, value); + } + + public void setWARCFilename(String warcFilenameWithoutOccupiedSuffix) { + this.warcFilename = warcFilenameWithoutOccupiedSuffix; + } + + public String getWARCFilename() { + return warcFilename; + } + + public void setWARCFileOffset(Long startPosition) { + this.warcFileOffset = startPosition; + } + + public Long getWARCFileOffset() { + return warcFileOffset; + } +} diff --git a/commons/src/main/java/org/archive/io/warc/WARCWriter.java b/commons/src/main/java/org/archive/io/warc/WARCWriter.java index 8cfccdd1..9c7b737e 100644 --- a/commons/src/main/java/org/archive/io/warc/WARCWriter.java +++ b/commons/src/main/java/org/archive/io/warc/WARCWriter.java @@ -23,11 +23,11 @@ import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.File; import java.io.IOException; -import java.io.InputStream; import java.io.OutputStream; import java.net.URI; import java.util.HashMap; import java.util.Iterator; +import java.util.LinkedList; import java.util.Map; import java.util.Map.Entry; import java.util.concurrent.ConcurrentMap; @@ -36,10 +36,11 @@ import java.util.concurrent.atomic.AtomicLong; import java.util.logging.Level; import java.util.logging.Logger; +import org.apache.commons.lang.StringUtils; +import org.archive.io.ArchiveFileConstants; import org.archive.io.UTF8Bytes; import org.archive.io.WriterPoolMember; import org.archive.util.ArchiveUtils; -import org.archive.util.anvl.ANVLRecord; import org.archive.util.anvl.Element; @@ -84,7 +85,10 @@ implements WARCConstants { * {@link #resetTmpStats()}, write some records, then add * {@link #getTmpStats()} into its long-term running totals. */ - private Map> tmpStats; + private Map> tmpStats; + + /** Temporarily accumulates info on written warc records for use externally. */ + private LinkedList tmpRecordLog = new LinkedList(); /** * Constructor. @@ -173,76 +177,58 @@ implements WARCConstants { return sb.toString(); } - protected String createRecordHeader(final String type, - final String url, final String create14DigitDate, - final String mimetype, final URI recordId, - final ANVLRecord xtraHeaders, final long contentLength) +// protected String createRecordHeader(final String type, +// final String url, final String create14DigitDate, +// final String mimetype, final URI recordId, +// final ANVLRecord xtraHeaders, final long contentLength) + protected String createRecordHeader(WARCRecordInfo metaRecord) throws IllegalArgumentException { final StringBuilder sb = new StringBuilder(2048/*A SWAG: TODO: Do analysis.*/); sb.append(WARC_ID).append(CRLF); - sb.append(HEADER_KEY_TYPE).append(COLON_SPACE).append(type). + sb.append(HEADER_KEY_TYPE).append(COLON_SPACE).append(metaRecord.getType()). append(CRLF); // Do not write a subject-uri if not one present. - if (url != null && url.length() > 0) { + if (!StringUtils.isEmpty(metaRecord.getUrl())) { sb.append(HEADER_KEY_URI).append(COLON_SPACE). - append(checkHeaderValue(url)).append(CRLF); + append(checkHeaderValue(metaRecord.getUrl())).append(CRLF); } sb.append(HEADER_KEY_DATE).append(COLON_SPACE). - append(create14DigitDate).append(CRLF); - if (xtraHeaders != null) { - for (final Iterator i = xtraHeaders.iterator(); i.hasNext();) { + append(metaRecord.getCreate14DigitDate()).append(CRLF); + if (metaRecord.getExtraHeaders() != null) { + for (final Iterator i = metaRecord.getExtraHeaders().iterator(); i.hasNext();) { sb.append(i.next()).append(CRLF); } } sb.append(HEADER_KEY_ID).append(COLON_SPACE).append('<'). - append(recordId.toString()).append('>').append(CRLF); - if (contentLength > 0) { + append(metaRecord.getRecordId().toString()).append('>').append(CRLF); + if (metaRecord.getContentLength() > 0) { sb.append(CONTENT_TYPE).append(COLON_SPACE).append( - checkHeaderLineMimetypeParameter(mimetype)).append(CRLF); + checkHeaderLineMimetypeParameter(metaRecord.getMimetype())).append(CRLF); } sb.append(CONTENT_LENGTH).append(COLON_SPACE). - append(Long.toString(contentLength)).append(CRLF); + append(Long.toString(metaRecord.getContentLength())).append(CRLF); return sb.toString(); } - /** - * @deprecated Use {@link #writeRecord(String,String,String,String,URI,ANVLRecord,InputStream,long,boolean)} instead - */ - protected void writeRecord(final String type, final String url, - final String create14DigitDate, final String mimetype, - final URI recordId, ANVLRecord xtraHeaders, - final InputStream contentStream, final long contentLength) + public void writeRecord(WARCRecordInfo recordInfo) throws IOException { - writeRecord(type, url, create14DigitDate, mimetype, recordId, xtraHeaders, contentStream, contentLength, true); - } - protected void writeRecord(final String type, final String url, - final String create14DigitDate, final String mimetype, - final URI recordId, ANVLRecord xtraHeaders, - final InputStream contentStream, final long contentLength, - boolean enforceLength) - throws IOException { - if (!TYPES_LIST.contains(type)) { - throw new IllegalArgumentException("Unknown record type: " + type); - } - if (contentLength == 0 && - (xtraHeaders == null || xtraHeaders.size() <= 0)) { + if (recordInfo.getContentLength() == 0 && + (recordInfo.getExtraHeaders() == null || recordInfo.getExtraHeaders().size() <= 0)) { throw new IllegalArgumentException("Cannot write record " + "of content-length zero and base headers only."); } String header; try { - header = createRecordHeader(type, url, - create14DigitDate, mimetype, recordId, xtraHeaders, - contentLength); + header = createRecordHeader(recordInfo); } catch (IllegalArgumentException e) { - logger.log(Level.SEVERE,"could not write record type: " + type - + "for URL: " + url, e); + logger.log(Level.SEVERE,"could not write record type: " + recordInfo.getType() + + "for URL: " + recordInfo.getUrl(), e); return; } @@ -259,12 +245,13 @@ implements WARCConstants { write(bytes); totalBytes += bytes.length; - - if (contentStream != null && contentLength > 0) { + if (recordInfo.getContentStream() != null && recordInfo.getContentLength() > 0) { // Write out the header/body separator. write(CRLF_BYTES); // TODO: should this be written even for zero-length? totalBytes += CRLF_BYTES.length; - contentBytes += copyFrom(contentStream, contentLength, enforceLength); + contentBytes += copyFrom(recordInfo.getContentStream(), + recordInfo.getContentLength(), + recordInfo.getEnforceLength()); totalBytes += contentBytes; } @@ -272,26 +259,37 @@ implements WARCConstants { write(CRLF_BYTES); write(CRLF_BYTES); totalBytes += 2 * CRLF_BYTES.length; + + tally(recordInfo.getType(), contentBytes, totalBytes, getPosition() - startPosition); + + recordInfo.setWARCFilename(getFilenameWithoutOccupiedSuffix()); + recordInfo.setWARCFileOffset(startPosition); + tmpRecordLog.add(recordInfo); } finally { postWriteRecordTasks(); } - - // TODO: should this be in the finally block? - tally(type, contentBytes, totalBytes, getPosition() - startPosition); + } + + public String getFilenameWithoutOccupiedSuffix() { + String name = getFile().getName(); + if (name.endsWith(ArchiveFileConstants.OCCUPIED_SUFFIX)) { + name = name.substring(0, name.length() - ArchiveFileConstants.OCCUPIED_SUFFIX.length()); + } + return name; } // if compression is enabled, sizeOnDisk means compressed bytes; if not, it // should be the same as totalBytes (right?) - protected void tally(String recordType, long contentBytes, long totalBytes, long sizeOnDisk) { + protected void tally(WARCRecordType warcRecordType, long contentBytes, long totalBytes, long sizeOnDisk) { if (tmpStats == null) { tmpStats = new HashMap>(); } // add to stats for this record type - Map substats = tmpStats.get(recordType); + Map substats = tmpStats.get(warcRecordType.toString()); if (substats == null) { substats = new HashMap(); - tmpStats.put(recordType, substats); + tmpStats.put(warcRecordType.toString(), substats); } subtally(substats, contentBytes, totalBytes, sizeOnDisk); @@ -349,16 +347,21 @@ implements WARCConstants { public URI writeWarcinfoRecord(String filename, final String description) throws IOException { + WARCRecordInfo recordInfo = new WARCRecordInfo(); + recordInfo.setType(WARCRecordType.WARCINFO); + recordInfo.setCreate14DigitDate(ArchiveUtils.getLog14Date()); + recordInfo.setMimetype("application/warc-fields"); + // Strip .open suffix if present. if (filename.endsWith(WriterPoolMember.OCCUPIED_SUFFIX)) { filename = filename.substring(0, filename.length() - WriterPoolMember.OCCUPIED_SUFFIX.length()); } - ANVLRecord record = new ANVLRecord(2); - record.addLabelValue(HEADER_KEY_FILENAME, filename); + recordInfo.addExtraHeader(HEADER_KEY_FILENAME, filename); if (description != null && description.length() > 0) { - record.addLabelValue(CONTENT_DESCRIPTION, description); + recordInfo.addExtraHeader(CONTENT_DESCRIPTION, description); } + // Add warcinfo body. byte [] warcinfoBody = null; if (settings.getMetadata() == null) { @@ -372,119 +375,19 @@ implements WARCConstants { } warcinfoBody = baos.toByteArray(); } - URI uri = writeWarcinfoRecord("application/warc-fields", record, - new ByteArrayInputStream(warcinfoBody), warcinfoBody.length); + recordInfo.setContentStream(new ByteArrayInputStream(warcinfoBody)); + recordInfo.setContentLength((long) warcinfoBody.length); + recordInfo.setEnforceLength(true); + + recordInfo.setRecordId(generateRecordId(TYPE, WARCRecordType.WARCINFO.toString())); + + writeRecord(recordInfo); + // TODO: If at start of file, and we're writing compressed, // write out our distinctive GZIP extensions. - return uri; + return recordInfo.getRecordId(); } - /** - * Write a warcinfo to current file. - * TODO: Write crawl metadata or pointers to crawl description. - * @param mimetype Mimetype of the fileMetadata block. - * @param namedFields Named fields. Pass null if none. - * @param fileMetadata Metadata about this WARC as RDF, ANVL, etc. - * @param fileMetadataLength Length of fileMetadata. - * @throws IOException - * @return Generated record-id made with - * data: scheme and - * the current filename. - */ - public URI writeWarcinfoRecord(final String mimetype, - final ANVLRecord namedFields, final InputStream fileMetadata, - final long fileMetadataLength) - throws IOException { - final URI recordid = generateRecordId(TYPE, WARCINFO); - writeWarcinfoRecord(ArchiveUtils.getLog14Date(), mimetype, recordid, - namedFields, fileMetadata, fileMetadataLength); - return recordid; - } - - /** - * Write a warcinfo to current file. - * The warcinfo type uses its recordId as its URL. - * @param recordId URI to use for this warcinfo. - * @param create14DigitDate Record creation date as 14 digit date. - * @param mimetype Mimetype of the fileMetadata. - * @param namedFields Named fields. - * @param fileMetadata Metadata about this WARC as RDF, ANVL, etc. - * @param fileMetadataLength Length of fileMetadata. - * @throws IOException - */ - public void writeWarcinfoRecord(final String create14DigitDate, - final String mimetype, final URI recordId, final ANVLRecord namedFields, - final InputStream fileMetadata, final long fileMetadataLength) - throws IOException { - writeRecord(WARCINFO, null, create14DigitDate, mimetype, - recordId, namedFields, fileMetadata, fileMetadataLength, true); - } - - public void writeRequestRecord(final String url, - final String create14DigitDate, final String mimetype, - final URI recordId, - final ANVLRecord namedFields, final InputStream request, - final long requestLength) - throws IOException { - writeRecord(REQUEST, url, create14DigitDate, - mimetype, recordId, namedFields, request, - requestLength, true); - } - - public void writeResourceRecord(final String url, - final String create14DigitDate, final String mimetype, - final ANVLRecord namedFields, final InputStream response, - final long responseLength) - throws IOException { - writeResourceRecord(url, create14DigitDate, mimetype, - ((WARCWriterPoolSettings)settings).getRecordIDGenerator().getRecordID(), - namedFields, response, responseLength); - } - - public void writeResourceRecord(final String url, - final String create14DigitDate, final String mimetype, - final URI recordId, - final ANVLRecord namedFields, final InputStream response, - final long responseLength) - throws IOException { - writeRecord(RESOURCE, url, create14DigitDate, - mimetype, recordId, namedFields, response, - responseLength, true); - } - - public void writeResponseRecord(final String url, - final String create14DigitDate, final String mimetype, - final URI recordId, - final ANVLRecord namedFields, final InputStream response, - final long responseLength) - throws IOException { - writeRecord(RESPONSE, url, create14DigitDate, - mimetype, recordId, namedFields, response, - responseLength, true); - } - - public void writeRevisitRecord(final String url, - final String create14DigitDate, final String mimetype, - final URI recordId, - final ANVLRecord namedFields, final InputStream response, - final long responseLength) - throws IOException { - writeRecord(REVISIT, url, create14DigitDate, - mimetype, recordId, namedFields, response, - responseLength, false); - } - - public void writeMetadataRecord(final String url, - final String create14DigitDate, final String mimetype, - final URI recordId, - final ANVLRecord namedFields, final InputStream metadata, - final long metadataLength) - throws IOException { - writeRecord(METADATA, url, create14DigitDate, - mimetype, recordId, namedFields, metadata, - metadataLength, true); - } - /** * @see WARCWriter#tmpStats for usage model */ @@ -522,4 +425,12 @@ implements WARCConstants { return 0l; } } + + public void resetTmpRecordLog() { + tmpRecordLog.clear(); + } + + public Iterable getTmpRecordLog() { + return tmpRecordLog; + } } diff --git a/commons/src/main/java/org/archive/util/anvl/ANVLRecord.java b/commons/src/main/java/org/archive/util/anvl/ANVLRecord.java index cfabb1dc..de2d3101 100644 --- a/commons/src/main/java/org/archive/util/anvl/ANVLRecord.java +++ b/commons/src/main/java/org/archive/util/anvl/ANVLRecord.java @@ -22,10 +22,10 @@ import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.InputStream; import java.io.UnsupportedEncodingException; -import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; import java.util.Iterator; +import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.logging.Level; @@ -42,8 +42,7 @@ import org.archive.io.UTF8Bytes; * Language (ANVL) * @author stack */ -public class ANVLRecord extends ArrayList implements UTF8Bytes { - private static final long serialVersionUID = -4610638888453052958L; +public class ANVLRecord extends LinkedList implements UTF8Bytes { private static final Logger logger = Logger.getLogger(ANVLRecord.class.getName()); @@ -73,8 +72,9 @@ public class ANVLRecord extends ArrayList implements UTF8Bytes { super(c); } + /** @deprecated */ public ANVLRecord(int initialCapacity) { - super(initialCapacity); + super(); } public boolean addLabel(final String l) { diff --git a/commons/src/main/java/org/archive/util/anvl/ANVLRecords.java b/commons/src/main/java/org/archive/util/anvl/ANVLRecords.java deleted file mode 100644 index 8ed5f43c..00000000 --- a/commons/src/main/java/org/archive/util/anvl/ANVLRecords.java +++ /dev/null @@ -1,61 +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.util.anvl; - -import java.io.UnsupportedEncodingException; -import java.util.ArrayList; -import java.util.Collection; -import java.util.Iterator; - -import org.archive.io.UTF8Bytes; - -/** - * List of {@link ANVLRecord}s. - * @author stack - * @version $Date$ $Version$ - */ -public class ANVLRecords extends ArrayList implements UTF8Bytes { - private static final long serialVersionUID = 5361551920550106113L; - - public ANVLRecords() { - super(); - } - - public ANVLRecords(int initialCapacity) { - super(initialCapacity); - } - - public ANVLRecords(Collection c) { - super(c); - } - - public byte[] getUTF8Bytes() throws UnsupportedEncodingException { - return toString().getBytes(UTF8); - } - - @Override - public String toString() { - StringBuilder sb = new StringBuilder(); - for (final Iterator i = iterator(); i.hasNext();) { - sb.append(i.next().toString()); - } - return super.toString(); - } -} \ No newline at end of file diff --git a/commons/src/test/java/org/archive/io/warc/WARCWriterTest.java b/commons/src/test/java/org/archive/io/warc/WARCWriterTest.java index fad081fe..300e948a 100644 --- a/commons/src/test/java/org/archive/io/warc/WARCWriterTest.java +++ b/commons/src/test/java/org/archive/io/warc/WARCWriterTest.java @@ -130,29 +130,51 @@ extends TmpDirTestCase implements WARCConstants { private void writeWarcinfoRecord(WARCWriter writer) throws IOException { + WARCRecordInfo recordInfo = new WARCRecordInfo(); + recordInfo.setType(WARCRecordType.WARCINFO); + recordInfo.setUrl(null); + recordInfo.setCreate14DigitDate(ArchiveUtils.getLog14Date()); + recordInfo.setMimetype(ANVLRecord.MIMETYPE); + recordInfo.setExtraHeaders(null); + recordInfo.setEnforceLength(true); + ANVLRecord meta = new ANVLRecord(); meta.addLabelValue("size", "1G"); meta.addLabelValue("operator", "igor"); byte [] bytes = meta.getUTF8Bytes(); - writer.writeWarcinfoRecord(ANVLRecord.MIMETYPE, null, - new ByteArrayInputStream(bytes), bytes.length); + recordInfo.setContentStream(new ByteArrayInputStream(bytes)); + recordInfo.setContentLength((long) bytes.length); + + final URI recordid = writer.generateRecordId(WARCWriter.TYPE, WARCRecordType.WARCINFO.toString()); + recordInfo.setRecordId(recordid); + + writer.writeRecord(recordInfo); } protected void writeBasicRecords(final WARCWriter writer) throws IOException { + WARCRecordInfo recordInfo = new WARCRecordInfo(); + recordInfo.setType(WARCRecordType.METADATA); + recordInfo.setUrl("http://www.archive.org/"); + recordInfo.setCreate14DigitDate(ArchiveUtils.get14DigitDate()); + recordInfo.setMimetype("no/type"); + recordInfo.setEnforceLength(true); + ANVLRecord headerFields = new ANVLRecord(); headerFields.addLabelValue("x", "y"); headerFields.addLabelValue("a", "b"); + recordInfo.setExtraHeaders(headerFields); + + URI rid = (new UUIDGenerator()).getQualifiedRecordID(TYPE, WARCRecordType.METADATA.toString()); + recordInfo.setRecordId(rid); - URI rid = (new UUIDGenerator()).getQualifiedRecordID(TYPE, METADATA); final String content = "Any old content."; for (int i = 0; i < 10; i++) { String body = i + ". " + content; byte [] bodyBytes = body.getBytes(UTF8Bytes.UTF8); - writer.writeRecord(METADATA, "http://www.archive.org/", - ArchiveUtils.get14DigitDate(), "no/type", - rid, headerFields, new ByteArrayInputStream(bodyBytes), - (long)bodyBytes.length, true); + recordInfo.setContentStream(new ByteArrayInputStream(bodyBytes)); + recordInfo.setContentLength((long)bodyBytes.length); + writer.writeRecord(recordInfo); } } @@ -186,23 +208,29 @@ extends TmpDirTestCase implements WARCConstants { */ protected int writeRandomHTTPRecord(WARCWriter w, int index) throws IOException { - ByteArrayOutputStream baos = new ByteArrayOutputStream(); + WARCRecordInfo recordInfo = new WARCRecordInfo(); + recordInfo.setType(WARCRecordType.RESOURCE); + recordInfo.setCreate14DigitDate(ArchiveUtils.get14DigitDate()); + recordInfo.setMimetype("text/html; charset=UTF-8"); + recordInfo.setRecordId(w.generateRecordId(null)); + recordInfo.setEnforceLength(true); + String indexStr = Integer.toString(index); + recordInfo.setUrl("http://www.one.net/id=" + indexStr); + byte[] record = (getContent(indexStr)).getBytes(); - int recordLength = record.length; + recordInfo.setContentLength((long) record.length); + + ByteArrayOutputStream baos = new ByteArrayOutputStream(); baos.write(record); + recordInfo.setContentStream(new ByteArrayInputStream(baos.toByteArray())); + // Add named fields for ip, checksum, and relate the metadata // and request to the resource field. - ANVLRecord r = new ANVLRecord(1); - r.addLabelValue(NAMED_FIELD_IP_LABEL, "127.0.0.1"); - w.writeResourceRecord( - "http://www.one.net/id=" + indexStr, - ArchiveUtils.get14DigitDate(), - "text/html; charset=UTF-8", - r, - new ByteArrayInputStream(baos.toByteArray()), - recordLength); - return recordLength; + recordInfo.addExtraHeader(NAMED_FIELD_IP_LABEL, "127.0.0.1"); + + w.writeRecord(recordInfo); + return record.length; } /** @@ -352,12 +380,18 @@ extends TmpDirTestCase implements WARCConstants { protected static void writeRecord(WARCWriter w, String url, String mimetype, int len, ByteArrayOutputStream baos) throws IOException { - w.writeResourceRecord(url, - ArchiveUtils.get14DigitDate(), - mimetype, - null, - new ByteArrayInputStream(baos.toByteArray()), - len); + WARCRecordInfo recordInfo = new WARCRecordInfo(); + recordInfo.setType(WARCRecordType.RESOURCE); + recordInfo.setUrl(url); + recordInfo.setCreate14DigitDate(ArchiveUtils.get14DigitDate()); + recordInfo.setMimetype(mimetype); + recordInfo.setRecordId(w.generateRecordId(null)); + recordInfo.setExtraHeaders(null); + recordInfo.setContentStream(new ByteArrayInputStream(baos.toByteArray())); + recordInfo.setContentLength((long) len); + recordInfo.setEnforceLength(true); + + w.writeRecord(recordInfo); } protected int iterateRecords(WARCReader r) diff --git a/modules/src/main/java/org/archive/modules/CrawlURI.java b/modules/src/main/java/org/archive/modules/CrawlURI.java index e2d392f8..ed8a7a79 100644 --- a/modules/src/main/java/org/archive/modules/CrawlURI.java +++ b/modules/src/main/java/org/archive/modules/CrawlURI.java @@ -57,6 +57,7 @@ import static org.archive.modules.fetcher.FetchStatusCodes.S_TOO_MANY_RETRIES; import static org.archive.modules.fetcher.FetchStatusCodes.S_UNATTEMPTED; import static org.archive.modules.fetcher.FetchStatusCodes.S_UNFETCHABLE_URI; import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_FETCH_HISTORY; +import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_CONTENT_DIGEST_HISTORY; import java.io.IOException; import java.io.ObjectInputStream; @@ -1895,4 +1896,19 @@ implements Reporter, Serializable, OverlayContext { return (HashMap[]) getData().get(A_FETCH_HISTORY); } + public HashMap getContentDigestHistory() { + @SuppressWarnings("unchecked") + HashMap contentDigestHistory = (HashMap) getData().get(A_CONTENT_DIGEST_HISTORY); + + if (contentDigestHistory == null) { + contentDigestHistory = new HashMap(); + getData().put(A_CONTENT_DIGEST_HISTORY, contentDigestHistory); + } + + return contentDigestHistory; + } + + public boolean hasContentDigestHistory() { + return getData().get(A_CONTENT_DIGEST_HISTORY) != null; + } } diff --git a/modules/src/main/java/org/archive/modules/recrawl/AbstractContentDigestHistory.java b/modules/src/main/java/org/archive/modules/recrawl/AbstractContentDigestHistory.java new file mode 100644 index 00000000..e0f39934 --- /dev/null +++ b/modules/src/main/java/org/archive/modules/recrawl/AbstractContentDigestHistory.java @@ -0,0 +1,59 @@ +/* + * 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.recrawl; + +import org.archive.modules.CrawlURI; + +/** + * Represents a store of information, presumably persistent, keyed by content + * digest. + * + * @contributor nlevitt + */ +public abstract class AbstractContentDigestHistory { + /** + * Looks up the history by key {@code persistKeyFor(curi)} and loads it into + * {@code curi.getContentDigestHistory()}. + * + * @param curi + */ + public abstract void load(CrawlURI curi); + + /** + * Stores {@code curi.getContentDigestHistory()} for the key + * {@code persistKeyFor(curi)}. + * + * @param curi + */ + public abstract void store(CrawlURI curi); + + /** + * + * @param curi + * @return {@code curi.getContentDigestSchemeString()} + * @throws IllegalStateException if {@code curi.getContentDigestSchemeString()} is null + */ + protected String persistKeyFor(CrawlURI curi) { + String key = curi.getContentDigestSchemeString(); + if (key == null) { + throw new IllegalStateException("cannot load content digest history, CrawlURI does not have content digest value for " + curi); + } + return key; + } +} diff --git a/modules/src/main/java/org/archive/modules/recrawl/BdbContentDigestHistory.java b/modules/src/main/java/org/archive/modules/recrawl/BdbContentDigestHistory.java new file mode 100644 index 00000000..1d09e43e --- /dev/null +++ b/modules/src/main/java/org/archive/modules/recrawl/BdbContentDigestHistory.java @@ -0,0 +1,146 @@ +/* + * 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.recrawl; + +import java.util.HashMap; +import java.util.Map; +import java.util.logging.Level; +import java.util.logging.Logger; + +import org.archive.bdb.BdbModule; +import org.archive.modules.CrawlURI; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.Lifecycle; + +import com.sleepycat.bind.serial.SerialBinding; +import com.sleepycat.bind.serial.StoredClassCatalog; +import com.sleepycat.bind.tuple.StringBinding; +import com.sleepycat.collections.StoredSortedMap; +import com.sleepycat.je.Database; +import com.sleepycat.je.DatabaseException; + +/** + * Bdb content digest history store. Must be a toplevel bean in + * crawler-beans.cxml in order to receive {@link Lifecycle} events. + * + * @see AbstractContentDigestHistory + * @contributor nlevitt + */ +public class BdbContentDigestHistory extends AbstractContentDigestHistory implements Lifecycle { + + private static final Logger logger = + Logger.getLogger(BdbContentDigestHistory.class.getName()); + + protected BdbModule bdb; + @Autowired + public void setBdbModule(BdbModule bdb) { + this.bdb = bdb; + } + + protected String historyDbName = "contentDigestHistory"; + public String getHistoryDbName() { + return this.historyDbName; + } + public void setHistoryDbName(String name) { + this.historyDbName = name; + } + + @SuppressWarnings("rawtypes") + protected StoredSortedMap store; + protected Database historyDb; + + @Override + @SuppressWarnings("rawtypes") + public void start() { + if (isRunning()) { + return; + } + StoredSortedMap historyMap; + try { + StoredClassCatalog classCatalog = bdb.getClassCatalog(); + historyDb = bdb.openDatabase(getHistoryDbName(), historyDbConfig(), true); + historyMap = new StoredSortedMap( + historyDb, + new StringBinding(), + new SerialBinding(classCatalog, Map.class), + true); + } catch (DatabaseException e) { + throw new RuntimeException(e); + } + store = historyMap; + } + + @Override + public boolean isRunning() { + return historyDb != null; + } + + @Override + public void stop() { + if (!isRunning()) { + return; + } + // leave other cleanup to BdbModule + historyDb = null; + } + + protected transient BdbModule.BdbConfig historyDbConfig; + protected BdbModule.BdbConfig historyDbConfig() { + if (historyDbConfig == null) { + historyDbConfig = new BdbModule.BdbConfig(); + historyDbConfig.setTransactional(false); + historyDbConfig.setAllowCreate(true); + historyDbConfig.setDeferredWrite(true); + } + + return historyDbConfig; + } + + public void load(CrawlURI curi) { + // make this call in all cases so that the value is initialized and + // WARCWriterProcessor knows it should put the info in there + HashMap contentDigestHistory = curi.getContentDigestHistory(); + + @SuppressWarnings("unchecked") + Map loadedHistory = store.get(persistKeyFor(curi)); + if (loadedHistory != null) { + if (logger.isLoggable(Level.FINER)) { + logger.finer("loaded history by digest " + persistKeyFor(curi) + + " for uri " + curi + " - " + loadedHistory); + } + contentDigestHistory.putAll(loadedHistory); + } + } + + public void store(CrawlURI curi) { + if (!curi.hasContentDigestHistory() + || curi.getContentDigestHistory().isEmpty()) { + logger.warning("not saving empty content digest history (do you " + + " have a ContentDigestHistoryLoader in your disposition" + + " chain?) - " + curi); + return; + } + if (logger.isLoggable(Level.FINER)) { + logger.finer("storing history by digest " + persistKeyFor(curi) + + " for uri " + curi + " - " + + curi.getContentDigestHistory()); + } + store.put(persistKeyFor(curi), curi.getContentDigestHistory()); + } +} \ No newline at end of file diff --git a/modules/src/main/java/org/archive/modules/recrawl/ContentDigestHistoryLoader.java b/modules/src/main/java/org/archive/modules/recrawl/ContentDigestHistoryLoader.java new file mode 100644 index 00000000..2f0f13cf --- /dev/null +++ b/modules/src/main/java/org/archive/modules/recrawl/ContentDigestHistoryLoader.java @@ -0,0 +1,43 @@ +/* + * 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.recrawl; + +import org.archive.modules.CrawlURI; +import org.archive.modules.Processor; +import org.springframework.beans.factory.annotation.Autowired; + +public class ContentDigestHistoryLoader extends Processor { + + protected AbstractContentDigestHistory contentDigestHistory; + @Autowired + public void setContentDigestHistory( + AbstractContentDigestHistory contentDigestHistory) { + this.contentDigestHistory = contentDigestHistory; + } + + @Override + protected boolean shouldProcess(CrawlURI uri) { + return uri.getContentDigest() != null; + } + + @Override + protected void innerProcess(CrawlURI curi) throws InterruptedException { + contentDigestHistory.load(curi); + } +} \ No newline at end of file diff --git a/modules/src/main/java/org/archive/modules/recrawl/ContentDigestHistoryStorer.java b/modules/src/main/java/org/archive/modules/recrawl/ContentDigestHistoryStorer.java new file mode 100644 index 00000000..f0c779ed --- /dev/null +++ b/modules/src/main/java/org/archive/modules/recrawl/ContentDigestHistoryStorer.java @@ -0,0 +1,43 @@ +/* + * 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.recrawl; + +import org.archive.modules.CrawlURI; +import org.archive.modules.Processor; +import org.springframework.beans.factory.annotation.Autowired; + +public class ContentDigestHistoryStorer extends Processor { + + protected AbstractContentDigestHistory contentDigestHistory; + @Autowired + public void setContentDigestHistory( + AbstractContentDigestHistory contentDigestHistory) { + this.contentDigestHistory = contentDigestHistory; + } + + @Override + protected boolean shouldProcess(CrawlURI uri) { + return uri.getContentDigest() != null; + } + + @Override + protected void innerProcess(CrawlURI curi) throws InterruptedException { + contentDigestHistory.store(curi); + } +} \ No newline at end of file diff --git a/modules/src/main/java/org/archive/modules/recrawl/RecrawlAttributeConstants.java b/modules/src/main/java/org/archive/modules/recrawl/RecrawlAttributeConstants.java index f95e4b2e..e7c41dce 100644 --- a/modules/src/main/java/org/archive/modules/recrawl/RecrawlAttributeConstants.java +++ b/modules/src/main/java/org/archive/modules/recrawl/RecrawlAttributeConstants.java @@ -32,14 +32,30 @@ public interface RecrawlAttributeConstants { public static final String A_FETCH_HISTORY = "fetch-history"; /** content digest */ public static final String A_CONTENT_DIGEST = "content-digest"; - /** header name (and AList key) for last-modified timestamp */ + /** header name (and AList key) for last-modified timestamp */ public static final String A_LAST_MODIFIED_HEADER = "last-modified"; - /** header name (and AList key) for ETag */ + /** header name (and AList key) for ETag */ public static final String A_ETAG_HEADER = "etag"; /** key for status (when in history) */ public static final String A_STATUS = "status"; /** reference length (content length or virtual length */ public static final String A_REFERENCE_LENGTH = "reference-length"; + + // constants for uri-agnostic content digest based dedupe + /** content digest history map */ + public static final String A_CONTENT_DIGEST_HISTORY = "content-digest-history"; + /** url that the content payload was written for */ + public static final String A_ORIGINAL_URL = "original-url"; + /** warc record id of warc record with the content payload */ + public static final String A_WARC_RECORD_ID = "warc-record-id"; + /** warc filename containing the content payload */ + public static final String A_WARC_FILENAME = "warc-filename"; + /** offset into warc file of warc record with content payload */ + public static final String A_WARC_FILE_OFFSET = "warc-file-offset"; + /** date content payload was written */ + public static final String A_ORIGINAL_DATE = "content-written-date"; + /** number of times we've seen this content digest (1 original + n duplicates) */ + public static final String A_CONTENT_DIGEST_COUNT = "content-digest-count"; /** * Writer processors of all types are encouraged to put a 'writeTag' diff --git a/modules/src/main/java/org/archive/modules/writer/WARCWriterProcessor.java b/modules/src/main/java/org/archive/modules/writer/WARCWriterProcessor.java index 293d560c..ec9fb4ca 100644 --- a/modules/src/main/java/org/archive/modules/writer/WARCWriterProcessor.java +++ b/modules/src/main/java/org/archive/modules/writer/WARCWriterProcessor.java @@ -26,16 +26,19 @@ import static org.archive.io.warc.WARCConstants.HEADER_KEY_IP; import static org.archive.io.warc.WARCConstants.HEADER_KEY_LAST_MODIFIED; import static org.archive.io.warc.WARCConstants.HEADER_KEY_PAYLOAD_DIGEST; import static org.archive.io.warc.WARCConstants.HEADER_KEY_PROFILE; +import static org.archive.io.warc.WARCConstants.HEADER_KEY_REFERS_TO; +import static org.archive.io.warc.WARCConstants.HEADER_KEY_REFERS_TO_DATE; +import static org.archive.io.warc.WARCConstants.HEADER_KEY_REFERS_TO_FILENAME; +import static org.archive.io.warc.WARCConstants.HEADER_KEY_REFERS_TO_FILE_OFFSET; +import static org.archive.io.warc.WARCConstants.HEADER_KEY_REFERS_TO_TARGET_URI; import static org.archive.io.warc.WARCConstants.HEADER_KEY_TRUNCATED; import static org.archive.io.warc.WARCConstants.HTTP_REQUEST_MIMETYPE; import static org.archive.io.warc.WARCConstants.HTTP_RESPONSE_MIMETYPE; -import static org.archive.io.warc.WARCConstants.METADATA; import static org.archive.io.warc.WARCConstants.NAMED_FIELD_TRUNCATED_VALUE_HEAD; import static org.archive.io.warc.WARCConstants.NAMED_FIELD_TRUNCATED_VALUE_LENGTH; import static org.archive.io.warc.WARCConstants.NAMED_FIELD_TRUNCATED_VALUE_TIME; import static org.archive.io.warc.WARCConstants.PROFILE_REVISIT_IDENTICAL_DIGEST; import static org.archive.io.warc.WARCConstants.PROFILE_REVISIT_NOT_MODIFIED; -import static org.archive.io.warc.WARCConstants.REQUEST; import static org.archive.io.warc.WARCConstants.TYPE; import static org.archive.modules.CoreAttributeConstants.A_DNS_SERVER_IP_LABEL; import static org.archive.modules.CoreAttributeConstants.A_FTP_CONTROL_CONVERSATION; @@ -44,8 +47,14 @@ import static org.archive.modules.CoreAttributeConstants.A_SOURCE_TAG; import static org.archive.modules.CoreAttributeConstants.HEADER_TRUNC; import static org.archive.modules.CoreAttributeConstants.LENGTH_TRUNC; import static org.archive.modules.CoreAttributeConstants.TIMER_TRUNC; +import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_CONTENT_DIGEST_COUNT; import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_ETAG_HEADER; import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_LAST_MODIFIED_HEADER; +import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_ORIGINAL_DATE; +import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_ORIGINAL_URL; +import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_WARC_FILENAME; +import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_WARC_FILE_OFFSET; +import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_WARC_RECORD_ID; import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_WRITE_TAG; import java.io.ByteArrayInputStream; @@ -69,8 +78,9 @@ import java.util.logging.Logger; import org.apache.commons.httpclient.HttpStatus; import org.apache.commons.io.IOUtils; import org.apache.commons.lang.StringUtils; -import org.archive.io.ArchiveFileConstants; import org.archive.io.ReplayInputStream; +import org.archive.io.warc.WARCConstants.WARCRecordType; +import org.archive.io.warc.WARCRecordInfo; import org.archive.io.warc.WARCWriter; import org.archive.io.warc.WARCWriterPool; import org.archive.io.warc.WARCWriterPoolSettings; @@ -143,6 +153,9 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit /** * Whether to write 'revisit' type records when a URI's history indicates * the previous fetch had an identical content digest. Default is true. + * + * Decision applies to either URI-based fetch history or URI-agnostic + * content digest-based history. */ { setWriteRevisitForIdenticalDigests(true); @@ -240,6 +253,8 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit // They'll be added to totals below, in finally block, after records // have been written. writer.resetTmpStats(); + writer.resetTmpRecordLog(); + // Write a request, response, and metadata all in the one // 'transaction'. final URI baseid = getRecordID(); @@ -266,33 +281,60 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit throw e; } finally { if (writer != null) { - if (WARCWriter.getStat(writer.getTmpStats(), WARCWriter.TOTALS, WARCWriter.NUM_RECORDS) > 0l) { - addStats(writer.getTmpStats()); - urlsWritten.incrementAndGet(); - } - if (logger.isLoggable(Level.FINE)) { - logger.fine("wrote " - + WARCWriter.getStat(writer.getTmpStats(), WARCWriter.TOTALS, WARCWriter.SIZE_ON_DISK) - + " bytes to " + writer.getFile().getName() + " for " + curi); - } - setTotalBytesWritten(getTotalBytesWritten() + - (writer.getPosition() - position)); + updateMetadataAfterWrite(curi, writer, position); getPool().returnFile(writer); - - String filename = writer.getFile().getName(); - if (filename.endsWith(ArchiveFileConstants.OCCUPIED_SUFFIX)) { - filename = filename.substring(0, filename.length() - ArchiveFileConstants.OCCUPIED_SUFFIX.length()); - } - curi.addExtraInfo("warcFilename", filename); - - Map[] history = curi.getFetchHistory(); - if (history != null && history[0] != null) { - history[0].put(A_WRITE_TAG, filename); - } } } return checkBytesWritten(); } + + protected void updateMetadataAfterWrite(final CrawlURI curi, + WARCWriter writer, long startPosition) { + if (WARCWriter.getStat(writer.getTmpStats(), WARCWriter.TOTALS, WARCWriter.NUM_RECORDS) > 0l) { + addStats(writer.getTmpStats()); + urlsWritten.incrementAndGet(); + } + if (logger.isLoggable(Level.FINE)) { + logger.fine("wrote " + + WARCWriter.getStat(writer.getTmpStats(), WARCWriter.TOTALS, WARCWriter.SIZE_ON_DISK) + + " bytes to " + writer.getFile().getName() + " for " + curi); + } + setTotalBytesWritten(getTotalBytesWritten() + (writer.getPosition() - startPosition)); + + curi.addExtraInfo("warcFilename", writer.getFilenameWithoutOccupiedSuffix()); + // curi.addExtraInfo("warcOffset", startPosition); + + // history for uri-based dedupe + Map[] history = curi.getFetchHistory(); + if (history != null && history[0] != null) { + history[0].put(A_WRITE_TAG, writer.getFilenameWithoutOccupiedSuffix()); + } + + // history for uri-agnostic, content digest based dedupe + if (curi.getContentDigest() != null && curi.hasContentDigestHistory()) { + for (WARCRecordInfo warcRecord: writer.getTmpRecordLog()) { + if ((warcRecord.getType() == WARCRecordType.RESPONSE + || warcRecord.getType() == WARCRecordType.RESOURCE) + && warcRecord.getContentStream() != null + && warcRecord.getContentLength() > 0) { + curi.getContentDigestHistory().put(A_ORIGINAL_URL, warcRecord.getUrl()); + curi.getContentDigestHistory().put(A_WARC_RECORD_ID, warcRecord.getRecordId()); + curi.getContentDigestHistory().put(A_WARC_FILENAME, warcRecord.getWARCFilename()); + curi.getContentDigestHistory().put(A_WARC_FILE_OFFSET, warcRecord.getWARCFileOffset()); + curi.getContentDigestHistory().put(A_ORIGINAL_DATE, warcRecord.getCreate14DigitDate()); + curi.getContentDigestHistory().put(A_CONTENT_DIGEST_COUNT, 1); + } else if (warcRecord.getType() == WARCRecordType.REVISIT + && curi.getAnnotations().contains("warcRevisit:uriAgnosticDigest")) { + Integer oldCount = (Integer) curi.getContentDigestHistory().get(A_CONTENT_DIGEST_COUNT); + if (oldCount == null) { + // shouldn't happen, log a warning? + oldCount = 1; + } + curi.getContentDigestHistory().put(A_CONTENT_DIGEST_COUNT, oldCount + 1); + } + } + } + } protected void addStats(Map> substats) { for (String key: substats.keySet()) { @@ -314,35 +356,71 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit } } - private void writeDnsRecords(final CrawlURI curi, WARCWriter w, + protected void writeDnsRecords(final CrawlURI curi, WARCWriter w, final URI baseid, final String timestamp) throws IOException { - ANVLRecord headers = null; + WARCRecordInfo recordInfo = new WARCRecordInfo(); + recordInfo.setType(WARCRecordType.RESPONSE); + recordInfo.setUrl(curi.toString()); + recordInfo.setCreate14DigitDate(timestamp); + recordInfo.setMimetype(curi.getContentType()); + recordInfo.setRecordId(baseid); + + recordInfo.setContentLength(curi.getRecorder().getRecordedInput().getSize()); + recordInfo.setEnforceLength(true); + String ip = (String)curi.getData().get(A_DNS_SERVER_IP_LABEL); if (ip != null && ip.length() > 0) { - headers = new ANVLRecord(1); - headers.addLabelValue(HEADER_KEY_IP, ip); + recordInfo.addExtraHeader(HEADER_KEY_IP, ip); } - writeResponse(w, timestamp, curi.getContentType(), baseid, - curi, headers); + + ReplayInputStream ris = + curi.getRecorder().getRecordedInput().getReplayInputStream(); + recordInfo.setContentStream(ris); + + try { + w.writeRecord(recordInfo); + } finally { + IOUtils.closeQuietly(ris); + } + + recordInfo.getRecordId(); } - private void writeWhoisRecords(WARCWriter w, CrawlURI curi, URI baseid, + protected void writeWhoisRecords(WARCWriter w, CrawlURI curi, URI baseid, String timestamp) throws IOException { - ANVLRecord headers = new ANVLRecord(1); + WARCRecordInfo recordInfo = new WARCRecordInfo(); + recordInfo.setType(WARCRecordType.RESPONSE); + recordInfo.setUrl(curi.toString()); + recordInfo.setCreate14DigitDate(timestamp); + recordInfo.setMimetype(curi.getContentType()); + recordInfo.setRecordId(baseid); + recordInfo.setContentLength(curi.getRecorder().getRecordedInput().getSize()); + recordInfo.setEnforceLength(true); + Object whoisServerIP = curi.getData().get(CoreAttributeConstants.A_WHOIS_SERVER_IP); if (whoisServerIP != null) { - headers.addLabelValue(HEADER_KEY_IP, whoisServerIP.toString()); + recordInfo.addExtraHeader(HEADER_KEY_IP, whoisServerIP.toString()); } - writeResponse(w, timestamp, curi.getContentType(), baseid, curi, headers); + + ReplayInputStream ris = + curi.getRecorder().getRecordedInput().getReplayInputStream(); + recordInfo.setContentStream(ris); + + try { + w.writeRecord(recordInfo); + } finally { + IOUtils.closeQuietly(ris); + } + recordInfo.getRecordId(); } - private void writeHttpRecords(final CrawlURI curi, WARCWriter w, + protected void writeHttpRecords(final CrawlURI curi, WARCWriter w, final URI baseid, final String timestamp) throws IOException { // Add named fields for ip, checksum, and relate the metadata // and request to the resource field. // TODO: Use other than ANVL (or rename ANVL as NameValue or // use RFC822 (commons-httpclient?). - ANVLRecord headers = new ANVLRecord(5); + ANVLRecord headers = new ANVLRecord(); if (curi.getContentDigest() != null) { headers.addLabelValue(HEADER_KEY_PAYLOAD_DIGEST, curi.getContentDigestSchemeString()); @@ -350,7 +428,12 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit headers.addLabelValue(HEADER_KEY_IP, getHostAddress(curi)); URI rid; - if (IdenticalDigestDecideRule.hasIdenticalDigest(curi) && + if (getWriteRevisitForIdenticalDigests() + && curi.hasContentDigestHistory() + && curi.getContentDigestHistory().get(A_ORIGINAL_URL) != null) { + rid = writeRevisitUriAgnosticDigest(w, timestamp, + HTTP_RESPONSE_MIMETYPE, baseid, curi, headers); + } else if (IdenticalDigestDecideRule.hasIdenticalDigest(curi) && getWriteRevisitForIdenticalDigests()) { rid = writeRevisitDigest(w, timestamp, HTTP_RESPONSE_MIMETYPE, baseid, curi, headers); @@ -377,7 +460,7 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit baseid, curi, headers); } - headers = new ANVLRecord(1); + headers = new ANVLRecord(); headers.addLabelValue(HEADER_KEY_CONCURRENT_TO, '<' + rid.toString() + '>'); @@ -390,9 +473,9 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit } } - private void writeFtpRecords(WARCWriter w, final CrawlURI curi, final URI baseid, + protected void writeFtpRecords(WARCWriter w, final CrawlURI curi, final URI baseid, final String timestamp) throws IOException { - ANVLRecord headers = new ANVLRecord(3); + ANVLRecord headers = new ANVLRecord(); headers.addLabelValue(HEADER_KEY_IP, getHostAddress(curi)); String controlConversation = curi.getData().get(A_FTP_CONTROL_CONVERSATION).toString(); URI rid = writeFtpControlConversation(w, timestamp, baseid, curi, headers, controlConversation); @@ -408,7 +491,7 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit rid = writeRevisitDigest(w, timestamp, null, baseid, curi, headers, 0); } else { - headers = new ANVLRecord(3); + headers = new ANVLRecord(); // Check for truncated annotation String value = null; Collection anno = curi.getAnnotations(); @@ -433,7 +516,7 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit } } if (getWriteMetadata()) { - headers = new ANVLRecord(1); + headers = new ANVLRecord(); headers.addLabelValue(HEADER_KEY_CONCURRENT_TO, '<' + rid.toString() + '>'); writeMetadata(w, timestamp, baseid, curi, headers); } @@ -442,12 +525,25 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit protected URI writeFtpControlConversation(WARCWriter w, String timestamp, URI baseid, CrawlURI curi, ANVLRecord headers, String controlConversation) throws IOException { - final URI uid = qualifyRecordID(baseid, TYPE, METADATA); + + WARCRecordInfo recordInfo = new WARCRecordInfo(); + recordInfo.setCreate14DigitDate(timestamp); + recordInfo.setUrl(curi.toString()); + recordInfo.setMimetype(FTP_CONTROL_CONVERSATION_MIMETYPE); + recordInfo.setExtraHeaders(headers); + recordInfo.setEnforceLength(true); + recordInfo.setType(WARCRecordType.METADATA); + + recordInfo.setRecordId(qualifyRecordID(baseid, TYPE, WARCRecordType.METADATA.toString())); + byte[] b = controlConversation.getBytes("UTF-8"); - w.writeMetadataRecord(curi.toString(), timestamp, - FTP_CONTROL_CONVERSATION_MIMETYPE, uid, headers, - new ByteArrayInputStream(b), b.length); - return uid; + + recordInfo.setContentStream(new ByteArrayInputStream(b)); + recordInfo.setContentLength((long) b.length); + + w.writeRecord(recordInfo); + + return recordInfo.getRecordId(); } protected URI writeRequest(final WARCWriter w, @@ -455,17 +551,29 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit final URI baseid, final CrawlURI curi, final ANVLRecord namedFields) throws IOException { - final URI uid = qualifyRecordID(baseid, TYPE, REQUEST); + WARCRecordInfo recordInfo = new WARCRecordInfo(); + recordInfo.setType(WARCRecordType.REQUEST); + recordInfo.setUrl(curi.toString()); + recordInfo.setCreate14DigitDate(timestamp); + recordInfo.setMimetype(mimetype); + recordInfo.setExtraHeaders(namedFields); + recordInfo.setContentLength(curi.getRecorder().getRecordedOutput().getSize()); + recordInfo.setEnforceLength(true); + + final URI uid = qualifyRecordID(baseid, TYPE, WARCRecordType.REQUEST.toString()); + recordInfo.setRecordId(uid); + ReplayInputStream ris = curi.getRecorder().getRecordedOutput().getReplayInputStream(); + recordInfo.setContentStream(ris); + try { - w.writeRequestRecord(curi.toString(), timestamp, mimetype, uid, - namedFields, ris, - curi.getRecorder().getRecordedOutput().getSize()); + w.writeRecord(recordInfo); } finally { IOUtils.closeQuietly(ris); } - return uid; + + return recordInfo.getRecordId(); } protected URI writeResponse(final WARCWriter w, @@ -473,16 +581,27 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit final URI baseid, final CrawlURI curi, final ANVLRecord namedFields) throws IOException { + WARCRecordInfo recordInfo = new WARCRecordInfo(); + recordInfo.setType(WARCRecordType.RESPONSE); + recordInfo.setUrl(curi.toString()); + recordInfo.setCreate14DigitDate(timestamp); + recordInfo.setMimetype(mimetype); + recordInfo.setRecordId(baseid); + recordInfo.setExtraHeaders(namedFields); + recordInfo.setContentLength(curi.getRecorder().getRecordedInput().getSize()); + recordInfo.setEnforceLength(true); + ReplayInputStream ris = curi.getRecorder().getRecordedInput().getReplayInputStream(); + recordInfo.setContentStream(ris); + try { - w.writeResponseRecord(curi.toString(), timestamp, mimetype, baseid, - namedFields, ris, - curi.getRecorder().getRecordedInput().getSize()); + w.writeRecord(recordInfo); } finally { IOUtils.closeQuietly(ris); } - return baseid; + + return recordInfo.getRecordId(); } protected URI writeResource(final WARCWriter w, @@ -490,15 +609,25 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit final URI baseid, final CrawlURI curi, final ANVLRecord namedFields) throws IOException { + WARCRecordInfo recordInfo = new WARCRecordInfo(); + recordInfo.setType(WARCRecordType.RESOURCE); + recordInfo.setUrl(curi.toString()); + recordInfo.setCreate14DigitDate(timestamp); + recordInfo.setMimetype(mimetype); + recordInfo.setRecordId(baseid); + recordInfo.setExtraHeaders(namedFields); + recordInfo.setContentLength(curi.getRecorder().getRecordedInput().getSize()); + recordInfo.setEnforceLength(true); + ReplayInputStream ris = curi.getRecorder().getRecordedInput().getReplayInputStream(); + recordInfo.setContentStream(ris); try { - w.writeResourceRecord(curi.toString(), timestamp, mimetype, baseid, - namedFields, ris, - curi.getRecorder().getRecordedInput().getSize()); + w.writeRecord(recordInfo); } finally { IOUtils.closeQuietly(ris); } - return baseid; + + return recordInfo.getRecordId(); } protected URI writeRevisitDigest(final WARCWriter w, @@ -518,20 +647,87 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit final String timestamp, final String mimetype, final URI baseid, final CrawlURI curi, final ANVLRecord namedFields, long contentLength) throws IOException { + WARCRecordInfo recordInfo = new WARCRecordInfo(); + recordInfo.setType(WARCRecordType.REVISIT); + recordInfo.setUrl(curi.toString()); + recordInfo.setCreate14DigitDate(timestamp); + recordInfo.setMimetype(mimetype); + recordInfo.setRecordId(baseid); + recordInfo.setContentLength(contentLength); + recordInfo.setEnforceLength(false); + namedFields.addLabelValue( HEADER_KEY_PROFILE, PROFILE_REVISIT_IDENTICAL_DIGEST); namedFields.addLabelValue( HEADER_KEY_TRUNCATED, NAMED_FIELD_TRUNCATED_VALUE_LENGTH); + recordInfo.setExtraHeaders(namedFields); + ReplayInputStream ris = curi.getRecorder().getRecordedInput().getReplayInputStream(); + recordInfo.setContentStream(ris); + try { - w.writeRevisitRecord(curi.toString(), timestamp, mimetype, baseid, - namedFields, ris, contentLength); + w.writeRecord(recordInfo); } finally { IOUtils.closeQuietly(ris); } curi.getAnnotations().add("warcRevisit:digest"); - return baseid; + + return recordInfo.getRecordId(); + } + + protected URI writeRevisitUriAgnosticDigest(WARCWriter w, String timestamp, + String mimetype, URI baseid, CrawlURI curi, + ANVLRecord headers) throws IOException { + + WARCRecordInfo recordInfo = new WARCRecordInfo(); + recordInfo.setType(WARCRecordType.REVISIT); + recordInfo.setUrl(curi.toString()); + recordInfo.setCreate14DigitDate(timestamp); + recordInfo.setMimetype(mimetype); + recordInfo.setRecordId(baseid); + recordInfo.setEnforceLength(false); + + long revisedLength = curi.getRecorder().getRecordedInput().getContentBegin(); + revisedLength = revisedLength > 0 ? revisedLength : curi.getRecorder().getRecordedInput().getSize(); + recordInfo.setContentLength(revisedLength); + + headers.addLabelValue( + HEADER_KEY_PROFILE, PROFILE_REVISIT_IDENTICAL_DIGEST); + headers.addLabelValue( + HEADER_KEY_TRUNCATED, NAMED_FIELD_TRUNCATED_VALUE_LENGTH); + + /* + * ISO 28500 WARC ISO standard draft says: "The WARC-Refers-To field may + * also be used to associate a record of type 'revisit' or 'conversion' + * with the preceding record which helped determine the present record + * content." + */ + headers.addLabelValue(HEADER_KEY_REFERS_TO, + curi.getContentDigestHistory().get(A_WARC_RECORD_ID).toString()); + headers.addLabelValue(HEADER_KEY_REFERS_TO_TARGET_URI, + curi.getContentDigestHistory().get(A_ORIGINAL_URL).toString()); + headers.addLabelValue(HEADER_KEY_REFERS_TO_DATE, + curi.getContentDigestHistory().get(A_ORIGINAL_DATE).toString()); + headers.addLabelValue(HEADER_KEY_REFERS_TO_FILENAME, + curi.getContentDigestHistory().get(A_WARC_FILENAME).toString()); + headers.addLabelValue(HEADER_KEY_REFERS_TO_FILE_OFFSET, + curi.getContentDigestHistory().get(A_WARC_FILE_OFFSET).toString()); + + recordInfo.setExtraHeaders(headers); + + ReplayInputStream ris = + curi.getRecorder().getRecordedInput().getReplayInputStream(); + recordInfo.setContentStream(ris); + + try { + w.writeRecord(recordInfo); + } finally { + IOUtils.closeQuietly(ris); + } + curi.getAnnotations().add("warcRevisit:uriAgnosticDigest"); + + return recordInfo.getRecordId(); } protected URI writeRevisitNotModified(final WARCWriter w, @@ -539,10 +735,22 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit final URI baseid, final CrawlURI puri, final ANVLRecord namedFields) throws IOException { - CrawlURI curi = (CrawlURI) puri; + CrawlURI curi = (CrawlURI) puri; + + WARCRecordInfo recordInfo = new WARCRecordInfo(); + recordInfo.setType(WARCRecordType.REVISIT); + recordInfo.setUrl(curi.toString()); + recordInfo.setCreate14DigitDate(timestamp); + recordInfo.setMimetype(null); + recordInfo.setRecordId(baseid); + recordInfo.setContentLength((long) 0); + recordInfo.setEnforceLength(false); + namedFields.addLabelValue( HEADER_KEY_PROFILE, PROFILE_REVISIT_NOT_MODIFIED); // save just enough context to understand basis of not-modified + recordInfo.setExtraHeaders(namedFields); + if(curi.isHttpTransaction()) { saveHeader(curi, namedFields, A_ETAG_HEADER, HEADER_KEY_ETAG); saveHeader(curi, namedFields, A_LAST_MODIFIED_HEADER, HEADER_KEY_LAST_MODIFIED); @@ -552,14 +760,15 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit NAMED_FIELD_TRUNCATED_VALUE_LENGTH); ReplayInputStream ris = curi.getRecorder().getRecordedInput().getReplayInputStream(); + recordInfo.setContentStream(ris); + try { - w.writeRevisitRecord(curi.toString(), timestamp, null, baseid, - namedFields, ris, 0); + w.writeRecord(recordInfo); } finally { IOUtils.closeQuietly(ris); } curi.getAnnotations().add("warcRevisit:notModified"); - return baseid; + return recordInfo.getRecordId(); } /** @@ -579,7 +788,16 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit final URI baseid, final CrawlURI curi, final ANVLRecord namedFields) throws IOException { - final URI uid = qualifyRecordID(baseid, TYPE, METADATA); + WARCRecordInfo recordInfo = new WARCRecordInfo(); + recordInfo.setType(WARCRecordType.METADATA); + recordInfo.setUrl(curi.toString()); + recordInfo.setCreate14DigitDate(timestamp); + recordInfo.setMimetype(ANVLRecord.MIMETYPE); + recordInfo.setExtraHeaders(namedFields); + recordInfo.setEnforceLength(true); + + recordInfo.setRecordId(qualifyRecordID(baseid, TYPE, WARCRecordType.METADATA.toString())); + // Get some metadata from the curi. // TODO: Get all curi metadata. // TODO: Use other than ANVL (or rename ANVL as NameValue or use @@ -640,9 +858,12 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit // Annotations. byte [] b = r.getUTF8Bytes(); - w.writeMetadataRecord(curi.toString(), timestamp, ANVLRecord.MIMETYPE, - uid, namedFields, new ByteArrayInputStream(b), b.length); - return uid; + recordInfo.setContentStream(new ByteArrayInputStream(b)); + recordInfo.setContentLength((long) b.length); + + w.writeRecord(recordInfo); + + return recordInfo.getRecordId(); } protected URI getRecordID() throws IOException { @@ -661,7 +882,7 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit if (cachedMetadata != null) { return cachedMetadata; } - ANVLRecord record = new ANVLRecord(7); + ANVLRecord record = new ANVLRecord(); record.addLabelValue("software", "Heritrix/" + ArchiveUtils.VERSION + " http://crawler.archive.org"); try { @@ -726,10 +947,10 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit buf.append("Processor: " + getClass().getName() + "\n"); buf.append(" Function: Writes WARCs\n"); buf.append(" Total CrawlURIs: " + urlsWritten + "\n"); - buf.append(" Revisit records: " + WARCWriter.getStat(stats, WARCWriter.REVISIT, WARCWriter.NUM_RECORDS) + "\n"); + buf.append(" Revisit records: " + WARCWriter.getStat(stats, WARCRecordType.REVISIT.toString(), WARCWriter.NUM_RECORDS) + "\n"); - long bytes = WARCWriter.getStat(stats, WARCWriter.RESPONSE, WARCWriter.CONTENT_BYTES) - + WARCWriter.getStat(stats, WARCWriter.RESOURCE, WARCWriter.CONTENT_BYTES); + long bytes = WARCWriter.getStat(stats, WARCRecordType.RESPONSE.toString(), WARCWriter.CONTENT_BYTES) + + WARCWriter.getStat(stats, WARCRecordType.RESOURCE.toString(), WARCWriter.CONTENT_BYTES); buf.append(" Crawled content bytes (including http headers): " + bytes + " (" + ArchiveUtils.formatBytesForDisplay(bytes) + ")\n"); diff --git a/modules/src/test/java/org/archive/modules/recrawl/ContentDigestHistoryTest.java b/modules/src/test/java/org/archive/modules/recrawl/ContentDigestHistoryTest.java new file mode 100644 index 00000000..371bcedd --- /dev/null +++ b/modules/src/test/java/org/archive/modules/recrawl/ContentDigestHistoryTest.java @@ -0,0 +1,137 @@ +/* + * 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.recrawl; + +import static org.archive.modules.recrawl.RecrawlAttributeConstants.A_ORIGINAL_URL; + +import java.io.IOException; +import java.util.logging.Logger; + +import org.apache.commons.io.FileUtils; +import org.archive.bdb.BdbModule; +import org.archive.modules.CrawlURI; +import org.archive.net.UURIFactory; +import org.archive.spring.ConfigPath; +import org.archive.util.Base32; +import org.archive.util.TmpDirTestCase; + +public class ContentDigestHistoryTest extends TmpDirTestCase { + + private static Logger logger = Logger.getLogger(ContentDigestHistoryTest.class.getName()); + + protected BdbModule bdb; + protected BdbContentDigestHistory historyStore; + protected ContentDigestHistoryStorer storer; + protected ContentDigestHistoryLoader loader; + + protected ContentDigestHistoryLoader loader() throws IOException { + if (loader == null) { + loader = new ContentDigestHistoryLoader(); + loader.setContentDigestHistory(historyStore()); + logger.info("created " + loader); + } + return loader; + } + + protected ContentDigestHistoryStorer storer() throws IOException { + if (storer == null) { + storer = new ContentDigestHistoryStorer(); + storer.setContentDigestHistory(historyStore()); + logger.info("created " + storer); + } + return storer; + } + + protected BdbContentDigestHistory historyStore() throws IOException { + if (historyStore == null) { + historyStore = new BdbContentDigestHistory(); + historyStore.setBdbModule(bdb()); + historyStore.start(); + logger.info("created " + historyStore); + } + return historyStore; + } + + protected BdbModule bdb() throws IOException { + if (bdb == null) { + ConfigPath basePath = new ConfigPath("testBase",getTmpDir().getAbsolutePath()); + ConfigPath bdbDir = new ConfigPath("bdb","bdb"); + bdbDir.setBase(basePath); + FileUtils.deleteDirectory(bdbDir.getFile()); + + bdb = new BdbModule(); + bdb.setDir(bdbDir); + bdb.start(); + logger.info("created " + bdb); + } + return bdb; + } + + public void testBasics() throws InterruptedException, IOException { + CrawlURI curi1 = new CrawlURI(UURIFactory.getInstance("http://example.org/1")); + + assertFalse(loader().shouldProcess(curi1)); + assertFalse(storer().shouldProcess(curi1)); + + // sha1 of "monkey\n", point is to have a value there + curi1.setContentDigest("sha1", Base32.decode("orfjublpcrnymm4seg5uk6vfoeu7kw6c")); + + assertTrue(loader().shouldProcess(curi1)); + assertTrue(storer().shouldProcess(curi1)); + + assertEquals("sha1:ORFJUBLPCRNYMM4SEG5UK6VFOEU7KW6C", historyStore().persistKeyFor(curi1)); + + assertFalse(curi1.hasContentDigestHistory()); + + loader().process(curi1); + + assertTrue(curi1.hasContentDigestHistory()); + assertTrue(curi1.getContentDigestHistory().isEmpty()); + + storer().process(curi1); + assertTrue(historyStore().store.isEmpty()); + + curi1.getContentDigestHistory().put(A_ORIGINAL_URL, "http://example.org/original"); + // curi1.getContentDigestHistory().put(A_WARC_RECORD_ID, ""); + // curi1.getContentDigestHistory().put(A_WARC_FILENAME, "test.warc.gz"); + // curi1.getContentDigestHistory().put(A_WARC_FILE_OFFSET, 98765432l); + // curi1.getContentDigestHistory().put(A_ORIGINAL_DATE, "20120101000000"); + // curi1.getContentDigestHistory().put(A_CONTENT_DIGEST_COUNT, 1); + + loader().process(curi1); + assertEquals("http://example.org/original", curi1.getContentDigestHistory().get(A_ORIGINAL_URL)); + + storer().process(curi1); + + assertFalse(historyStore().store.isEmpty()); + assertEquals(1, historyStore().store.size()); + + CrawlURI curi2 = new CrawlURI(UURIFactory.getInstance("http://example.org/2")); + curi2.setContentDigest("sha1", Base32.decode("orfjublpcrnymm4seg5uk6vfoeu7kw6c")); + + assertFalse(curi2.hasContentDigestHistory()); + + loader().process(curi2); + + assertTrue(curi2.hasContentDigestHistory()); + assertEquals("http://example.org/original", curi2.getContentDigestHistory().get(A_ORIGINAL_URL)); + } + + +} diff --git a/modules/src/test/java/org/archive/modules/writer/WARCWriterProcessorTest.java b/modules/src/test/java/org/archive/modules/writer/WARCWriterProcessorTest.java index 132366a0..64915e19 100644 --- a/modules/src/test/java/org/archive/modules/writer/WARCWriterProcessorTest.java +++ b/modules/src/test/java/org/archive/modules/writer/WARCWriterProcessorTest.java @@ -21,8 +21,6 @@ package org.archive.modules.writer; import java.io.File; import java.io.IOException; -import java.io.InputStream; -import java.net.URI; import java.util.Arrays; import java.util.Collection; import java.util.Collections; @@ -32,6 +30,7 @@ import org.apache.commons.httpclient.methods.GetMethod; import org.archive.io.WriterPool; import org.archive.io.WriterPoolMember; import org.archive.io.WriterPoolSettings; +import org.archive.io.warc.WARCRecordInfo; import org.archive.io.warc.WARCWriter; import org.archive.io.warc.WARCWriterPoolSettingsData; import org.archive.modules.CrawlMetadata; @@ -45,7 +44,6 @@ import org.archive.uid.RecordIDGenerator; import org.archive.uid.UUIDGenerator; import org.archive.util.FileUtils; import org.archive.util.TmpDirTestCase; -import org.archive.util.anvl.ANVLRecord; /** * Unit test for {@link WARCWriterProcessor}. @@ -141,11 +139,9 @@ public class WARCWriterProcessorTest extends ProcessorTestBase { public FailWARCWriter(AtomicInteger serial, WARCWriterPoolSettingsData settings) { super(serial, settings); } + @Override - protected void writeRecord(String type, String url, - String create14DigitDate, String mimetype, URI recordId, - ANVLRecord xtraHeaders, InputStream contentStream, - long contentLength, boolean enforceLength) throws IOException { + public void writeRecord(WARCRecordInfo recordInfo) throws IOException { throw new IOException("pretend no space left on device"); } }