From bc9a6cc64053d70140bf82513d30d2af00aa05ec Mon Sep 17 00:00:00 2001 From: Noah Levitt Date: Thu, 18 Feb 2016 18:17:30 -0800 Subject: [PATCH] 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