class AMQPPublishProcessor extends AMQPProducerProcessor implements Serializable, ApplicationContextAware {

    private static final long serialVersionUID = 2L;

    public static final String A_SENT_TO_AMQP = "sentToAMQP"; // annotation

    protected ApplicationContext appCtx;
    public void setApplicationContext(ApplicationContext appCtx) throws BeansException {
        this.appCtx = appCtx;
    }

    public AMQPPublishProcessor() {
        // set default values
        setExchange("umbra");
        setRoutingKey("urls");
    }

    {
        setClientId("requests");
    }
    public String getClientId() {
        return (String) kp.get("clientId");
    }
    /**
     * Client id to include in the json payload. AMQPUrlReceiver queueName
     * should have the same value, since umbra will route request urls based on
     * this key.
     */
    public void setClientId(String 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);
    }

    /**
     * @return true iff url is http or https, is not robots.txt, was not
     *         received via AMQP
     */
    protected boolean shouldProcess(CrawlURI curi) {
        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);
        }
    }

    /**
     * Constructs the json to send via AMQP. This includes the url, and some
     * metadata from the CrawlURI. The metadata should be passed back to
     * heritrix with each url discovered from this url. (XXX need context in
     * class javadoc)
     * 
     * @return the message to send via AMQP
     * @see CrawlURI#inheritFrom(CrawlURI)
     */
    protected JSONObject buildJsonMessage(CrawlURI curi) {<FILL_FUNCTION_BODY>}

    @Override
    protected byte[] buildMessage(CrawlURI curi) {
        try {
            return buildJsonMessage(curi).toString().getBytes("UTF-8");
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    protected void success(CrawlURI curi, byte[] message, BasicProperties props) {
        super.success(curi, message, props);
        curi.getAnnotations().add(A_SENT_TO_AMQP);
        appCtx.publishEvent(new AMQPUrlPublishedEvent(AMQPPublishProcessor.this, curi));
    }

    protected BasicProperties props = new AMQP.BasicProperties.Builder().
            contentType("application/json").build();

    @Override
    protected BasicProperties amqpMessageProperties() {
        return props;
    }
}

        JSONObject message = new JSONObject().put("url", curi.toString());

        if (getClientId() != null) {
            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());

        @SuppressWarnings("unchecked")
        Set<String> heritableKeys = (Set<String>) curi.getData().get(A_HERITABLE_KEYS);
        HashMap<String, Object> heritableData = new HashMap<String,Object>();
        if (heritableKeys != null) {
            for (String key: heritableKeys) {
                heritableData.put(key, curi.getData().get(key));
            }
        }
        metadata.put("heritableData", heritableData);

        message.put("metadata", metadata);

        return message;
    
<methods>public non-sealed void <init>() ,public java.lang.String getAmqpUri() ,public java.lang.String getExchange() ,public java.lang.String getRoutingKey() ,public void setAmqpUri(java.lang.String) ,public void setExchange(java.lang.String) ,public void setRoutingKey(java.lang.String) ,public synchronized void stop() <variables>protected transient org.archive.modules.AMQPProducer amqpProducer,protected final java.util.logging.Logger logger