don't wait if h3 sends no url

contrib/src/main/java/org/archive/crawler/event/AMQPUrlPublishedEvent.java

@@ -0,0 +1,47 @@
+/*
+ *  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.crawler.event;
+
+import org.archive.modules.AMQPPublishProcessor;
+import org.archive.modules.CrawlURI;
+import org.springframework.context.ApplicationEvent;
+
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.AMQP.BasicProperties;
+
+/**
+ * ApplicationEvent published when Heritrix sends a URL to AMQP.
+ * Other modules can observe this event to learn when Heritrix sends a URL.
+ *
+ * @contributor galgeek
+ */
+public class AMQPUrlPublishedEvent extends ApplicationEvent {
+    private static final long serialVersionUID = 1L;
+
+    protected CrawlURI curi;
+    public CrawlURI getCuri() {
+        return curi;
+    }
+
+    public AMQPUrlPublishedEvent(AMQPPublishProcessor source, CrawlURI curi) {
+        super(source);
+        this.curi = curi;
+    }
+}

contrib/src/main/java/org/archive/modules/AMQPPublishProcessor.java

@@ -28,9 +28,14 @@ import java.util.Map;
 import java.util.Set;
 
 import org.apache.commons.httpclient.URIException;
+import org.archive.crawler.event.AMQPUrlPublishedEvent;
 import org.archive.crawler.frontier.AMQPUrlReceiver;
 import org.archive.modules.fetcher.FetchHTTP;
 import org.json.JSONObject;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.context.ApplicationEvent;
+import org.springframework.beans.BeansException;
 
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.AMQP.BasicProperties;
@@ -39,12 +44,17 @@ import com.rabbitmq.client.AMQP.BasicProperties;
  * @author eldondev
  * @contributor nlevitt
  */
-public class AMQPPublishProcessor extends AMQPProducerProcessor implements Serializable {
+public class AMQPPublishProcessor extends AMQPProducerProcessor implements Serializable, ApplicationContextAware {
 
     private static final long serialVersionUID = 2L;
 
     public static final String A_SENT_TO_AMQP = "sentToAMQP"; // annotation
 
+    protected ApplicationContext appCtx;
+    public void setApplicationContext(ApplicationContext appCtx) throws BeansException {
+        this.appCtx = appCtx;
+    }
+
     public AMQPPublishProcessor() {
         // set default values
         setExchange("umbra");
@@ -145,6 +155,7 @@ public class AMQPPublishProcessor extends AMQPProducerProcessor implements Seria
     protected void success(CrawlURI curi, byte[] message, BasicProperties props) {
         super.success(curi, message, props);
         curi.getAnnotations().add(A_SENT_TO_AMQP);
+        appCtx.publishEvent(new AMQPUrlPublishedEvent(AMQPPublishProcessor.this, curi));
     }
 
     protected BasicProperties props = new AMQP.BasicProperties.Builder().

contrib/src/main/java/org/archive/modules/AMQPUrlWaiter.java

@@ -22,6 +22,7 @@ package org.archive.modules;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
+import org.archive.crawler.event.AMQPUrlPublishedEvent;
 import org.archive.crawler.event.AMQPUrlReceivedEvent;
 import org.archive.crawler.event.StatSnapshotEvent;
 import org.archive.crawler.framework.CrawlController;
@@ -42,6 +43,7 @@ public class AMQPUrlWaiter implements ApplicationListener<ApplicationEvent> {
 
     static protected final Logger logger = Logger.getLogger(AMQPUrlWaiter.class.getName());
 
+    protected int urlsPublished = 0;
     protected int urlsReceived = 0;
 
     protected CrawlController controller;
@@ -65,7 +67,9 @@ public class AMQPUrlWaiter implements ApplicationListener<ApplicationEvent> {
 
     @Override
     public void onApplicationEvent(ApplicationEvent event) {
-        if (event instanceof AMQPUrlReceivedEvent) {
+        if (event instanceof AMQPUrlPublishedEvent) {
+            urlsPublished += 1;
+        } else if (event instanceof AMQPUrlReceivedEvent) {
             urlsReceived += 1;
         } else if (event instanceof StatSnapshotEvent) {
             checkAMQPUrlWait();
@@ -73,9 +77,10 @@ public class AMQPUrlWaiter implements ApplicationListener<ApplicationEvent> {
     }
 
     protected void checkAMQPUrlWait() {
-        if (frontier.isEmpty() && urlsReceived > 0) {
+        if (frontier.isEmpty() && (urlsPublished == 0 || urlsReceived > 0)) {
             logger.info("frontier is empty and we have received " + urlsReceived +
-                        " urls from AMQP, stopping crawl with status " + CrawlStatus.FINISHED);
+                        " urls from AMQP, and published " + urlsPublished +
+                        ", stopping crawl with status " + CrawlStatus.FINISHED);
             controller.requestCrawlStop(CrawlStatus.FINISHED);
         }
     }