support mixing fetch histories from multiple sources (hbase, WBM).

main/java/org/archive/modules/recrawl/FetchHistoryHelper.java

@@ -0,0 +1,99 @@
+package org.archive.modules.recrawl;
+
+import java.text.DateFormat;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.archive.modules.CrawlURI;
+
+public class FetchHistoryHelper {
+  private static final Log logger = LogFactory.getLog(FetchHistoryHelper.class);
+  /**
+   * key for storing timestamp in crawl history map.
+   */
+  public static final String A_TIMESTAMP = ".ts";
+
+  public static String getHeaderValue(org.apache.commons.httpclient.HttpMethod method, String name) {
+      org.apache.commons.httpclient.Header header = method.getResponseHeader(name);
+      return header != null ? header.getValue() : null;
+  }
+
+  /**
+   * returns a Map to store recrawl data, positioned property in CrawlURI's
+   * fetch history array property, according to {@code timestamp}. this makes it possible
+   * to import crawl history data from multiple sources.
+   * @param uri target {@link CrawlURI}
+   * @param timestamp timestamp (in ms) of crawl history to be added.
+   * @return Map object to store recrawl data, or null if {@code timestamp} is older
+   * than existing crawl history entry and there's no room for it.
+   * @see #setHistoryLength(int)
+   */
+  @SuppressWarnings("unchecked")
+  public static Map<String, Object> getFetchHistory(CrawlURI uri, long timestamp, int historyLength) {
+    Map<String, Object> data = uri.getData();
+    Map<String, Object>[] history = (Map[])data.get(RecrawlAttributeConstants.A_FETCH_HISTORY);
+    if (history == null) {
+      // there's no history records at all.
+      // FetchHistoryProcessor assumes history is HashMap[], not Map[].
+      history = new HashMap[historyLength];
+      data.put(RecrawlAttributeConstants.A_FETCH_HISTORY, history);
+    }
+    for (int i = 0; i < history.length; i++) {
+      if (history[i] == null) {
+        history[i] = new HashMap<String, Object>();
+        history[i].put(A_TIMESTAMP, timestamp);
+        return history[i];
+      }
+      Object ts = history[i].get(A_TIMESTAMP);
+      // no timestamp value is regarded as older than anything.
+      if (!(ts instanceof Long) || timestamp > (Long)ts) {
+        if (i < history.length - 2) {
+          System.arraycopy(history, i, history, i + 1, history.length - i - 1);
+        } else if (i == history.length - 2) {
+          history[i + 1] = history[i];
+        }
+        history[i] = new HashMap<String, Object>();
+        history[i].put(A_TIMESTAMP, timestamp);
+        return history[i];
+      }
+    }
+    return null;
+  }
+
+  protected static final DateFormat HTTP_DATE_FORMAT = new SimpleDateFormat("EEE, dd MMM yyyy HH:mm:ss zzz");
+
+  protected FetchHistoryHelper() {
+  }
+
+  /**
+   * converts time in HTTP Date format {@code dateStr} to seconds
+   * since epoch. 
+   * @param dateStr time in HTTP Date format.
+   * @return seconds since epoch
+   */
+  public static long parseHttpDate(String dateStr) {
+      synchronized (HTTP_DATE_FORMAT) {
+          try {
+              Date d = HTTP_DATE_FORMAT.parse(dateStr);
+              return d.getTime() / 1000;
+          } catch (ParseException ex) {
+            if (logger.isDebugEnabled())
+              logger.debug("bad HTTP DATE: " + dateStr);
+              return 0;
+          }
+      }
+  }
+
+  public static String formatHttpDate(long time) {
+      synchronized (HTTP_DATE_FORMAT) {
+          // format is not thread safe either
+          return HTTP_DATE_FORMAT.format(new Date(time * 1000));
+      }
+  }
+
+}

main/java/org/archive/modules/recrawl/wbm/WbmPersistLoadProcessor.java

@@ -23,7 +23,9 @@ import java.io.InputStream;
 import java.io.UnsupportedEncodingException;
 import java.net.URLEncoder;
 import java.nio.ByteBuffer;
+import java.text.ParseException;
 import java.util.Date;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -49,16 +51,17 @@ import org.archive.modules.CrawlURI;
 import org.archive.modules.ProcessResult;
 import org.archive.modules.Processor;
 import org.archive.modules.hq.GzipInflatingHttpEntityWrapper;
-import org.archive.modules.hq.recrawl.RecrawlDataSchemaBase;
+import org.archive.modules.recrawl.FetchHistoryHelper;
 import org.archive.modules.recrawl.RecrawlAttributeConstants;
 import org.archive.util.ArchiveUtils;
+import org.archive.util.DateUtils;
 
 /**
  * A {@link Processor} for retrieving recrawl info from remote Wayback Machine index.
  * This is currently in the early stage of experiment. Both low-level protocol and WBM API
  * semantics will certainly undergo several revisions.
  * <p>Current interface:</p>
- * <p>http://web-beta.archive.org/cdx/search?url=archive.org&startDate=1999 will return raw
+ * <p>http://web-beta.archive.org/cdx/search/cdx?url=archive.org&startDate=1999 will return raw
  * CDX lines for archive.org, since 1999-01-01 00:00:00.
  * </p>
  * <p>As index is updated in a separate batch processing job, there's no "Store" counterpart.</p>
@@ -69,6 +72,15 @@ public class WbmPersistLoadProcessor extends Processor {
 
     private HttpClient client;
     
+    private int historyLength = 2;
+    
+    public void setHistoryLength(int historyLength) {
+      this.historyLength = historyLength;
+    }
+    public int getHistoryLength() {
+      return historyLength;
+    }
+    
     // ~Jan 2, 2013
     //private String queryURL = "http://web-beta.archive.org/cdx/search";
     private String queryURL = "http://web.archive.org/cdx/search/cdx";
@@ -249,17 +261,19 @@ public class WbmPersistLoadProcessor extends Processor {
 	return ProcessResult.PROCEED;
       }
       InputStream is = null;
-      String hash = null;
+      Map<String, Object> info = null;
       try {
-	hash = getLastHash(is = entity.getContent());
+	info = getLastCrawl(is = entity.getContent());
       } catch (IOException ex) {
       } finally {
 	if (is != null)
 	  ArchiveUtils.closeQuietly(is);
       }
-      if (hash != null) {
-	Map<String, Object> history = RecrawlDataSchemaBase.getFetchHistory(curi);
-	history.put(RecrawlAttributeConstants.A_CONTENT_DIGEST, contentDigestScheme + hash);
+      if (info != null) {
+	Map<String, Object> history = FetchHistoryHelper.getFetchHistory(curi,
+	    (Long)info.get(FetchHistoryHelper.A_TIMESTAMP), historyLength);
+	if (history != null)
+	  history.putAll(info);
 	loadedCount.incrementAndGet();
       } else {
 	failedCount.incrementAndGet();
@@ -267,34 +281,37 @@ public class WbmPersistLoadProcessor extends Processor {
       return ProcessResult.PROCEED;
     }
     
-    protected String getLastHash(InputStream is) throws IOException {
+    protected HashMap<String, Object> getLastCrawl(InputStream is) throws IOException {
       // read CDX lines, save most recent (at the end) hash.
       ByteBuffer buffer = ByteBuffer.allocate(32);
+      ByteBuffer tsbuffer = ByteBuffer.allocate(14);
       int field = 0;
       int c;
       do {
-	if (field == 5) {
-	  buffer.clear();
-	  while (buffer.remaining() > 0) {
+	c = is.read();
+	if (field == 1) {
+	  // 14-digits timestamp
+	  tsbuffer.clear();
+	  while (Character.isDigit(c) && tsbuffer.remaining() > 0) {
+	    tsbuffer.put((byte)c);
 	    c = is.read();
-	    if (c >= 'A' && c <= 'Z' || c >= '0' && c <= '9') {
-		buffer.put((byte)c);
-	    } else {
-	      break;
-	    }
 	  }
-	  if (buffer.remaining() == 0) {
+	  if (c != ' ' || tsbuffer.position() != 14) {
+	    tsbuffer.clear();
+	  }
+	  // fall through to skip the rest
+	} else if (field == 5) {
+	  buffer.clear();
+	  while ((c >= 'A' && c <= 'Z' || c >= '0' && c <= '9') && buffer.remaining() > 0) {
+	    buffer.put((byte)c);
 	    c = is.read();
-	    if (c == ' ') {
-	      field++;
-	      continue;
-	    }
-	  } else {
+	  }
+	  if (c != ' ' || buffer.position() != 32) {
 	    buffer.clear();
 	  }
+	  // fall through to skip the rest
 	}
-	while  (true) {
-	  c = is.read();
+	while (true) {
 	  if (c == -1) {
 	    break;
 	  } else if (c == '\n') {
@@ -304,13 +321,21 @@ public class WbmPersistLoadProcessor extends Processor {
 	    field++;
 	    break;
 	  }
+	  c = is.read();
 	}
       } while (c != -1);
+      
+      HashMap<String, Object> info = new HashMap<String, Object>();
       if (buffer.remaining() == 0) {
-	return new String(buffer.array());
-      } else {
-	return null;
+	info.put(RecrawlAttributeConstants.A_CONTENT_DIGEST, contentDigestScheme + new String(buffer.array()));
       }
+      if (tsbuffer.remaining() == 0) {
+	try {
+	  info.put(FetchHistoryHelper.A_TIMESTAMP, DateUtils.parse14DigitDate(new String(tsbuffer.array())).getTime());
+	} catch (ParseException ex) {
+	}
+      }
+      return info.isEmpty() ? null : info;
     }
 
     /**

test/java/org/archive/modules/recrawl/wbm/WbmPersistLoadProcessorTest.java

@@ -8,6 +8,7 @@ import java.io.BufferedReader;
 import java.io.InputStream;
 import java.io.InputStreamReader;
 import java.net.URLEncoder;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -23,9 +24,10 @@ import org.apache.http.message.BasicHttpResponse;
 import org.apache.http.message.BasicStatusLine;
 import org.archive.modules.CrawlURI;
 import org.archive.modules.ProcessResult;
-import org.archive.modules.hq.recrawl.RecrawlDataSchemaBase;
+import org.archive.modules.recrawl.FetchHistoryHelper;
 import org.archive.modules.recrawl.RecrawlAttributeConstants;
 import org.archive.net.UURIFactory;
+import org.archive.util.DateUtils;
 import org.easymock.EasyMock;
 import org.junit.Test;
 
@@ -37,6 +39,7 @@ import com.google.common.util.concurrent.ExecutionList;
  * 
  * TODO:
  * <ul>
+ * <li>test for pathological cases: illegal chars, incorrect length, etc.
  * <li>test if connection is properly released back to the pool for all possible cases.
  * </ul>
  * @author kenji
@@ -57,16 +60,17 @@ public class WbmPersistLoadProcessorTest {
   }
   
   /**
-   * stub HttpResponse for normal case
+   * stub HttpResponse for normal case.
    * @author kenji
    *
    */
   public static class TestNormalHttpResponse extends BasicHttpResponse {
+    public static final String EXPECTED_TS = "20121101155310";
     public static final String EXPECTED_HASH = "GHN5VKF3TBKNSEZTASOM23BJRTKFFNJK";
     public TestNormalHttpResponse() {
       super(new BasicStatusLine(new ProtocolVersion("HTTP", 1, 0), 200, "OK"));
       setEntity(new ByteArrayEntity(
-	  ("org,archive)/ 20121101155310 http://archive.org/ text/html 200 "+
+	  ("org,archive)/ "+EXPECTED_TS+" http://archive.org/ text/html 200 "+
 	      EXPECTED_HASH+" - - 6908 982548871 "+
 	      "google.es-20121101-155506/IA-FOC-google.es-20121101073708-00001.warc.gz\n"
 	  ).getBytes()
@@ -74,6 +78,15 @@ public class WbmPersistLoadProcessorTest {
     }
   }
   
+  protected Map<String, Object> getFetchHistory(CrawlURI curi, int idx) {
+    Map<String, Object> data = curi.getData();
+    @SuppressWarnings("unchecked")
+    Map<String, Object>[] historyArray = (Map[])data.get(RecrawlAttributeConstants.A_FETCH_HISTORY);
+    assertNotNull(historyArray);
+    Map<String, Object> history = historyArray[idx];
+    return history;
+  }
+  
   @Test
   public void testInnerProcessResultSingleShotWithMock() throws Exception {
     // because AbstractHttpClient marks most execute(...) methods final, it is very tiresome to implement
@@ -88,13 +101,31 @@ public class WbmPersistLoadProcessorTest {
     t.setHttpClient(client);
     t.setContentDigestScheme(CONTENT_DIGEST_SCHEME);
     CrawlURI curi = new CrawlURI(UURIFactory.getInstance("http://archive.org/"));
+    
+    // put history entry newer than being loaded
+    long expected_ts = DateUtils.parse14DigitDate(TestNormalHttpResponse.EXPECTED_TS).getTime();
+    Map<String, Object>[] fetchHistory = (Map[])curi.getData().get(RecrawlAttributeConstants.A_FETCH_HISTORY);
+    if (fetchHistory == null) {
+      fetchHistory = new HashMap[2];
+      curi.getData().put(RecrawlAttributeConstants.A_FETCH_HISTORY, fetchHistory);
+    }
+    fetchHistory[0] = new HashMap<String, Object>();
+    fetchHistory[0].put(FetchHistoryHelper.A_TIMESTAMP, expected_ts + 2000);
+    fetchHistory[1] = new HashMap<String, Object>();
+    fetchHistory[1].put(FetchHistoryHelper.A_TIMESTAMP, expected_ts - 2000);
+    
     ProcessResult result = t.innerProcessResult(curi);
     assertEquals("result is PROCEED", ProcessResult.PROCEED, result);
     
-    Map<String, Object> history = RecrawlDataSchemaBase.getFetchHistory(curi);
+    // newly loaded history entry should fall in between two existing entries (index=1)
+    Map<String, Object> history = getFetchHistory(curi, 1);
     assertNotNull("history", history);
     String hash = (String)history.get(RecrawlAttributeConstants.A_CONTENT_DIGEST);
     assertEquals("CONTENT_DIGEST", CONTENT_DIGEST_SCHEME+TestNormalHttpResponse.EXPECTED_HASH, hash);
+    
+    Long ts = (Long)history.get(FetchHistoryHelper.A_TIMESTAMP);
+    assertNotNull("ts is non-null", ts);
+    assertEquals("'ts' has expected timestamp", expected_ts, ts.longValue());
   }
   
   @Test
@@ -103,7 +134,7 @@ public class WbmPersistLoadProcessorTest {
     //CrawlURI curi = new CrawlURI(UURIFactory.getInstance("http://archive.org/"));
     CrawlURI curi = new CrawlURI(UURIFactory.getInstance("http://www.mext.go.jp/null.gif"));
     ProcessResult result = t.innerProcessResult(curi);
-    Map<String, Object> history = RecrawlDataSchemaBase.getFetchHistory(curi);
+    Map<String, Object> history = getFetchHistory(curi, 0);
     assertNotNull("getFetchHistory returns non-null", history);
     String hash = (String)history.get(RecrawlAttributeConstants.A_CONTENT_DIGEST);
     assertNotNull("CONTENT_DIGEST is non-null", hash);