From 4485b8c8c7e3bcb6cf136b22778e335b52469691 Mon Sep 17 00:00:00 2001 From: Barbara Miller Date: Thu, 23 Jun 2016 19:21:57 -0700 Subject: [PATCH] AMQP URL Waiter --- .../crawler/event/AMQPUrlReceivedEvent.java | 47 +++++++++++ .../crawler/frontier/AMQPUrlReceiver.java | 14 +++- .../org/archive/modules/AMQPUrlWaiter.java | 82 +++++++++++++++++++ 3 files changed, 142 insertions(+), 1 deletion(-) create mode 100644 contrib/src/main/java/org/archive/crawler/event/AMQPUrlReceivedEvent.java create mode 100644 contrib/src/main/java/org/archive/modules/AMQPUrlWaiter.java diff --git a/contrib/src/main/java/org/archive/crawler/event/AMQPUrlReceivedEvent.java b/contrib/src/main/java/org/archive/crawler/event/AMQPUrlReceivedEvent.java new file mode 100644 index 00000000..ba09180e --- /dev/null +++ b/contrib/src/main/java/org/archive/crawler/event/AMQPUrlReceivedEvent.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.crawler.frontier.AMQPUrlReceiver; +import org.archive.modules.CrawlURI; +import org.springframework.context.ApplicationEvent; + +import com.rabbitmq.client.AMQP; +import com.rabbitmq.client.AMQP.BasicProperties; + +/** + * ApplicationEvent published when AMQPUrlReceiver receives a URL. + * Other modules can observe this event to learn when AMQPUrlReceiver receives a URL. + * + * @contributor galgeek + */ +public class AMQPUrlReceivedEvent extends ApplicationEvent { + private static final long serialVersionUID = 1L; + + protected CrawlURI curi; + public CrawlURI getCuri() { + return curi; + } + + public AMQPUrlReceivedEvent(AMQPUrlReceiver source, CrawlURI curi) { + super(source); + this.curi = curi; + } +} 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 859d716d..40da520a 100644 --- a/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java +++ b/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java @@ -32,6 +32,7 @@ import java.util.logging.Level; import java.util.logging.Logger; import org.apache.commons.httpclient.URIException; +import org.archive.crawler.event.AMQPUrlReceivedEvent; import org.archive.crawler.event.CrawlStateEvent; import org.archive.crawler.postprocessor.CandidatesProcessor; import org.archive.modules.CrawlURI; @@ -45,7 +46,11 @@ import org.json.JSONArray; import org.json.JSONException; import org.json.JSONObject; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationListener; +import org.springframework.beans.BeansException; import org.springframework.context.Lifecycle; import com.rabbitmq.client.AMQP.BasicProperties; @@ -60,7 +65,8 @@ import com.rabbitmq.client.ShutdownSignalException; /** * @contributor nlevitt */ -public class AMQPUrlReceiver implements Lifecycle, ApplicationListener { +public class AMQPUrlReceiver + implements Lifecycle, ApplicationContextAware, ApplicationListener { @SuppressWarnings("unused") private static final long serialVersionUID = 2L; @@ -70,6 +76,11 @@ public class AMQPUrlReceiver implements Lifecycle, ApplicationListener { + + public AMQPUrlWaiter() {} + + static protected final Logger logger = Logger.getLogger(AMQPUrlWaiter.class.getName()); + + protected int urlsReceived = 0; + + protected CrawlController controller; + public CrawlController getCrawlController() { + return this.controller; + } + @Autowired + public void setCrawlController(CrawlController controller) { + this.controller = controller; + } + + protected Frontier frontier; + public Frontier getFrontier() { + return this.frontier; + } + /** Autowired frontier, needed to determine when a url is finished. */ + @Autowired + public void setFrontier(Frontier frontier) { + this.frontier = frontier; + } + + @Override + public void onApplicationEvent(ApplicationEvent event) { + if (event instanceof AMQPUrlReceivedEvent) { + urlsReceived += 1; + } else if (event instanceof StatSnapshotEvent) { + checkAMQPUrlWait(); + } + } + + protected void checkAMQPUrlWait() { + if (frontier.isEmpty() && urlsReceived > 0) { + logger.info("frontier is empty and we have received " + urlsReceived + + " urls from AMQP, stopping crawl with status " + CrawlStatus.FINISHED); + controller.requestCrawlStop(CrawlStatus.FINISHED); + } + } +}