From 7b3c9e51e3c13736449176b5b0a634875fc02707 Mon Sep 17 00:00:00 2001
From: Andrew Jackson <Andrew.Jackson@bl.uk>
Date: Fri, 12 Feb 2016 11:20:51 +0000
Subject: [PATCH 1/5] Add support for setting the hop and the prefetch count.

---
 .../crawler/frontier/AMQPUrlReceiver.java     | 43 +++++++++++++++----
 1 file changed, 35 insertions(+), 8 deletions(-)

diff --git a/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java b/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java
index 54ce3ffc..cc4bbfed 100644
--- a/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java
+++ b/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java
@@ -137,6 +137,28 @@ public class AMQPUrlReceiver implements Lifecycle, ApplicationListener<CrawlStat
         this.autoDelete = autoDelete;
     }
 
+    /**
+     * The maximum prefetch count to use, meaning the maximum number of messages
+     * to be consumed without being acknowledged. Use 'null' to specify there
+     * should be no upper limit (the default).
+     */
+    private Integer prefetchCount = null;
+
+    /**
+     * @return the prefetchCount
+     */
+    public Integer getPrefetchCount() {
+        return prefetchCount;
+    }
+
+    /**
+     * @param prefetchCount
+     *            the prefetchCount to set
+     */
+    public void setPrefetchCount(Integer prefetchCount) {
+        this.prefetchCount = prefetchCount;
+    }
+
     private transient Lock lock = new ReentrantLock(true);
 
     private transient boolean pauseConsumer = false;
@@ -194,6 +216,8 @@ public class AMQPUrlReceiver implements Lifecycle, ApplicationListener<CrawlStat
             channel().queueDeclare(getQueueName(), durable,
                     false, autoDelete, null);
             channel().queueBind(getQueueName(), getExchange(), getQueueName());
+            if (prefetchCount != null)
+                channel().basicQos(prefetchCount);
             consumerTag = channel().basicConsume(getQueueName(), false, consumer);
             logger.info("started AMQP consumer uri=" + getAmqpUri() + " exchange=" + getExchange() + " queueName=" + getQueueName() + " consumerTag=" + consumerTag);
         }
@@ -331,7 +355,9 @@ public class AMQPUrlReceiver implements Lifecycle, ApplicationListener<CrawlStat
                         getFrontier().schedule(curi);
                     }
                     if (logger.isLoggable(Level.FINE)) {
-                        logger.fine("scheduled " + curi);
+                        logger.fine("scheduled " + curi + " seed="
+                                + curi.isSeed() + " force=" + curi.forceFetch()
+                                + " pathFromSeed=" + curi.getPathFromSeed());
                     }
                 } catch (URIException e) {
                     logger.log(Level.WARNING,
@@ -351,6 +377,7 @@ public class AMQPUrlReceiver implements Lifecycle, ApplicationListener<CrawlStat
                         + decodedBody);
             }
 
+            logger.finest("Now ACKing: " + decodedBody);
             this.getChannel().basicAck(envelope.getDeliveryTag(), false);
         }
 
@@ -384,7 +411,8 @@ public class AMQPUrlReceiver implements Lifecycle, ApplicationListener<CrawlStat
 
             JSONObject parentUrlMetadata = jo.getJSONObject("parentUrlMetadata");
             String parentHopPath = parentUrlMetadata.getString("pathFromSeed");
-            String hopPath = parentHopPath + Hop.INFERRED.getHopString();
+            String hop = jo.optString("hop", Hop.INFERRED.getHopString());
+            String hopPath = parentHopPath + hop;
 
             CrawlURI curi = new CrawlURI(uuri, hopPath, via, LinkContext.INFERRED_MISC);
 
@@ -418,22 +446,21 @@ public class AMQPUrlReceiver implements Lifecycle, ApplicationListener<CrawlStat
              * urls really get crawled ahead of others. 
              * See https://webarchive.jira.com/wiki/display/Heritrix/Precedence+Feature+Notes
              */
-            curi.setSchedulingDirective(SchedulingConstants.HIGH);
-            curi.setPrecedence(1);
+            if (Hop.INFERRED.getHopString().equals(curi.getLastHop())
+                    || Hop.EMBED.getHopString().equals(curi.getLastHop())) {
+                curi.setSchedulingDirective(SchedulingConstants.HIGH);
+                curi.setPrecedence(1);
+            }
 
             // optional forceFetch instruction:
             if (jo.has("forceFetch")) {
                 boolean forceFetch = jo.getBoolean("forceFetch");
-                if (forceFetch)
-                    logger.info("Setting forceFetch=" + forceFetch);
                 curi.setForceFetch(forceFetch);
             }
 
             // optional isSeed instruction:
             if (jo.has("isSeed")) {
                 boolean isSeed = jo.getBoolean("isSeed");
-                if (isSeed)
-                    logger.info("Setting isSeed=" + isSeed);
                 curi.setSeed(isSeed);
                 // We want to force the seed version of a URL to be fetched even
                 // if the URL has been seen before. See

From 576b1d234bb78a34d4dbfe4d6b5d5a7d563a375d Mon Sep 17 00:00:00 2001
From: Andrew Jackson <Andrew.Jackson@bl.uk>
Date: Thu, 10 Mar 2016 14:41:40 +0000
Subject: [PATCH 2/5] Use sensible defaults for normal links.

---
 .../java/org/archive/crawler/frontier/AMQPUrlReceiver.java  | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java b/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java
index b11462e1..8b7e3157 100644
--- a/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java
+++ b/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java
@@ -423,7 +423,8 @@ public class AMQPUrlReceiver implements Lifecycle, ApplicationListener<CrawlStat
             }
             curi.getData().put("customHttpRequestHeaders", customHttpRequestHeaders);
 
-            /* Crawl job must be configured to use
+            /*
+             * Crawl job must be configured to use
              * HighestUriQueuePrecedencePolicy to ensure these high priority
              * urls really get crawled ahead of others. See
              * https://webarchive.jira.com/wiki/display/Heritrix/Precedence+
@@ -433,6 +434,9 @@ public class AMQPUrlReceiver implements Lifecycle, ApplicationListener<CrawlStat
                     || Hop.EMBED.getHopString().equals(curi.getLastHop())) {
                 curi.setSchedulingDirective(SchedulingConstants.HIGH);
                 curi.setPrecedence(1);
+            } else {
+                curi.setSchedulingDirective(SchedulingConstants.NORMAL);
+                curi.setPrecedence(1);
             }
 
             curi.setForceFetch(forceFetch || jo.optBoolean("forceFetch"));

From 621468107bb89f2911b12fcd95982b4ace1eded8 Mon Sep 17 00:00:00 2001
From: Andrew Jackson <Andrew.Jackson@bl.uk>
Date: Sun, 13 Mar 2016 20:22:27 +0000
Subject: [PATCH 3/5] Bump incoming links to MEDIUM to compete with e.g. runs
 of redirects.

---
 .../org/archive/crawler/frontier/AMQPUrlReceiver.java     | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java b/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java
index 8b7e3157..14210804 100644
--- a/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java
+++ b/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java
@@ -435,7 +435,13 @@ public class AMQPUrlReceiver implements Lifecycle, ApplicationListener<CrawlStat
                 curi.setSchedulingDirective(SchedulingConstants.HIGH);
                 curi.setPrecedence(1);
             } else {
-                curi.setSchedulingDirective(SchedulingConstants.NORMAL);
+                /*
+                 * By default, redirects get set to MEDIUM, so to ensure the
+                 * incoming links are caught promptly, even if the crawler hits
+                 * a lot of redirects (which we have observed for e.g. The
+                 * Guardian), we bump them to MEDIUM so these links can compete.
+                 */
+                curi.setSchedulingDirective(SchedulingConstants.MEDIUM);
                 curi.setPrecedence(1);
             }
 

From d349c192985856ef416f2b16866b7e815d6d20f7 Mon Sep 17 00:00:00 2001
From: Andrew Jackson <Andrew.Jackson@bl.uk>
Date: Mon, 23 May 2016 11:00:59 +0100
Subject: [PATCH 4/5] Update Kryo to latest version.

---
 commons/pom.xml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/commons/pom.xml b/commons/pom.xml
index 75c31503..952acabf 100644
--- a/commons/pom.xml
+++ b/commons/pom.xml
@@ -163,7 +163,7 @@
 		<dependency>
 			<groupId>com.esotericsoftware</groupId>
 			<artifactId>kryo</artifactId>
-			<version>1.01</version>
+			<version>1.04</version>
 			<scope>compile</scope>
 		</dependency>
 		<dependency>

From aba94c7ac1cec9c08c47ad8b2093a271fcfb2a2b Mon Sep 17 00:00:00 2001
From: Andrew Jackson <Andrew.Jackson@bl.uk>
Date: Mon, 23 May 2016 12:11:15 +0100
Subject: [PATCH 5/5] Fix package name.

---
 commons/pom.xml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/commons/pom.xml b/commons/pom.xml
index 952acabf..85ebf0e8 100644
--- a/commons/pom.xml
+++ b/commons/pom.xml
@@ -161,7 +161,7 @@
 		</dependency>
 
 		<dependency>
-			<groupId>com.esotericsoftware</groupId>
+			<groupId>com.googlecode</groupId>
 			<artifactId>kryo</artifactId>
 			<version>1.04</version>
 			<scope>compile</scope>