try {
Protocol protocol = protocolFactory.getProtocol(new URL(
fit.url));
ProtocolResponse response = protocol
.getProtocolOutput(fit.url);
LOG.info("[Fetcher #" + taskIndex + "] Fetched " + fit.url
+ " with status " + response.getStatusCode());
eventCounter.scope("fetched").incrBy(1);
response.getMetadata().put(
"fetch.statusCode",
new String[] { Integer.toString(response
.getStatusCode()) });
// update the stats
// eventStats.scope("KB downloaded").update((long)
// content.length / 1024l);
// eventStats.scope("# pages").update(1);
if (fit.t.contains("metadata")) {
HashMap<String, String[]> metadata = (HashMap<String, String[]>) fit.t
.getValueByField("metadata");
if (metadata != null && !metadata.isEmpty()) {
for (Entry<String, String[]> entry : metadata
.entrySet())
response.getMetadata().put(entry.getKey(),
entry.getValue());
}
}
emitQueue.add(new Object[] {
Utils.DEFAULT_STREAM_ID,
fit.t,
new Values(fit.url, response.getContent(), response
.getMetadata()) });
synchronized (ackQueue) {
ackQueue.add(fit.t);
}