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] 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