From dd5e51fd9d5d0e7db33eaa24148bc331031a440c Mon Sep 17 00:00:00 2001 From: Noah Levitt Date: Mon, 10 Sep 2012 15:43:37 -0700 Subject: [PATCH] Working on cleaning up WARC writer code. * WARCRecordInfo.java new class to hold warc record info to write, so we don't have to pass around all these long lists of variables * WARCConstants.java new enum WARCRecordType replaces bunch of string constants * WARCWriter.java, WARCWriterProcessor.java, WARCWriterTest.java use WARCRecordInfo, WARCRecordType --- .../org/archive/io/warc/WARCConstants.java | 48 ++++---- .../org/archive/io/warc/WARCRecordInfo.java | 111 ++++++++++++++++++ .../java/org/archive/io/warc/WARCWriter.java | 108 +++++++++-------- .../org/archive/io/warc/WARCWriterTest.java | 4 +- .../modules/writer/WARCWriterProcessor.java | 26 ++-- .../writer/WARCWriterProcessorTest.java | 2 +- 6 files changed, 204 insertions(+), 95 deletions(-) create mode 100644 commons/src/main/java/org/archive/io/warc/WARCRecordInfo.java 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..685c9364 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 */ 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..a4422508 --- /dev/null +++ b/commons/src/main/java/org/archive/io/warc/WARCRecordInfo.java @@ -0,0 +1,111 @@ +/* + * 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; + + public WARCRecordInfo(WARCRecordType type, String url) { + this.type = type; + 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; + } +} 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..c0127bf3 100644 --- a/commons/src/main/java/org/archive/io/warc/WARCWriter.java +++ b/commons/src/main/java/org/archive/io/warc/WARCWriter.java @@ -36,8 +36,10 @@ 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.UTF8Bytes; import org.archive.io.WriterPoolMember; +import org.archive.modules.writer.WARCWriterProcessor; import org.archive.util.ArchiveUtils; import org.archive.util.anvl.ANVLRecord; import org.archive.util.anvl.Element; @@ -84,7 +86,7 @@ implements WARCConstants { * {@link #resetTmpStats()}, write some records, then add * {@link #getTmpStats()} into its long-term running totals. */ - private Map> tmpStats; + private Map> tmpStats; /** * Constructor. @@ -173,37 +175,38 @@ 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(); } @@ -211,7 +214,7 @@ implements WARCConstants { /** * @deprecated Use {@link #writeRecord(String,String,String,String,URI,ANVLRecord,InputStream,long,boolean)} instead */ - protected void writeRecord(final String type, final String url, + protected void writeRecord(final WARCRecordType type, final String url, final String create14DigitDate, final String mimetype, final URI recordId, ANVLRecord xtraHeaders, final InputStream contentStream, final long contentLength) @@ -219,30 +222,42 @@ implements WARCConstants { writeRecord(type, url, create14DigitDate, mimetype, recordId, xtraHeaders, contentStream, contentLength, true); } - protected void writeRecord(final String type, final String url, + /** @deprecated */ + public void writeRecord(final WARCRecordType 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)) { + + WARCRecordInfo metaRecord = new WARCRecordInfo(type, url); + metaRecord.setCreate14DigitDate(create14DigitDate); + metaRecord.setMimetype(mimetype); + metaRecord.setRecordId(recordId); + metaRecord.setExtraHeaders(xtraHeaders); + metaRecord.setContentStream(contentStream); + metaRecord.setContentLength(contentLength); + metaRecord.setEnforceLength(enforceLength); + + writeRecord(metaRecord); + } + + protected void writeRecord(WARCRecordInfo metaRecord) + throws IOException { + + if (metaRecord.getContentLength() == 0 && + (metaRecord.getExtraHeaders() == null || metaRecord.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(metaRecord); } 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: " + metaRecord.getType() + + "for URL: " + metaRecord.getUrl(), e); return; } @@ -260,11 +275,13 @@ implements WARCConstants { totalBytes += bytes.length; - if (contentStream != null && contentLength > 0) { + if (metaRecord.getContentStream() != null && metaRecord.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(metaRecord.getContentStream(), + metaRecord.getContentLength(), + metaRecord.getEnforceLength()); totalBytes += contentBytes; } @@ -277,21 +294,21 @@ implements WARCConstants { } // TODO: should this be in the finally block? - tally(type, contentBytes, totalBytes, getPosition() - startPosition); + tally(metaRecord.getType(), contentBytes, totalBytes, getPosition() - startPosition); } // 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); @@ -395,12 +412,12 @@ implements WARCConstants { final ANVLRecord namedFields, final InputStream fileMetadata, final long fileMetadataLength) throws IOException { - final URI recordid = generateRecordId(TYPE, WARCINFO); + final URI recordid = generateRecordId(TYPE, WARCRecordType.WARCINFO.toString()); 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. @@ -416,7 +433,7 @@ implements WARCConstants { final String mimetype, final URI recordId, final ANVLRecord namedFields, final InputStream fileMetadata, final long fileMetadataLength) throws IOException { - writeRecord(WARCINFO, null, create14DigitDate, mimetype, + writeRecord(WARCRecordType.WARCINFO, null, create14DigitDate, mimetype, recordId, namedFields, fileMetadata, fileMetadataLength, true); } @@ -426,7 +443,7 @@ implements WARCConstants { final ANVLRecord namedFields, final InputStream request, final long requestLength) throws IOException { - writeRecord(REQUEST, url, create14DigitDate, + writeRecord(WARCRecordType.REQUEST, url, create14DigitDate, mimetype, recordId, namedFields, request, requestLength, true); } @@ -447,7 +464,7 @@ implements WARCConstants { final ANVLRecord namedFields, final InputStream response, final long responseLength) throws IOException { - writeRecord(RESOURCE, url, create14DigitDate, + writeRecord(WARCRecordType.RESOURCE, url, create14DigitDate, mimetype, recordId, namedFields, response, responseLength, true); } @@ -458,7 +475,7 @@ implements WARCConstants { final ANVLRecord namedFields, final InputStream response, final long responseLength) throws IOException { - writeRecord(RESPONSE, url, create14DigitDate, + writeRecord(WARCRecordType.RESPONSE, url, create14DigitDate, mimetype, recordId, namedFields, response, responseLength, true); } @@ -469,22 +486,11 @@ implements WARCConstants { final ANVLRecord namedFields, final InputStream response, final long responseLength) throws IOException { - writeRecord(REVISIT, url, create14DigitDate, + writeRecord(WARCRecordType.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 */ 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..bd4f3b3e 100644 --- a/commons/src/test/java/org/archive/io/warc/WARCWriterTest.java +++ b/commons/src/test/java/org/archive/io/warc/WARCWriterTest.java @@ -144,12 +144,12 @@ extends TmpDirTestCase implements WARCConstants { headerFields.addLabelValue("x", "y"); headerFields.addLabelValue("a", "b"); - URI rid = (new UUIDGenerator()).getQualifiedRecordID(TYPE, METADATA); + URI rid = (new UUIDGenerator()).getQualifiedRecordID(TYPE, WARCRecordType.METADATA.toString()); 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/", + writer.writeRecord(WARCRecordType.METADATA, "http://www.archive.org/", ArchiveUtils.get14DigitDate(), "no/type", rid, headerFields, new ByteArrayInputStream(bodyBytes), (long)bodyBytes.length, true); 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 7aa407f3..1ce328d2 100644 --- a/modules/src/main/java/org/archive/modules/writer/WARCWriterProcessor.java +++ b/modules/src/main/java/org/archive/modules/writer/WARCWriterProcessor.java @@ -29,13 +29,11 @@ import static org.archive.io.warc.WARCConstants.HEADER_KEY_PROFILE; 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; @@ -74,6 +72,7 @@ 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.WARCWriter; import org.archive.io.warc.WARCWriterPool; import org.archive.io.warc.WARCWriterPoolSettings; @@ -446,11 +445,11 @@ 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); + final URI uid = 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); + w.writeRecord(WARCRecordType.METADATA, curi.toString(), timestamp, + FTP_CONTROL_CONVERSATION_MIMETYPE, uid, headers, new ByteArrayInputStream(b), + (long) b.length, true); return uid; } @@ -459,7 +458,7 @@ 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); + final URI uid = qualifyRecordID(baseid, TYPE, WARCRecordType.REQUEST.toString()); ReplayInputStream ris = curi.getRecorder().getRecordedOutput().getReplayInputStream(); try { @@ -588,7 +587,7 @@ 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); + final URI uid = 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 @@ -649,8 +648,9 @@ 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); + w.writeRecord(WARCRecordType.METADATA, curi.toString(), timestamp, + ANVLRecord.MIMETYPE, uid, namedFields, new ByteArrayInputStream(b), + (long) b.length, true); return uid; } @@ -735,10 +735,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/writer/WARCWriterProcessorTest.java b/modules/src/test/java/org/archive/modules/writer/WARCWriterProcessorTest.java index 8ea8b660..8dad6ec5 100644 --- a/modules/src/test/java/org/archive/modules/writer/WARCWriterProcessorTest.java +++ b/modules/src/test/java/org/archive/modules/writer/WARCWriterProcessorTest.java @@ -141,7 +141,7 @@ public class WARCWriterProcessorTest extends ProcessorTestBase { super(serial, settings); } @Override - protected void writeRecord(String type, String url, + public void writeRecord(WARCRecordType type, String url, String create14DigitDate, String mimetype, URI recordId, ANVLRecord xtraHeaders, InputStream contentStream, long contentLength, boolean enforceLength) throws IOException {