

package eu.socialsensor.framework.streams;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;

import eu.socialsensor.framework.common.domain.Feed;
import eu.socialsensor.framework.common.domain.Item;
import eu.socialsensor.framework.common.domain.dysco.Dysco;
import eu.socialsensor.framework.monitors.FeedsMonitor;
import eu.socialsensor.framework.retrievers.Retriever;



/**
 * Class handles the stream of information regarding a social network.
 * It is responsible for its configuration, its wrapper's initialization
 * and its retrieval process.
 * @author manosetro
 * @email  manosetro@iti.gr
 * @author ailiakop
 * @email  ailiakop@iti.gr
 *
 */
public abstract class Stream implements Runnable {

	protected static final String KEY = "Key";
	protected static final String SECRET = "Secret";
	protected static final String ACCESS_TOKEN = "AccessToken";
	protected static final String ACCESS_TOKEN_SECRET = "AccessTokenSecret";
	protected static final String CLIENT_ID = "ClientId";
	protected static final String MAX_RESULTS = "maxResults";
	protected static final String MAX_REQUESTS = "maxRequests";
	
	protected FeedsMonitor monitor;
	protected BlockingQueue<Feed> feedsQueue;
	protected Retriever retriever;
	protected StreamHandler handler;
	
	/**
	 * Open a stream for updates delivery
	 * @param config
	 *      Stream configuration parameters
	 * @throws StreamException
	 *      In any case of error during stream open
	 */
	public abstract void open(StreamConfiguration config) throws StreamException;
	
	/**
	 * Close a stream 
	 * @throws StreamException
	 *      In any case of error during stream close
	 */
	public abstract void close() throws StreamException;
		
	/**
	 * Search for a certain dysco
	 * @param dysco
	 * @throws StreamException
	 */
	public void search(Dysco dysco) throws StreamException {
		
	}
	
	/**
	 * Set the handler that is responsible for the handling 
	 * of the retrieved items
	 * @param handler
	 */
	public void setHandler(StreamHandler handler){
		this.handler = handler;
	}
	
	/**
	 * Sets the feeds monitor for the stream
	 * @return
	 */
	public boolean setMonitor(){
		if(retriever == null)
			return false;
		
		monitor = new FeedsMonitor(retriever);
		return true;
	}
	
	/**
	 * Searches with the wrapper of the stream for a particular
	 * set of feeds (feeds can be keywordsFeeds, userFeeds or locationFeeds)
	 * @param feeds
	 * @return the total number of retrieved items for the stream
	 * @throws StreamException
	 */
	public Integer search(List<Feed> feeds) throws StreamException {
		List<Item> items = new ArrayList<Item>();
		
		if(retriever != null) {
		
			if(feeds == null)
				return null ;
				
			for(Feed feed : feeds){
				
				List<Item> retrievedItems = retriever.retrieve(feed);
				items.addAll(retrievedItems);
				
			}
			
			if(items != null){
				store(items);
			}	
			
			System.out.println("Retrieved items for "+this.getClass().getName()+ "are : "+items.size());
		}
		
		return items.size();
	
	}
	
	/**
	 * Store a set of items in the selected databases
	 * @param items
	 */
	public void store(List<Item>items) {
		for(Item item : items) {
			handler.update(item);
		}
	}
	
	/**
	 * Store an item in the selected databases
	 * @param item
	 */
	public void store(Item item) {
		if(handler == null){
			System.out.println("NULL Handler!");
			return;
		}
			
		handler.update(item);
	}
	
	/**
	 * Deletes an item from the selected databases
	 * @param item
	 */
	public void delete(Item item){
		handler.delete(item);
	}
	
	/**
	 * Adds a feed to the stream for future searching
	 * @param feed
	 * @return
	 */
	public boolean addFeed(Feed feed) {
		if(feedsQueue == null)
			return false;
		
		return feedsQueue.offer(feed);
	}
	
	@Override
	public void run() {
		while(true) {
			try {
				Feed feed = feedsQueue.take();
				monitor.addFeed(feed);
				monitor.startMonitor(feed);
				
			} catch (InterruptedException e) {
				return;
			}
		}
	}
}

@@ -300,16 +300,83 @@
 .Dysco;%0A
+import eu.socialsensor.framework.common.domain.feeds.KeywordsFeed;%0A
 import e
@@ -3039,16 +3039,17 @@
 ame()+ %22
+ 
 are : %22+
@@ -3063,16 +3063,16 @@
 ize());%0A
-
 %09%09%7D%0A%09%09%0A%09
@@ -3178,32 +3178,45 @@
 ms%0A%09 */%0A%09public 
+synchronized 
 void store(List%3C
@@ -3259,29 +3259,20 @@
 s) %7B%0A%09%09%09
-handler.updat
+stor
 e(item);
@@ -3347,32 +3347,32 @@
 m item%0A%09 */%0A
-
 %09public 
 void store(I
@@ -3355,24 +3355,37 @@
  */%0A%09public 
+synchronized 
 void store(I

swap from source name to source id