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

Merge pull request #150 from nlevitt/ari-4638

Browse Source 
various changes to amqp publish and receive
This commit is contained in:
Adam Miller
2016-03-08 09:26:58 -08:00
parent 0dc95ab7f7 9bdafc3c33
commit 7f752d8d45
5 changed files with 75 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java
+18 -4
View File
@@ -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<CrawlStat
public void setAutoDelete(boolean autoDelete) {
this.autoDelete = autoDelete;
}
private boolean forceFetch = false;
public boolean isForceFetch() {
return forceFetch;
}
public void setForceFetch(boolean forceFetch) {
this.forceFetch = forceFetch;
}
private transient Lock lock = new ReentrantLock(true);
@@ -291,6 +300,9 @@ public class AMQPUrlReceiver implements Lifecycle, ApplicationListener<CrawlStat
lock.unlock();
}
}
protected static final Set<String> REQUEST_HEADER_BLACKLIST = new HashSet<String>(Arrays.asList(
"accept-encoding", "upgrade-insecure-requests", "host", "connection"));
// XXX should we be using QueueingConsumer because of possible blocking in
// frontier.schedule()?
@@ -377,9 +389,11 @@ public class AMQPUrlReceiver implements Lifecycle, ApplicationListener<CrawlStat
// set the http headers from the amqp message
Map<String, String> customHttpRequestHeaders = new HashMap<String, String>();
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 +406,7 @@ public class AMQPUrlReceiver implements Lifecycle, ApplicationListener<CrawlStat
curi.setSchedulingDirective(SchedulingConstants.HIGH);
curi.setPrecedence(1);
curi.setForceFetch(jo.optBoolean("forceFetch"));
curi.setForceFetch(forceFetch || jo.optBoolean("forceFetch"));
curi.setSeed(jo.optBoolean("isSeed"));
curi.getAnnotations().add(A_RECEIVED_FROM_AMQP);
contrib/src/main/java/org/archive/modules/AMQPProducerProcessor.java
+11 -11
View File
@@ -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)");
}
}
}
contrib/src/main/java/org/archive/modules/AMQPPublishProcessor.java
+25 -5
View File
@@ -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<String, Object> getExtraInfo() {
return (Map<String, Object>) kp.get("extraInfo");
}
/**
* Arbitrary additional information to include in the json payload.
*/
public void setExtraInfo(Map<String, Object> 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<String, Object> metadata = new HashMap<String,Object>();
metadata.put("pathFromSeed", curi.getPathFromSeed());
contrib/src/main/java/org/archive/modules/postprocessor/AMQPCrawlLogFeed.java
+2 -2
View File
@@ -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
modules/src/main/java/org/archive/modules/fetcher/FetchHTTPRequest.java
+19 -6
View File
@@ -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<String, String> uriCustomHeaders = (Map<String, String>) curi.getData().get("customHttpRequestHeaders");
if (uriCustomHeaders != null) {
for (Entry<String, String> 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<String, String> uriCustomHeaders = (Map<String, String>) curi.getData().get("customHttpRequestHeaders");
if (uriCustomHeaders != null) {
for (Entry<String, String> h: uriCustomHeaders.entrySet()) {
request.setHeader(h.getKey(), h.getValue());
}
}
}
}
});
}
/**