lee/heritrix3
1
0
Fork 0
mirror of https://github.com/internetarchive/heritrix3.git synced 2026-09-24 14:45:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #166 from internetarchive/amqpWaiterStarter

Browse Source 
Don't wait to receive Umbra urls if Heritrix sends no url to Umbra
This commit is contained in:
Noah Levitt
2016-07-21 18:05:10 -05:00
committed by GitHub
parent af2530edc4 cbc73a6e8d
commit fbb0173459
3 changed files with 68 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
contrib/src/main/java/org/archive/crawler/event/AMQPUrlPublishedEvent.java
+47
View File
@@ -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
+12 -1
View File
@@ -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
+9 -15
View File
@@ -22,11 +22,11 @@ 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;
import org.archive.crawler.framework.CrawlStatus;
import org.archive.crawler.framework.Frontier;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
@@ -42,6 +42,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;
@@ -53,19 +54,11 @@ public class AMQPUrlWaiter implements ApplicationListener<ApplicationEvent> {
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) {
if (event instanceof AMQPUrlPublishedEvent) {
urlsPublished += 1;
} else if (event instanceof AMQPUrlReceivedEvent) {
urlsReceived += 1;
} else if (event instanceof StatSnapshotEvent) {
checkAMQPUrlWait();
@@ -73,9 +66,10 @@ public class AMQPUrlWaiter implements ApplicationListener<ApplicationEvent> {
}
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);
if (controller.getState() == CrawlController.State.EMPTY && (urlsPublished == 0 || urlsReceived > 0)) {
logger.info("crawl controller state is empty and we have received " + urlsReceived +
" urls from AMQP, and published " + urlsPublished +
", stopping crawl with status " + CrawlStatus.FINISHED);
controller.requestCrawlStop(CrawlStatus.FINISHED);
}
}