From a64551296a044e1cd2e50606faa680050356e2ca Mon Sep 17 00:00:00 2001 From: Noah Levitt Date: Thu, 18 Feb 2016 18:15:22 -0800 Subject: [PATCH 1/4] new boolean forceFetch setting; REQUEST_HEADER_BLACKLIST to exclude certain http headers, since with https://github.com/internetarchive/umbra/pull/55 umbra sends over more stuff --- .../crawler/frontier/AMQPUrlReceiver.java | 23 +++++++++++++++---- 1 file changed, 19 insertions(+), 4 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 880d770f..816e2053 100644 --- a/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java +++ b/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java @@ -21,6 +21,7 @@ package org.archive.crawler.frontier; import java.io.IOException; import java.io.UnsupportedEncodingException; +import java.util.Arrays; import java.util.HashMap; import java.util.HashSet; import java.util.Map; @@ -131,6 +132,14 @@ public class AMQPUrlReceiver implements Lifecycle, ApplicationListener REQUEST_HEADER_BLACKLIST = new HashSet(Arrays.asList( + "accept-encoding", "upgrade-insecure-requests", "host", "connection")); // XXX should we be using QueueingConsumer because of possible blocking in // frontier.schedule()? @@ -318,6 +330,7 @@ public class AMQPUrlReceiver implements Lifecycle, ApplicationListener customHttpRequestHeaders = new HashMap(); - for (Object key : joHeaders.keySet()) { - customHttpRequestHeaders.put(key.toString(), - joHeaders.getString(key.toString())); + for (Object key: joHeaders.keySet()) { + String k = key.toString(); + if (!k.startsWith(":") && !REQUEST_HEADER_BLACKLIST.contains(k)) { + customHttpRequestHeaders.put(k, joHeaders.getString(key.toString())); + } } curi.getData().put("customHttpRequestHeaders", customHttpRequestHeaders); @@ -392,7 +407,7 @@ public class AMQPUrlReceiver implements Lifecycle, ApplicationListener Date: Thu, 18 Feb 2016 18:16:03 -0800 Subject: [PATCH 2/4] set custom http headers last, so they override headers set elsewhere (e.g. cookies) --- .../modules/fetcher/FetchHTTPRequest.java | 25 ++++++++++++++----- 1 file changed, 19 insertions(+), 6 deletions(-) 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 ea4c047f..b4af96aa 100644 --- a/modules/src/main/java/org/archive/modules/fetcher/FetchHTTPRequest.java +++ b/modules/src/main/java/org/archive/modules/fetcher/FetchHTTPRequest.java @@ -53,6 +53,7 @@ import org.apache.http.HttpException; import org.apache.http.HttpHeaders; import org.apache.http.HttpHost; import org.apache.http.HttpRequest; +import org.apache.http.HttpRequestInterceptor; import org.apache.http.HttpResponse; import org.apache.http.NameValuePair; import org.apache.http.ProtocolVersion; @@ -347,13 +348,25 @@ public class FetchHTTPRequest { request.addHeader("X-Requested-With", "XMLHttpRequest"); } - @SuppressWarnings("unchecked") - Map uriCustomHeaders = (Map) curi.getData().get("customHttpRequestHeaders"); - if (uriCustomHeaders != null) { - for (Entry h: uriCustomHeaders.entrySet()) { - request.setHeader(h.getKey(), h.getValue()); + + /* + * set custom request headers in last interceptor, so they override + * anything else (this could just as well belong in + * configureHttpClientBuilder()) + */ + httpClientBuilder.addInterceptorLast(new HttpRequestInterceptor() { + @Override + public void process(HttpRequest request, HttpContext context) throws HttpException, IOException { + @SuppressWarnings("unchecked") + Map uriCustomHeaders = (Map) curi.getData().get("customHttpRequestHeaders"); + if (uriCustomHeaders != null) { + for (Entry h: uriCustomHeaders.entrySet()) { + request.setHeader(h.getKey(), h.getValue()); + } + } } - } + }); + } /** From bc9a6cc64053d70140bf82513d30d2af00aa05ec Mon Sep 17 00:00:00 2001 From: Noah Levitt Date: Thu, 18 Feb 2016 18:17:30 -0800 Subject: [PATCH 3/4] new parameter extraInfo, "Arbitrary additional information to include in the json payload", on AMQPPublishProcessor; used KeyedProperties so parameters can be overridden with the sheet/overlay system --- .../modules/AMQPProducerProcessor.java | 22 +++++++------- .../archive/modules/AMQPPublishProcessor.java | 30 +++++++++++++++---- .../postprocessor/AMQPCrawlLogFeed.java | 4 +-- 3 files changed, 38 insertions(+), 18 deletions(-) diff --git a/contrib/src/main/java/org/archive/modules/AMQPProducerProcessor.java b/contrib/src/main/java/org/archive/modules/AMQPProducerProcessor.java index 05faa758..b5b308d6 100644 --- a/contrib/src/main/java/org/archive/modules/AMQPProducerProcessor.java +++ b/contrib/src/main/java/org/archive/modules/AMQPProducerProcessor.java @@ -33,28 +33,28 @@ public abstract class AMQPProducerProcessor extends Processor { protected final Logger logger = Logger.getLogger(getClass().getName()); - protected String amqpUri = "amqp://guest:guest@localhost:5672/%2f"; + { + setAmqpUri("amqp://guest:guest@localhost:5672/%2f"); + } public String getAmqpUri() { - return this.amqpUri; + return (String) kp.get("amqpUri"); } public void setAmqpUri(String uri) { - this.amqpUri = uri; + kp.put("amqpUri", uri); } - protected String exchange; public String getExchange() { - return exchange; + return (String) kp.get("exchange"); } public void setExchange(String exchange) { - this.exchange = exchange; + kp.put("exchange", exchange); } - protected String routingKey; public String getRoutingKey() { - return routingKey; + return (String) kp.get("routingKey"); } public void setRoutingKey(String routingKey) { - this.routingKey = routingKey; + kp.put("routingKey", routingKey); } transient protected AMQPProducer amqpProducer; @@ -112,10 +112,10 @@ public abstract class AMQPProducerProcessor extends Processor { if (logger.isLoggable(Level.FINE)) { try { logger.fine("sent to amqp exchange=" + getExchange() - + " routingKey=" + routingKey + ": " + new String(message, "UTF-8")); + + " routingKey=" + getRoutingKey() + ": " + new String(message, "UTF-8")); } catch (UnsupportedEncodingException e) { logger.fine("sent to amqp exchange=" + getExchange() - + " routingKey=" + routingKey + ": " + message + " (" + message.length + " bytes)"); + + " routingKey=" + getRoutingKey() + ": " + message + " (" + message.length + " bytes)"); } } } diff --git a/contrib/src/main/java/org/archive/modules/AMQPPublishProcessor.java b/contrib/src/main/java/org/archive/modules/AMQPPublishProcessor.java index b4bdd469..d1f2ac29 100644 --- a/contrib/src/main/java/org/archive/modules/AMQPPublishProcessor.java +++ b/contrib/src/main/java/org/archive/modules/AMQPPublishProcessor.java @@ -24,6 +24,7 @@ import static org.archive.modules.CoreAttributeConstants.A_HERITABLE_KEYS; import java.io.Serializable; import java.io.UnsupportedEncodingException; import java.util.HashMap; +import java.util.Map; import java.util.Set; import org.apache.commons.httpclient.URIException; @@ -46,13 +47,15 @@ public class AMQPPublishProcessor extends AMQPProducerProcessor implements Seria public AMQPPublishProcessor() { // set default values - exchange = "umbra"; - routingKey = "urls"; + setExchange("umbra"); + setRoutingKey("urls"); } - protected String clientId = "requests"; + { + setClientId("requests"); + } public String getClientId() { - return clientId; + return (String) kp.get("clientId"); } /** * Client id to include in the json payload. AMQPUrlReceiver queueName @@ -60,7 +63,18 @@ public class AMQPPublishProcessor extends AMQPProducerProcessor implements Seria * this key. */ public void setClientId(String clientId) { - this.clientId = clientId; + kp.put("clientId", clientId); + } + + @SuppressWarnings("unchecked") + public Map getExtraInfo() { + return (Map) kp.get("extraInfo"); + } + /** + * Arbitrary additional information to include in the json payload. + */ + public void setExtraInfo(Map extraInfo) { + kp.put("extraInfo", extraInfo); } /** @@ -94,6 +108,12 @@ public class AMQPPublishProcessor extends AMQPProducerProcessor implements Seria message.put("clientId", getClientId()); } + if (getExtraInfo() != null) { + for (String k: getExtraInfo().keySet()) { + message.put(k, getExtraInfo().get(k)); + } + } + HashMap metadata = new HashMap(); metadata.put("pathFromSeed", curi.getPathFromSeed()); diff --git a/contrib/src/main/java/org/archive/modules/postprocessor/AMQPCrawlLogFeed.java b/contrib/src/main/java/org/archive/modules/postprocessor/AMQPCrawlLogFeed.java index 77f45cd2..2448ca08 100644 --- a/contrib/src/main/java/org/archive/modules/postprocessor/AMQPCrawlLogFeed.java +++ b/contrib/src/main/java/org/archive/modules/postprocessor/AMQPCrawlLogFeed.java @@ -86,8 +86,8 @@ public class AMQPCrawlLogFeed extends AMQPProducerProcessor implements Lifecycle public AMQPCrawlLogFeed() { // set default values - exchange = "heritrix.realTimeFeed"; - routingKey = "crawlLog"; + setExchange("heritrix.realTimeFeed"); + setRoutingKey("crawlLog"); } @Override From 9bdafc3c33bf1e4e76b37cc5c2769ed512672207 Mon Sep 17 00:00:00 2001 From: Noah Levitt Date: Thu, 18 Feb 2016 18:19:44 -0800 Subject: [PATCH 4/4] remove stray line --- .../main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java | 1 - 1 file changed, 1 deletion(-) 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 816e2053..859d716d 100644 --- a/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java +++ b/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java @@ -330,7 +330,6 @@ public class AMQPUrlReceiver implements Lifecycle, ApplicationListener