diff --git a/contrib/src/main/java/org/archive/crawler/event/AMQPUrlPublishedEvent.java b/contrib/src/main/java/org/archive/crawler/event/AMQPUrlPublishedEvent.java new file mode 100644 index 00000000..c8249171 --- /dev/null +++ b/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; + } +} diff --git a/contrib/src/main/java/org/archive/modules/AMQPPublishProcessor.java b/contrib/src/main/java/org/archive/modules/AMQPPublishProcessor.java index d1f2ac29..aa11abcd 100644 --- a/contrib/src/main/java/org/archive/modules/AMQPPublishProcessor.java +++ b/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(). diff --git a/contrib/src/main/java/org/archive/modules/AMQPUrlWaiter.java b/contrib/src/main/java/org/archive/modules/AMQPUrlWaiter.java index 7c8f12c5..86c79631 100644 --- a/contrib/src/main/java/org/archive/modules/AMQPUrlWaiter.java +++ b/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 { 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 { @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 { } 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); } }