lee/heritrix3
1
0
Fork 0
mirror of https://github.com/internetarchive/heritrix3.git synced 2026-07-18 21:47:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #38 from nlevitt/umbra

Browse Source 
Umbra - AMQPUrlReceiver.java receive urls via amqp and add to frontier, related changes
This commit is contained in:
Eldon
2014-02-06 16:51:03 -05:00
parent 63b70b427e 6dbcac40d9
commit 704f4aa443
4 changed files with 315 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
commons/pom.xml
-5
View File
@@ -162,11 +162,6 @@
<artifactId>spring-expression</artifactId>
<version>3.0.5.RELEASE</version>
</dependency>
<dependency>
<groupId>org.json</groupId>
<artifactId>json</artifactId>
<version>20090211</version>
</dependency>
<dependency>
<groupId>com.esotericsoftware</groupId>
contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java
+244
View File
@@ -0,0 +1,244 @@
/*
* 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.frontier;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.HashMap;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.commons.httpclient.URIException;
import org.apache.commons.lang.StringUtils;
import org.archive.crawler.framework.Frontier;
import org.archive.modules.CrawlURI;
import org.archive.modules.SchedulingConstants;
import org.archive.modules.extractor.LinkContext;
import org.archive.net.UURI;
import org.archive.net.UURIFactory;
import org.json.JSONException;
import org.json.JSONObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.Lifecycle;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
/**
* @contributor nlevitt
*/
public class AMQPUrlReceiver implements Lifecycle {
@SuppressWarnings("unused")
private static final long serialVersionUID = 1L;
private static final Logger logger =
Logger.getLogger(AMQPUrlReceiver.class.getName());
public static final String A_RECEIVED_FROM_AMQP = "receivedFromAMQP";
protected Frontier frontier;
public Frontier getFrontier() {
return this.frontier;
}
@Autowired
public void setFrontier(Frontier frontier) {
this.frontier = frontier;
}
protected String amqpUri = "amqp://guest:guest@localhost:5672/%2f";
public String getAmqpUri() {
return this.amqpUri;
}
public void setAmqpUri(String uri) {
this.amqpUri = uri;
}
protected String queueName = "requests";
public String getQueueName() {
return queueName;
}
public void setQueueName(String queueName) {
this.queueName = queueName;
}
protected boolean isRunning = false;
@Override
public boolean isRunning() {
return isRunning;
}
@Override
synchronized public void start() {
while (!isRunning) {
try {
Consumer consumer = new UrlConsumer(channel());
channel().basicConsume(getQueueName(), false, consumer);
isRunning = true;
} catch (IOException e) {
logger.log(Level.SEVERE, "problem starting AMQP consumer (will try again after 30 seconds)", e);
try {
Thread.sleep(30000);
} catch (InterruptedException e1) {
}
}
}
}
@Override
synchronized public void stop() {
logger.info("shutting down");
if (connection != null && connection.isOpen()) {
try {
connection.close();
} catch (IOException e) {
logger.log(Level.SEVERE, "problem closing AMQP connection", e);
}
}
}
transient protected Connection connection = null;
transient protected Channel channel = null;
synchronized protected Connection connection() throws IOException {
if (connection != null && !connection.isOpen()) {
logger.warning("connection is closed, creating a new one");
connection = null;
}
if (connection == null) {
ConnectionFactory factory = new ConnectionFactory();
try {
factory.setUri(getAmqpUri());
} catch (Exception e) {
throw new IOException("problem with AMQP uri " + getAmqpUri(), e);
}
connection = factory.newConnection();
}
return connection;
}
synchronized protected Channel channel() throws IOException {
if (channel != null && !channel.isOpen()) {
logger.warning("channel is not open, creating a new one");
channel = null;
}
if (channel == null) {
channel = connection().createChannel();
}
return channel;
}
// XXX should we be using QueueingConsumer because of possible blocking in
// frontier.schedule()?
// "Note: all methods of this interface are invoked inside the Connection's
// thread. This means they a) should be non-blocking and generally do little
// work, b) must not call Channel or Connection methods, or a deadlock will
// ensue. One way of ensuring this is to use/subclass QueueingConsumer."
protected class UrlConsumer extends DefaultConsumer {
public UrlConsumer(Channel channel) {
super(channel);
}
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
BasicProperties properties, byte[] body) throws IOException {
String decodedBody;
try {
decodedBody = new String(body, "UTF-8");
} catch (UnsupportedEncodingException e) {
throw new RuntimeException(e); // can't happen
}
JSONObject jo = new JSONObject(decodedBody);
CrawlURI curi;
try {
curi = makeCrawlUri(jo);
// bypasses scoping (unless rechecking is configured)
getFrontier().schedule(curi);
if (logger.isLoggable(Level.FINE)) {
logger.fine("scheduled " + curi);
}
} catch (URIException e) {
logger.log(Level.SEVERE,
"problem creating CrawlURI from json received via AMQP "
+ decodedBody, e);
} catch (JSONException e) {
logger.log(Level.SEVERE,
"problem creating CrawlURI from json received via AMQP "
+ decodedBody, e);
}
this.getChannel().basicAck(envelope.getDeliveryTag(), false);
}
// {
// "headers": {
// "Referer": "https://archive.org/",
// "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Ubuntu Chromium/32.0.1700.102 Chrome/32.0.1700.102 Safari/537.36",
// "Accept": "image/webp,*/*;q=0.8"
// },
// "url": "https://analytics.archive.org/0.gif?server_ms=256&server_name=www19.us.archive.org&service=ao&loadtime=358&timediff=-8&locale=en-US&referrer=-&version=2&count=9",
// "method": "GET"
// }
protected CrawlURI makeCrawlUri(JSONObject jo) throws URIException,
JSONException {
JSONObject joHeaders = jo.getJSONObject("headers");
UURI uuri = UURIFactory.getInstance(jo.getString("url"));
UURI via = null;
if (joHeaders.has("Referer")) {
String referer = joHeaders.getString("Referer");
if (StringUtils.isNotEmpty(referer)) {
via = UURIFactory.getInstance(referer);
}
}
// XXX pathFromSeed? viaContext?
CrawlURI curi = new CrawlURI(uuri, "?", via, LinkContext.INFERRED_MISC);
Map<String, String> customHttpRequestHeaders = new HashMap<String, String>();
for (Object key : joHeaders.keySet()) {
customHttpRequestHeaders.put(key.toString(),
joHeaders.getString(key.toString()));
}
curi.getData().put("customHttpRequestHeaders", customHttpRequestHeaders);
/* Use HighestUriQueuePrecedencePolicy to ensure these high priority
* 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);
curi.getAnnotations().add(A_RECEIVED_FROM_AMQP);
return curi;
}
}
}
contrib/src/main/java/org/archive/modules/AMQPPublishProcessor.java
+63 -36
View File
@@ -23,18 +23,20 @@ import java.io.IOException;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.commons.httpclient.URIException;
import org.archive.crawler.frontier.AMQPUrlReceiver;
import org.archive.modules.fetcher.FetchHTTP;
import org.json.JSONObject;
import org.springframework.beans.factory.annotation.Required;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/**
* @author eldondev
* @version $Date$, $Revision$
* @contributor nlevitt
*/
public class AMQPPublishProcessor extends Processor {
@@ -43,60 +45,85 @@ public class AMQPPublishProcessor extends Processor {
private static final Logger logger =
Logger.getLogger(AMQPPublishProcessor.class.getName());
public static final int S_SENT_TO_AMQP = 10001; // artificial fetch status
public static final String A_SENT_TO_AMQP = "sentToAMQP"; // annotation
protected String amqpUri = null;
protected Connection connection = null;
protected String amqpUri = "amqp://guest:guest@localhost:5672/%2f";
public String getAmqpUri() {
return this.amqpUri;
}
@Required
public void setAmqpUri(String uri) {
this.amqpUri = uri;
}
transient protected Connection connection = null;
transient protected ThreadLocal<Channel> threadChannel =
new ThreadLocal<Channel>();
private String queueName = "umbra";
public String getQueueName() {
return queueName;
protected String exchange = "umbra";
public String getExchange() {
return exchange;
}
public void setExchange(String exchange) {
this.exchange = exchange;
}
protected String routingKey = "url";
public String getRoutingKey() {
return routingKey;
}
public void setRoutingKey(String routingKey) {
this.routingKey = routingKey;
}
public void setQueueName(String queueName) {
this.queueName = queueName;
}
/**
* Constructor.
/*
* Don't send urls received via AMQP back to AMQP, don't handle robots.txt,
* only handle http/s urls.
*/
public AMQPPublishProcessor() {
super();
}
protected boolean shouldProcess(CrawlURI curi) {
if (!(curi.getUURI().getScheme().equals(FetchHTTP.HTTP_SCHEME) || curi.getUURI().getScheme().equals(FetchHTTP.HTTPS_SCHEME))) {
// handles only plain http and https
return false;
}
return true;
try {
return !curi.getAnnotations().contains(AMQPUrlReceiver.A_RECEIVED_FROM_AMQP)
&& !"/robots.txt".equals(curi.getUURI().getPath())
&& (curi.getUURI().getScheme().equals(FetchHTTP.HTTP_SCHEME)
|| curi.getUURI().getScheme().equals(FetchHTTP.HTTPS_SCHEME));
} catch (URIException e) {
throw new RuntimeException(e);
}
}
protected void innerProcess(CrawlURI uri) throws InterruptedException {
@Override
protected ProcessResult innerProcessResult(CrawlURI curi)
throws InterruptedException {
try {
Channel channel = getChannel();
if(channel != null) {
JSONObject message = new JSONObject();
message.put("url", uri.toString());
channel.basicPublish(queueName, "url", null, message.toString().getBytes());
if (channel != null) {
JSONObject message = new JSONObject().put("url", curi.toString());
BasicProperties props = new AMQP.BasicProperties.Builder().
contentType("application/json").build();
channel.basicPublish(getExchange(), getRoutingKey(), props,
message.toString().getBytes("UTF-8"));
if (logger.isLoggable(Level.FINE)) {
logger.fine("sent to amqp exchange=" + getExchange() + " routingKey=" + routingKey + ": " + message);
}
curi.setFetchStatus(S_SENT_TO_AMQP);
curi.getAnnotations().add(A_SENT_TO_AMQP);
// no further processing on this url
return ProcessResult.FINISH;
}
} catch (Exception e) {
logger.log(Level.SEVERE, "Attempting to send URI to AMQP server failed!", e);
logger.log(Level.SEVERE, "Attempting to send URI to AMQP server failed! " + curi, e);
}
};
return ProcessResult.PROCEED;
}
@Override
protected void innerProcess(CrawlURI uri) throws InterruptedException {
throw new RuntimeException("should never be called");
}
protected synchronized Channel getChannel() {
if (threadChannel.get() == null) {
modules/src/main/java/org/archive/modules/fetcher/FetchHTTPRequest.java
+8
View File
@@ -244,6 +244,14 @@ class FetchHTTPRequest {
logger.warning("Invalid accept header: " + headerString);
}
}
@SuppressWarnings("unchecked")
Map<String, String> uriCustomHeaders = (Map<String, String>) curi.getData().get("customHttpRequestHeaders");
if (uriCustomHeaders != null) {
for (Entry<String, String> h: uriCustomHeaders.entrySet()) {
request.addHeader(h.getKey(), h.getValue());
}
}
}
/**