diff --git a/commons/pom.xml b/commons/pom.xml index 06be676a..c894bb4f 100644 --- a/commons/pom.xml +++ b/commons/pom.xml @@ -162,11 +162,6 @@ spring-expression 3.0.5.RELEASE - - org.json - json - 20090211 - com.esotericsoftware diff --git a/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java b/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java new file mode 100644 index 00000000..36c29d23 --- /dev/null +++ b/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java @@ -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 customHttpRequestHeaders = new HashMap(); + 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; + } + } +} diff --git a/contrib/src/main/java/org/archive/modules/AMQPPublishProcessor.java b/contrib/src/main/java/org/archive/modules/AMQPPublishProcessor.java index e9035a8f..d78617f0 100644 --- a/contrib/src/main/java/org/archive/modules/AMQPPublishProcessor.java +++ b/contrib/src/main/java/org/archive/modules/AMQPPublishProcessor.java @@ -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 threadChannel = new ThreadLocal(); - 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) { diff --git a/modules/src/main/java/org/archive/modules/fetcher/FetchHTTPRequest.java b/modules/src/main/java/org/archive/modules/fetcher/FetchHTTPRequest.java index 8a93c203..7f48ae9f 100644 --- a/modules/src/main/java/org/archive/modules/fetcher/FetchHTTPRequest.java +++ b/modules/src/main/java/org/archive/modules/fetcher/FetchHTTPRequest.java @@ -244,6 +244,14 @@ class FetchHTTPRequest { logger.warning("Invalid accept header: " + headerString); } } + + @SuppressWarnings("unchecked") + Map uriCustomHeaders = (Map) curi.getData().get("customHttpRequestHeaders"); + if (uriCustomHeaders != null) { + for (Entry h: uriCustomHeaders.entrySet()) { + request.addHeader(h.getKey(), h.getValue()); + } + } } /**