package org.isep.maw;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;
import java.util.PriorityQueue;
import java.util.Queue;
import java.util.Scanner;
import java.util.StringTokenizer;

import com.google.common.io.Files;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

public class CSVTweetSpout extends BaseRichSpout {
	
	public static int MAX_QUEUE_SZ= 200;
	public static int THRESOLD = 100;
	
	
	
	private static final long serialVersionUID = 1L;
	
	private SpoutOutputCollector _collector;
	private List<BufferedReader> scanList = new ArrayList<BufferedReader>();
	
	private final List<String> fileList;
	private final Queue<Tweet> recQ = new PriorityQueue<Tweet>(MAX_QUEUE_SZ);
	private final String separator;
	
	
	public CSVTweetSpout(String separator, List<String> fileList) throws FileNotFoundException {
		this.fileList = fileList;
		this.separator = separator;
	}
	
	
	private void init(List<String> fileList) throws FileNotFoundException {
		for(String fileName: fileList) {
			File f  = new File(fileName);
			
			if(f.exists()) {
				System.out.println("OPENING: " + fileName);
				BufferedReader br = Files.newReader(f, Charset.forName("UTF-8"));
				scanList.add(br);
			}
		}
	}
	@Override
	public void nextTuple() {
		try {
			Thread.sleep(50);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		//Refill the list
		if(recQ.size() < THRESOLD ) { // Read from scanners.
			
			while(recQ.size() < MAX_QUEUE_SZ && scanList.size() > 0) {
				ListIterator<BufferedReader> it = scanList.listIterator();
				
				while(it.hasNext()) {
					
					BufferedReader br = it.next();
					
					String tweetStr = null;
					try {
						tweetStr = br.readLine();
					} catch (IOException e1) {
						// TODO Auto-generated catch block
						e1.printStackTrace();
					}
					if(tweetStr !=null) {
						Tweet t = scanTweet(tweetStr);
						if(t != null) recQ.offer(t);
						
					}else {
						System.out.println("Nothing to read");
						try {
							br.close();
						} catch (IOException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
						}
						it.remove();
					}
				}
			}
		}
		
		// emit next tweet
		Tweet tweet = recQ.poll();
		if(tweet != null ) {
			//System.out.println(tweet.getId() + " " + tweet.getText() );
			_collector.emit(new Values(tweet.getId(),tweet.getCreationDate(), tweet.getText()));
		}
	}

	private Tweet scanTweet(String  line) {
		
		System.out.println(line);
		StringTokenizer st = new StringTokenizer(line,separator,false);
		Tweet t = null;
		try {
			
			//System.out.println("==== TEST ==== " + st.nextToken());
		 t = new Tweet(Long.parseLong(st.nextToken()), Long.parseLong(st.nextToken()), st.nextToken());
		} catch (Exception e) {
			System.out.println(e.toString() + " " + e.getMessage());
		}
		return t;
	}


	@Override
	public void open(Map arg0, TopologyContext arg1, SpoutOutputCollector arg2) {
		// TODO Auto-generated method stub
		this._collector = arg2;
		try {
			init(fileList);
		} catch (FileNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer arg0) {
		// TODO Auto-generated method stub
		arg0.declare(new Fields("id","ts","text"));
		
	}

}

@@ -167,16 +167,99 @@
 harset;%0A
+import java.nio.file.Files;%0Aimport java.nio.file.Path;%0Aimport java.nio.file.Paths;%0A
 import j
@@ -414,34 +414,8 @@
 ue;%0A
-import java.util.Scanner;%0A
 impo
@@ -456,51 +456,17 @@
 ort 
-com.google.common.io.Files;%0A%0Aimport backtyp
+org.apach
 e.st
@@ -496,39 +496,41 @@
 llector;%0Aimport 
-backtyp
+org.apach
 e.storm.task.Top
@@ -542,39 +542,41 @@
 Context;%0Aimport 
-backtyp
+org.apach
 e.storm.topology
@@ -597,39 +597,41 @@
 eclarer;%0Aimport 
-backtyp
+org.apach
 e.storm.topology
@@ -650,39 +650,41 @@
 chSpout;%0Aimport 
-backtyp
+org.apach
 e.storm.tuple.Fi
@@ -700,15 +700,17 @@
 ort 
-backtyp
+org.apach
 e.st
@@ -728,16 +728,17 @@
 alues;%0A%0A
+%0A
 public c
@@ -1455,26 +1455,27 @@
 %0A%09%09%09
-File
+Path
  f  = 
-new File
+Paths.get
 (fil
@@ -1494,16 +1494,25 @@
 %09%09%09if(f.
+toFile().
 exists()
@@ -1586,16 +1586,42 @@
 der br =
+ null;%0A%09%09%09%09try %7B%0A%09%09%09%09%09br =
  Files.n
@@ -1622,16 +1622,24 @@
 iles.new
+Buffered
 Reader(f
@@ -1667,16 +1667,136 @@
 F-8%22));%0A
+%09%09%09%09%7D catch (IOException e) %7B%0A%09%09%09%09%09// TODO Auto-generated catch block%0A%09%09%09%09%09e.printStackTrace();%0A%09%09%09%09%7D%0A%09%09%09%09if(br!=null)%0A%09
 %09%09%09%09scan

Update to new Storm and NIO