package com.main;

import java.net.UnknownHostException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Random;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import com.mongodb.BasicDBObject;
import com.mongodb.DB;
import com.mongodb.DBCollection;
import com.mongodb.DBObject;
import com.mongodb.MapReduceCommand;
import com.mongodb.Mongo;
import com.mongodb.MongoException;

public class MapReduceTest {

	private Mongo mongo = null;
	private DB db = null;

	@Before
	public void init() {
		try {
			mongo = new Mongo("localhost");
			db = mongo.getDB("stat");
			initData();
		} catch (UnknownHostException e) {
			e.printStackTrace();
		} catch (MongoException e) {
			e.printStackTrace();
		}
	}

	/**
	 * map reduce for today
	 */
	@Test
	public void dayMapReduce() {
		String map = " function() {emit(this.videoid,this.play)}";
		String reduce = "function(key,values) {var count = 0;values.forEach(function(value){count += value;});return count;}";
		Calendar calendar = Calendar.getInstance();
		String datePattern = "yyyy-MM";
		SimpleDateFormat sdf = new SimpleDateFormat(datePattern);
		String monthOfYear = sdf.format(calendar.getTime());
		for (int day = 1; day <= 31; day++) {
			String collectionName = monthOfYear + "-" + day;
			DBCollection collection = db.getCollection(collectionName);
			collection
					.mapReduce(map, reduce, collectionName + "_daystat", null);
		}
	}

	/**
	 * map reduce for current month
	 */
	@Test
	public void monthMapReduce() {
		// do day mapreduce before month mapreduce
		dayMapReduce();
		String map = " function() {emit(this._id,this.value)}";
		String reduce = "function(key,values) {var count = 0;values.forEach(function(value){count += value;});return count;}";
		Calendar calendar = Calendar.getInstance();
		String datePattern = "yyyy-MM";
		SimpleDateFormat sdf = new SimpleDateFormat(datePattern);
		String monthOfYear = sdf.format(calendar.getTime());
		for (int day = 1; day <= 31; day++) {
			String collectionName = monthOfYear + "-" + day + "_daystat";
			DBCollection collection = db.getCollection(collectionName);
			MapReduceCommand cmd = new MapReduceCommand(collection, map,
					reduce, monthOfYear + "_monthstat",
					MapReduceCommand.OutputType.REDUCE, null);
			collection.mapReduce(cmd);
		}
	}

	private void initData() {
		db.dropDatabase();
		Calendar calendar = Calendar.getInstance();
		String datePattern = "yyyy-MM";
		SimpleDateFormat sdf = new SimpleDateFormat(datePattern);
		String monthOfYear = sdf.format(calendar.getTime());
		for (int day = 1; day <= 31; day++) {
			String collectionName = monthOfYear + "-" + day;
			DBCollection collection = db.getCollection(collectionName);
			for (int j = 1; j <= 100; j++) {
				DBObject play = new BasicDBObject();
				play.put("videoid", new Random().nextInt(100));
				play.put("play", new Random().nextInt(1000));
				collection.insert(play);
			}
		}
	}

	@After
	public void destory() {
		if (mongo != null)
			mongo.close();
		mongo = null;
		db = null;
	}
}