mirror of
https://github.com/internetarchive/heritrix3.git
synced 2026-09-24 14:45:56 +00:00
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
This commit is contained in:
@@ -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)");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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());
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user