From 5e8219b014d5b7af4478c2c4245a6f62b4981be6 Mon Sep 17 00:00:00 2001 From: Noah Levitt Date: Fri, 31 Jan 2014 20:41:05 -0800 Subject: [PATCH 1/9] pull in json dependency from ia-web-commons --- commons/pom.xml | 5 ----- 1 file changed, 5 deletions(-) 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 From c163f318511e2f3a3c3e168595961947a95c16ab Mon Sep 17 00:00:00 2001 From: Noah Levitt Date: Fri, 31 Jan 2014 20:42:27 -0800 Subject: [PATCH 2/9] AMQPUrlReceiver - receive urls and request headers via amqp in the format produced by https://github.com/internetarchive/umbra and add to frontier --- .../crawler/frontier/AMQPUrlReceiver.java | 241 ++++++++++++++++++ 1 file changed, 241 insertions(+) create mode 100644 contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java 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..7ce2a66e --- /dev/null +++ b/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java @@ -0,0 +1,241 @@ +package org.archive.crawler.frontier; + +import java.io.IOException; +import java.io.UnsupportedEncodingException; +import java.util.HashMap; +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; + +public class AMQPUrlReceiver implements Lifecycle, Runnable { + + @SuppressWarnings("unused") + private static final long serialVersionUID = 1L; + + private static final Logger logger = + Logger.getLogger(AMQPUrlReceiver.class.getName()); + + 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; + } + + transient protected Thread fred = null; + + @Override + public void start() { + if (isRunning()) { + return; + } + + fred = new Thread(this, getClass().getSimpleName()); + fred.start(); + } + + @Override + public void stop() { + logger.info("shutting down"); + boolean joined = false; + while (!joined) { + fred.interrupt(); + try { + fred.join(); + joined = true; + } catch (InterruptedException e) { + } + } + } + + @Override + public boolean isRunning() { + return fred != null && fred.isAlive(); + } + + transient protected Connection connection = 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; + } + + transient protected Channel channel = null; + 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; + } + + protected class UrlConsumer extends DefaultConsumer { + public UrlConsumer(Channel channel) { + super(channel); + } + + @Override + public void handleDelivery(String consumerTag, + Envelope envelope, BasicProperties properties, + byte[] body) { +// logger.info("consumerTag=" + consumerTag +// + " envelope=" + envelope +// + " properties=" + properties +// + " body=" + body); + 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); + logger.info("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); + } + } + + // { + // "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); + + HashMap customHttpRequestHeaders = new HashMap(); + for (Object key: joHeaders.keySet()) { + customHttpRequestHeaders.put(key.toString(), joHeaders.getString(key.toString())); + } + curi.getData().put("customHttpRequestHeaders", customHttpRequestHeaders); + + // https://webarchive.jira.com/wiki/display/Heritrix/Precedence+Feature+Notes + // use HighestUriQueuePrecedencePolicy to ensure these high priority urls really get crawled ahead of others + curi.setSchedulingDirective(SchedulingConstants.HIGH); + curi.setPrecedence(1); + + return curi; + } + } + + @Override + public void run() { + logger.info(Thread.currentThread() + " starting"); + while (true) { + try { + if (Thread.interrupted()) { + throw new InterruptedException(); + } + + try { + Consumer consumer = new UrlConsumer(channel()); + channel().basicConsume(getQueueName(), false, consumer); + } catch (IOException e) { + logger.log(Level.SEVERE, "problem consuming AMQP (will try again after 10 seconds)", e); + Thread.sleep(10000); + } + + } catch (InterruptedException e) { + logger.info(Thread.currentThread() + " interrupted, shutting down"); + shutdown(); + return; + } + } + } + + protected void shutdown() { + if (connection != null && connection.isOpen()) { + try { + connection.close(); + } catch (IOException e) { + logger.log(Level.SEVERE, "problem closing AMQP connection", e); + } + } + } + + public static void main(String[] args) throws InterruptedException { + AMQPUrlReceiver x = new AMQPUrlReceiver(); + // x.setAmqpUri("amqp://guest:guest@desktop-nlevitt.sf.archive.org:5672/%2f"); + x.start(); + Thread.sleep(90000); + x.stop(); + } +} From bf0590a0830eb2cfbf616d516ad9d52e9e7a60bc Mon Sep 17 00:00:00 2001 From: Noah Levitt Date: Sat, 1 Feb 2014 18:23:05 -0800 Subject: [PATCH 3/9] spaces not tabs for indentation --- .../crawler/frontier/AMQPUrlReceiver.java | 392 +++++++++--------- 1 file changed, 200 insertions(+), 192 deletions(-) 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 7ce2a66e..896e6558 100644 --- a/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java +++ b/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java @@ -29,12 +29,12 @@ import com.rabbitmq.client.Envelope; public class AMQPUrlReceiver implements Lifecycle, Runnable { - @SuppressWarnings("unused") - private static final long serialVersionUID = 1L; + @SuppressWarnings("unused") + private static final long serialVersionUID = 1L; + + private static final Logger logger = + Logger.getLogger(AMQPUrlReceiver.class.getName()); - private static final Logger logger = - Logger.getLogger(AMQPUrlReceiver.class.getName()); - protected Frontier frontier; public Frontier getFrontier() { return this.frontier; @@ -44,198 +44,206 @@ public class AMQPUrlReceiver implements Lifecycle, Runnable { 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 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 String queueName = "requests"; + public String getQueueName() { + return queueName; + } + public void setQueueName(String queueName) { + this.queueName = queueName; + } - transient protected Thread fred = null; - - @Override - public void start() { - if (isRunning()) { - return; - } - - fred = new Thread(this, getClass().getSimpleName()); - fred.start(); - } + transient protected Thread fred = null; - @Override - public void stop() { - logger.info("shutting down"); - boolean joined = false; - while (!joined) { - fred.interrupt(); - try { - fred.join(); - joined = true; - } catch (InterruptedException e) { - } - } - } + @Override + public void start() { + if (isRunning()) { + return; + } - @Override - public boolean isRunning() { - return fred != null && fred.isAlive(); - } + fred = new Thread(this, getClass().getSimpleName()); + fred.start(); + } - transient protected Connection connection = 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; - } - - transient protected Channel channel = null; - 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; - } - - protected class UrlConsumer extends DefaultConsumer { - public UrlConsumer(Channel channel) { - super(channel); - } + @Override + public void stop() { + logger.info("shutting down"); + boolean joined = false; + while (!joined) { + fred.interrupt(); + try { + fred.join(); + joined = true; + } catch (InterruptedException e) { + } + } + } - @Override - public void handleDelivery(String consumerTag, - Envelope envelope, BasicProperties properties, - byte[] body) { -// logger.info("consumerTag=" + consumerTag -// + " envelope=" + envelope -// + " properties=" + properties -// + " body=" + body); - 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); - logger.info("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); - } - } - - // { - // "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); - - HashMap customHttpRequestHeaders = new HashMap(); - for (Object key: joHeaders.keySet()) { - customHttpRequestHeaders.put(key.toString(), joHeaders.getString(key.toString())); - } - curi.getData().put("customHttpRequestHeaders", customHttpRequestHeaders); - - // https://webarchive.jira.com/wiki/display/Heritrix/Precedence+Feature+Notes - // use HighestUriQueuePrecedencePolicy to ensure these high priority urls really get crawled ahead of others - curi.setSchedulingDirective(SchedulingConstants.HIGH); - curi.setPrecedence(1); - - return curi; - } - } - - @Override - public void run() { - logger.info(Thread.currentThread() + " starting"); - while (true) { - try { - if (Thread.interrupted()) { - throw new InterruptedException(); - } - - try { - Consumer consumer = new UrlConsumer(channel()); - channel().basicConsume(getQueueName(), false, consumer); - } catch (IOException e) { - logger.log(Level.SEVERE, "problem consuming AMQP (will try again after 10 seconds)", e); - Thread.sleep(10000); - } - - } catch (InterruptedException e) { - logger.info(Thread.currentThread() + " interrupted, shutting down"); - shutdown(); - return; - } - } - } + @Override + public boolean isRunning() { + return fred != null && fred.isAlive(); + } - protected void shutdown() { - if (connection != null && connection.isOpen()) { - try { - connection.close(); - } catch (IOException e) { - logger.log(Level.SEVERE, "problem closing AMQP connection", e); - } - } - } - - public static void main(String[] args) throws InterruptedException { - AMQPUrlReceiver x = new AMQPUrlReceiver(); - // x.setAmqpUri("amqp://guest:guest@desktop-nlevitt.sf.archive.org:5672/%2f"); - x.start(); - Thread.sleep(90000); - x.stop(); - } + 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; + } + + protected class UrlConsumer extends DefaultConsumer { + public UrlConsumer(Channel channel) { + super(channel); + } + + @Override + public void handleDelivery(String consumerTag, Envelope envelope, + BasicProperties properties, byte[] body) { + // logger.info("consumerTag=" + consumerTag + // + " envelope=" + envelope + // + " properties=" + properties + // + " body=" + body); + 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); + logger.info("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); + } + } + + // { + // "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); + + HashMap 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); + + return curi; + } + } + + @Override + public void run() { + logger.info(Thread.currentThread() + " starting"); + while (true) { + try { + if (Thread.interrupted()) { + throw new InterruptedException(); + } + + try { + Consumer consumer = new UrlConsumer(channel()); + channel().basicConsume(getQueueName(), false, consumer); + } catch (IOException e) { + logger.log(Level.SEVERE, "problem consuming AMQP (will try again after 10 seconds)", e); + Thread.sleep(10000); + } + + } catch (InterruptedException e) { + logger.info(Thread.currentThread() + " interrupted, shutting down"); + shutdown(); + return; + } + } + } + + protected void shutdown() { + if (connection != null && connection.isOpen()) { + try { + connection.close(); + } catch (IOException e) { + logger.log(Level.SEVERE, "problem closing AMQP connection", e); + } + } + } + + public static void main(String[] args) throws InterruptedException { + AMQPUrlReceiver x = new AMQPUrlReceiver(); + // x.setAmqpUri("amqp://guest:guest@desktop-nlevitt.sf.archive.org:5672/%2f"); + x.start(); + Thread.sleep(90000); + x.stop(); + } } From c7cf62bf807a821f7e914ddab0a0a98fae1e51c9 Mon Sep 17 00:00:00 2001 From: Noah Levitt Date: Sat, 1 Feb 2014 18:23:49 -0800 Subject: [PATCH 4/9] license header --- .../crawler/frontier/AMQPUrlReceiver.java | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) 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 896e6558..f1344331 100644 --- a/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java +++ b/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java @@ -1,3 +1,22 @@ +/* + * 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; From c980b452c93c0ce7a83b2fcf43b3a7d9fdcc21ec Mon Sep 17 00:00:00 2001 From: Noah Levitt Date: Sat, 1 Feb 2014 18:26:07 -0800 Subject: [PATCH 5/9] annotate CrawlURI with "receivedViaAMQP" --- .../main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java | 2 ++ 1 file changed, 2 insertions(+) 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 f1344331..e4b0305b 100644 --- a/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java +++ b/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java @@ -219,6 +219,8 @@ public class AMQPUrlReceiver implements Lifecycle, Runnable { curi.setSchedulingDirective(SchedulingConstants.HIGH); curi.setPrecedence(1); + curi.getAnnotations().add("receivedViaAMQP"); + return curi; } } From ecb0eaf4353cafca70c82693adc40ca569fe5790 Mon Sep 17 00:00:00 2001 From: Noah Levitt Date: Mon, 3 Feb 2014 18:36:52 -0800 Subject: [PATCH 6/9] AMQPUrlReceiver - ack amqp messages, get rid of misguided thread starting infinite consumers --- .../crawler/frontier/AMQPUrlReceiver.java | 122 +++++++----------- 1 file changed, 48 insertions(+), 74 deletions(-) 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 e4b0305b..e54f7425 100644 --- a/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java +++ b/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java @@ -22,6 +22,7 @@ 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; @@ -46,7 +47,10 @@ import com.rabbitmq.client.Consumer; import com.rabbitmq.client.DefaultConsumer; import com.rabbitmq.client.Envelope; -public class AMQPUrlReceiver implements Lifecycle, Runnable { +/** + * @contributor nlevitt + */ +public class AMQPUrlReceiver implements Lifecycle { @SuppressWarnings("unused") private static final long serialVersionUID = 1L; @@ -54,6 +58,8 @@ public class AMQPUrlReceiver implements Lifecycle, Runnable { 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; @@ -79,37 +85,42 @@ public class AMQPUrlReceiver implements Lifecycle, Runnable { this.queueName = queueName; } - transient protected Thread fred = null; + protected boolean isRunning = false; + @Override + public boolean isRunning() { + return isRunning; + } @Override public void start() { - if (isRunning()) { - return; + 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) { + } + } } - fred = new Thread(this, getClass().getSimpleName()); - fred.start(); } @Override public void stop() { logger.info("shutting down"); - boolean joined = false; - while (!joined) { - fred.interrupt(); + if (connection != null && connection.isOpen()) { try { - fred.join(); - joined = true; - } catch (InterruptedException e) { + connection.close(); + } catch (IOException e) { + logger.log(Level.SEVERE, "problem closing AMQP connection", e); } } } - @Override - public boolean isRunning() { - return fred != null && fred.isAlive(); - } - transient protected Connection connection = null; transient protected Channel channel = null; @@ -145,6 +156,12 @@ public class AMQPUrlReceiver implements Lifecycle, Runnable { 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); @@ -152,11 +169,7 @@ public class AMQPUrlReceiver implements Lifecycle, Runnable { @Override public void handleDelivery(String consumerTag, Envelope envelope, - BasicProperties properties, byte[] body) { - // logger.info("consumerTag=" + consumerTag - // + " envelope=" + envelope - // + " properties=" + properties - // + " body=" + body); + BasicProperties properties, byte[] body) throws IOException { String decodedBody; try { decodedBody = new String(body, "UTF-8"); @@ -169,7 +182,9 @@ public class AMQPUrlReceiver implements Lifecycle, Runnable { curi = makeCrawlUri(jo); // bypasses scoping (unless rechecking is configured) getFrontier().schedule(curi); - logger.info("scheduled " + 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 " @@ -179,16 +194,18 @@ public class AMQPUrlReceiver implements Lifecycle, Runnable { "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" + // "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 { @@ -205,7 +222,7 @@ public class AMQPUrlReceiver implements Lifecycle, Runnable { // XXX pathFromSeed? viaContext? CrawlURI curi = new CrawlURI(uuri, "?", via, LinkContext.INFERRED_MISC); - HashMap customHttpRequestHeaders = new HashMap(); + Map customHttpRequestHeaders = new HashMap(); for (Object key : joHeaders.keySet()) { customHttpRequestHeaders.put(key.toString(), joHeaders.getString(key.toString())); @@ -219,52 +236,9 @@ public class AMQPUrlReceiver implements Lifecycle, Runnable { curi.setSchedulingDirective(SchedulingConstants.HIGH); curi.setPrecedence(1); - curi.getAnnotations().add("receivedViaAMQP"); + curi.getAnnotations().add(A_RECEIVED_FROM_AMQP); return curi; } } - - @Override - public void run() { - logger.info(Thread.currentThread() + " starting"); - while (true) { - try { - if (Thread.interrupted()) { - throw new InterruptedException(); - } - - try { - Consumer consumer = new UrlConsumer(channel()); - channel().basicConsume(getQueueName(), false, consumer); - } catch (IOException e) { - logger.log(Level.SEVERE, "problem consuming AMQP (will try again after 10 seconds)", e); - Thread.sleep(10000); - } - - } catch (InterruptedException e) { - logger.info(Thread.currentThread() + " interrupted, shutting down"); - shutdown(); - return; - } - } - } - - protected void shutdown() { - if (connection != null && connection.isOpen()) { - try { - connection.close(); - } catch (IOException e) { - logger.log(Level.SEVERE, "problem closing AMQP connection", e); - } - } - } - - public static void main(String[] args) throws InterruptedException { - AMQPUrlReceiver x = new AMQPUrlReceiver(); - // x.setAmqpUri("amqp://guest:guest@desktop-nlevitt.sf.archive.org:5672/%2f"); - x.start(); - Thread.sleep(90000); - x.stop(); - } } From af44522fdf7f478ee128a2d11e69d9069715685b Mon Sep 17 00:00:00 2001 From: Noah Levitt Date: Mon, 3 Feb 2014 18:41:23 -0800 Subject: [PATCH 7/9] FetchHTTPRequest - include CrawlURI.getData().get("customHttpRequestHeaders"), if any, in http request --- .../org/archive/modules/fetcher/FetchHTTPRequest.java | 8 ++++++++ 1 file changed, 8 insertions(+) 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()); + } + } } /** From 500dafd3907b23e5d5f23597837855b92832f8d4 Mon Sep 17 00:00:00 2001 From: Noah Levitt Date: Mon, 3 Feb 2014 18:41:33 -0800 Subject: [PATCH 8/9] AMQPPublishProcessor - annotate urls sent to AMQP; set special fetch status; add default amqpUri; replace queueName setting with exchange and routingKey, which seem to be what are needed; avoid sending urls that were received from AMQP, and robots.txt urls; set content-type of amqp message to application/json, which umbra seems to need --- .../archive/modules/AMQPPublishProcessor.java | 99 ++++++++++++------- 1 file changed, 63 insertions(+), 36 deletions(-) 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) { From 6dbcac40d94ef9b8b28be22c1562230eb0844320 Mon Sep 17 00:00:00 2001 From: Noah Levitt Date: Mon, 3 Feb 2014 20:27:51 -0800 Subject: [PATCH 9/9] syncronize start() and stop() --- .../java/org/archive/crawler/frontier/AMQPUrlReceiver.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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 e54f7425..36c29d23 100644 --- a/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java +++ b/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java @@ -92,7 +92,7 @@ public class AMQPUrlReceiver implements Lifecycle { } @Override - public void start() { + synchronized public void start() { while (!isRunning) { try { Consumer consumer = new UrlConsumer(channel()); @@ -110,7 +110,7 @@ public class AMQPUrlReceiver implements Lifecycle { } @Override - public void stop() { + synchronized public void stop() { logger.info("shutting down"); if (connection != null && connection.isOpen()) { try {