package eu.socialsensor.focused.crawler.bolts.webpages;

import java.util.Map;

import org.apache.log4j.Logger;

import eu.socialsensor.framework.client.search.solr.SolrWebPageHandler;
import eu.socialsensor.framework.common.domain.WebPage;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Tuple;


public class TextIndexerBolt extends BaseRichBolt {

	/**
	 * 
	 */
	private static final long serialVersionUID = -7500656732029697927L;
	
	private Logger logger;
	
	private String _indexService;
	private SolrWebPageHandler solrWebPageHandler = null;
	
	public TextIndexerBolt(String indexService) {
		this._indexService = indexService;
	}
	
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
    	
    }

	public void prepare(@SuppressWarnings("rawtypes") Map conf, TopologyContext context, OutputCollector collector) {
		logger = Logger.getLogger(TextIndexerBolt.class);
		solrWebPageHandler = SolrWebPageHandler.getInstance(_indexService);
	}

	public void execute(Tuple tuple) {
		try {
			WebPage webPage = (WebPage) tuple.getValueByField("WebPage");
			if(webPage != null && solrWebPageHandler != null) {
				solrWebPageHandler.insertWebPage(webPage);
			}
		}
		catch(Exception ex) {
			logger.error(ex);
		}
	}

}
@@ -67,19 +67,118 @@
 va.util.
-Map
+ArrayList;%0Aimport java.util.List;%0Aimport java.util.Map;%0Aimport java.util.concurrent.ArrayBlockingQueue
 ;%0A%0Aimpor
@@ -814,16 +814,62 @@
 null;%0A%09%0A
+%09private ArrayBlockingQueue%3CWebPage%3E queue;%0A%09%0A
 %09public 
@@ -1198,16 +1198,68 @@
 class);%0A
+%09%09%0A%09%09queue = new ArrayBlockingQueue%3CWebPage%3E(5000);%0A
 %09%09solrWe
@@ -1320,16 +1320,76 @@
 rvice);%0A
+%09%09%0A%09%09Thread t = new Thread(new TextIndexer());%0A%09%09t.start();%0A
 %09%7D%0A%0A%09pub
@@ -1493,16 +1493,24 @@
 Page%22);%0A
+%09%09%09%0A%09%09%09%0A
 %09%09%09if(we
@@ -1564,40 +1564,17 @@
 %09%09%09%09
-solrWebPageHandler.insertWebPage
+queue.add
 (web
@@ -1642,9 +1642,565 @@
 %09%09%7D%0A%09%7D%0A%0A
+%09public class TextIndexer implements Runnable %7B%0A%0A%09%09public void run() %7B%0A%09%09%09while(true) %7B%0A%09%09%09%09try %7B%0A%09%09%09%09%09Thread.sleep(60 * 1000);%0A%0A%09%09%09%09%09List%3CWebPage%3E webPages = new ArrayList%3CWebPage%3E();%0A%09%09%09%09%09queue.drainTo(webPages);%0A%09%09%09%09%09%0A%09%09%09%09%09boolean inserted = solrWebPageHandler.insertWebPages(webPages);%0A%09%09%09%09%09%0A%09%09%09%09%09if(inserted) %7B%0A%09%09%09%09%09%09logger.info(webPages.size() + %22 web pages indexed in Solr%22);%0A%09%09%09%09%09%7D%0A%09%09%09%09%09else %7B%0A%09%09%09%09%09%09logger.error(%22Indexing in Solr failed for web pages%22);%0A%09%09%09%09%09%7D%0A%09%09%09%09%7D catch (Exception e) %7B%0A%09%09%09%09%09logger.error(e);%0A%09%09%09%09%09continue;%0A%09%09%09%09%7D%0A%09%09%09%7D%0A%09%09%7D%0A%09%09%0A%09%7D%0A
 %7D

Fix web pages indexer