public class DateGenerator {
	public static long ONE_DAY      = 24L * 60L * 60L * 1000L;
	public static long SEVEN_DAYS   = 7L * ONE_DAY;
	public static long THIRTY_DAYS  = 30L * ONE_DAY;
	public static long ONE_YEAR     = 365L * ONE_DAY;
	public static long TWO_YEARS    = 2L * ONE_YEAR;
	public static long TEN_YEARS    = 10L * ONE_YEAR;
	public static long THIRTY_YEARS = 30L * ONE_YEAR;
	
	private long from;
	private long to;
	private long fromBirthDay;
	private long toBirthDay;
	GregorianCalendar birthCalendar;
    private long deltaTime;
	
	private PowerDistGenerator powerDist;
	
	// This constructor is for the case of friendship's created date generator
	public DateGenerator(GregorianCalendar from, GregorianCalendar to, 
			double alpha, long deltaTime )
	{
		this.from = from.getTimeInMillis();
		this.to = to.getTimeInMillis();
		powerDist = new PowerDistGenerator(0.0, 1.0, alpha );
        this.deltaTime = deltaTime;
		
		// For birthday from 1980 to 1990
		GregorianCalendar frombirthCalendar = new GregorianCalendar(1980,1,1);
		GregorianCalendar tobirthCalendar = new GregorianCalendar(1990,1,1);
		this.fromBirthDay = frombirthCalendar.getTimeInMillis();
		this.toBirthDay = tobirthCalendar.getTimeInMillis();
		this.birthCalendar = new GregorianCalendar();
	}
	
	/*
	 * Date between from and to
	 */
	public Long randomDateInMillis( Random random )
	{
		long date = (long)(random.nextDouble()*(to-from)+from);
		GregorianCalendar gc = new GregorianCalendar();
		gc.setTime(new Date(date));
		
		return gc.getTimeInMillis();
	}
	
	/*
	 * format the date
	 */
	public static String formatDate(GregorianCalendar c)
	{
		/*int day = c.get(Calendar.DAY_OF_MONTH);
		int month = c.get(Calendar.MONTH)+1;		// +1 because month can be 0
		int year = c.get(Calendar.YEAR);
		
		String prefixDay = "";
		String prefixMonth = "";		
		
		if(day<10)
			prefixDay = "0";
		
		if(month<10)
			prefixMonth = "0";
		
		return year+"-"+prefixMonth+month+"-"+prefixDay+day;
		*/

        c.setTimeZone(TimeZone.getTimeZone("GMT"));
        SimpleDateFormat gmtDateFormatter = new SimpleDateFormat("yyyy-MM-dd");
        gmtDateFormatter.setTimeZone(TimeZone.getTimeZone("GMT"));
        return gmtDateFormatter.format(c.getTime());
	}
	
	public static String formatYear(GregorianCalendar c)
	{
		int year = c.get(Calendar.YEAR);
		
		return year + "";
	}
	/*
	 * format the date with hours and minutes
	 */
	public static String formatDateDetail(GregorianCalendar c)
	{
	/*	int day = c.get(Calendar.DAY_OF_MONTH);
		int month = c.get(Calendar.MONTH)+1;
		int year = c.get(Calendar.YEAR);
		
		int hour = c.get(Calendar.HOUR_OF_DAY);
		int minute = c.get(Calendar.MINUTE);
		int second = c.get(Calendar.SECOND);
		
		String prefixDay = "";
		String prefixMonth = "";
		String prefixHour = "";
		String prefixMinute = "";
		String prefixSecond = "";
		
		if(day<10)
			prefixDay = "0";
		
		if(month<10)
			prefixMonth = "0";
		
		if (hour < 10)
			prefixHour = "0";
		
		if (minute < 10)
			prefixMinute = "0";

		if (second < 10)
			prefixSecond = "0";
		
		return year+"-"+prefixMonth+month+"-"+prefixDay+day +"T"
			   +prefixHour+hour+":"+prefixMinute+minute+":"+prefixSecond+second+"Z";
			   */
        c.setTimeZone(TimeZone.getTimeZone("GMT"));
        //SimpleDateFormat gmtDateFormatter = new SimpleDateFormat("yyyy-MM-ddHH:mm:ssZ");
        SimpleDateFormat gmtDateFormatter = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSSZ");
        gmtDateFormatter.setTimeZone(TimeZone.getTimeZone("GMT"));
        return gmtDateFormatter.format(c.getTime());
	}

	public static boolean isTravelSeason(long date){
		GregorianCalendar c = new GregorianCalendar();
		c.setTimeInMillis(date);
		
		int day = c.get(Calendar.DAY_OF_MONTH);
		int month = c.get(Calendar.MONTH)+1;
		
		if ((month > 5) && (month < 8)){
			return true; 
		} 
		else if ((month==12) &&  (day > 23)){
			return true; 
		}
		else 
			return false; 
	}
	
	/*
	 * Format date in xsd:dateTime format
	 */
	public static String formatDateTime(Long date) {
		GregorianCalendar c = new GregorianCalendar();
		c.setTimeInMillis(date);
		
		String dateString = formatDate(c);
		return dateString + "T00:00:00";
	}
	
	public static String formatDateTime(GregorianCalendar date) {
		String dateString = formatDate(date);
		return dateString + "T00:00:00";
	}
	
	public static int getNumberOfMonths(GregorianCalendar c, int startMonth, int startYear){
		int month = c.get(Calendar.MONTH)+1;
		int year = c.get(Calendar.YEAR);
		
		return (year-startYear)*12 + month-startMonth;
	}

	public Long randomDateInMillis(Random random, Long from, Long to)
	{
		long date = (long)(random.nextDouble()*(to-from)+from);
		GregorianCalendar gc = new GregorianCalendar();
		gc.setTime(new Date(date));
		
		return gc.getTimeInMillis();
	}
	
	public Long randomThirtyDaysSpan(Random random, Long from){
		long randomSpanMilis =  (long) (random.nextDouble()* (THIRTY_DAYS));
		return (from + randomSpanMilis);
	}

	public long randomFriendRequestedDate(Random random, ReducedUserProfile user1, ReducedUserProfile user2){
		long fromDate = Math.max(user1.getCreationDate(), user2.getCreationDate());
		return randomThirtyDaysSpan(random,fromDate);
	}
	
	public long randomFriendApprovedDate(Random random, long requestedDate){
		long randomSpanMilis =  (long) (random.nextDouble()* (SEVEN_DAYS));
		return (requestedDate + randomSpanMilis);
	}
	public long randomFriendDeclinedDate(Random random, long requestedDate){
		long randomSpanMilis =  (long) (random.nextDouble()* (SEVEN_DAYS));
		return (requestedDate + randomSpanMilis);
	}
	public long randomFriendReapprovedDate(Random random, long declined){
		long randomSpanMilis =  (long) (random.nextDouble()* (THIRTY_DAYS));
		return (declined + randomSpanMilis);
	}	
	public long numberOfMonths(ReducedUserProfile user){
		return (to - user.getCreationDate())/THIRTY_DAYS;
	}
	
	public long numberOfMonths(long fromDate){
		return (to - fromDate)/THIRTY_DAYS;
	}
	
    public long randomDate( Random random, long minDate ) {
        return  (long)(random.nextDouble()*(to-minDate)+minDate);
    }
	
    public long randomSevenDays(Random random ) {
        return (long)(random.nextDouble()*DateGenerator.SEVEN_DAYS);
    }
	
	public long powerlawCommDateDay(Random random, long lastCommentCreatedDate){
		return (long)(powerDist.getDouble(random) * ONE_DAY+lastCommentCreatedDate);
	}

	// The birthday is fixed during 1980 --> 1990
	public long getBirthDay(Random random, long userCreatedDate){
		return (long)(random.nextDouble()*(toBirthDay -fromBirthDay)+fromBirthDay);
	}
	
	public int getBirthYear(long birthDay){
		birthCalendar.setTimeInMillis(birthDay);
		return birthCalendar.get(GregorianCalendar.YEAR);
	}

	public int getBirthMonth(long birthDay){
		birthCalendar.setTimeInMillis(birthDay);
		return birthCalendar.get(GregorianCalendar.MONTH);
	}
	//If do not know the birthday, first randomly guess the age of user
	//Randomly get the age when user graduate
	//User's age for graduating is from 20 to 30

	public long getClassYear(Random random, long userCreatedDate, long birthday){
		long age;
		long graduateage = (random.nextInt(5) + 20) * ONE_YEAR; 
		if (birthday != -1){
			return (long)(birthday + graduateage); 
		}
		else{
			age = (long)(random.nextDouble() * THIRTY_YEARS + TEN_YEARS);
			return (userCreatedDate - age + graduateage);
		}
	}
	
	public long getWorkFromYear(Random random, long userCreatedDate, long birthday){
		long age;
		long workingage = (random.nextInt(10) + 25) * ONE_YEAR; 
		if (birthday != -1){
			return (long)(birthday + workingage); 
		}
		else{
			age = (long)(random.nextDouble() * THIRTY_YEARS + TEN_YEARS);
			return (userCreatedDate - age + workingage);
		}
	}
	
	public long getWorkFromYear(Random random, long classYear){
		return (classYear + (long)(random.nextDouble()*TWO_YEARS));
	}
	
	public long getStartDateTime(){
		return from;
	}
	public long getEndDateTime() {
		return to;
	}
    public long getMaxDateTime() {
        return to + SEVEN_DAYS + deltaTime;
    }
} public class ScalableGenerator{

    /** @brief A type of organization.**/
    public enum OrganisationType {
        university,
        company
    }


    Configuration  conf = null;
    public long postId  = 0;                             /**< @brief The post and comment identifier counter.*/
    public long groupId = 0;

    private static final int        NUM_FRIENDSHIP_HADOOP_JOBS = 3;                 /**< @brief The number of hadoop jobs used to generate the friendships.*/
    private static final double     friendRatioPerJob[] = { 0.45, 0.45, 0.1 };      /**< @brief The percentage of friendships generated in each hadoop job.*/
    private static final int        USER_RANDOM_ID_LIMIT = 100;                     /**< @brief The number of different random ids generated in the last hadoop job.*/

    private static final int startMonth = 1;
    private static final int startDate  = 1;
    private static final int endMonth   = 1;
    private static final int endDate    = 1;

    private static final double alpha = 0.4;        /**< @brief PowerLaw distribution alpha parameter.*/
    private static final int maxNumLikes = 1000;      /**< @brief The maximum number of likes per post*/

    private static final double levelProbs[] = { 0.5, 0.8, 1.0 };  /**< @brief Cumulative probability to join a group for the user direct friends, friends of friends and friends of the friends of the user friend.*/
    private static final double joinProbs[] = { 0.7, 0.4, 0.1 }; /**< @brief Probability to join a group for the user direct friends, friends of friends and friends of the friends of the user friend.*/

    //Files and folders
    private static final String  DICTIONARY_DIRECTORY = "/dictionaries/";
    private static final String  IPZONE_DIRECTORY     = "/ipaddrByCountries";
    private static final String  CSV_DIRECTORY        = "/csvSerializers";
    private static final String  PARAMETERS_FILE      = "params.ini";
    private static final String  SCALE_FACTORS_FILE   = "scale_factors.xml";
    private static final String  STATS_FILE           = "testdata.json";
    private static final String  RDF_OUTPUT_FILE      = "ldbc_socialnet_dbg";
    private static final String  PARAM_COUNT_FILE     = "factors.txt";

    // Dictionaries dataset files
    private static final String   browserDictonryFile       = DICTIONARY_DIRECTORY + "browsersDic.txt";
    private static final String   companiesDictionaryFile   = DICTIONARY_DIRECTORY + "companiesByCountry.txt";
    private static final String   countryAbbrMappingFile    = DICTIONARY_DIRECTORY + "countryAbbrMapping.txt";
    private static final String   tagDictionaryFile         = DICTIONARY_DIRECTORY + "dicCelebritiesByCountry.txt";
    private static final String   countryDictionaryFile     = DICTIONARY_DIRECTORY + "dicLocations.txt";
    private static final String   tagNamesFile              = DICTIONARY_DIRECTORY + "dicTopic.txt";
    private static final String   emailDictionaryFile       = DICTIONARY_DIRECTORY + "email.txt";
    private static final String   nameDictionaryFile        = DICTIONARY_DIRECTORY + "givennameByCountryBirthPlace.txt.freq.full";
    private static final String   universityDictionaryFile  = DICTIONARY_DIRECTORY + "institutesCityByCountry.txt";
    private static final String   cityDictionaryFile        = DICTIONARY_DIRECTORY + "citiesByCountry.txt";
    private static final String   languageDictionaryFile    = DICTIONARY_DIRECTORY + "languagesByCountry.txt";
    private static final String   popularDictionaryFile     = DICTIONARY_DIRECTORY + "popularPlacesByCountry.txt";
    private static final String   agentFile                 = DICTIONARY_DIRECTORY + "smartPhonesProviders.txt";
    private static final String   surnamDictionaryFile      = DICTIONARY_DIRECTORY + "surnameByCountryBirthPlace.txt.freq.sort";
    private static final String   tagClassFile              = DICTIONARY_DIRECTORY + "tagClasses.txt";
    private static final String   tagHierarchyFile          = DICTIONARY_DIRECTORY + "tagHierarchy.txt";
    private static final String   tagTextFile               = DICTIONARY_DIRECTORY + "tagText.txt";
    private static final String   tagTopicDictionaryFile    = DICTIONARY_DIRECTORY + "topicMatrixId.txt";
    private static final String   flashmobDistFile          = DICTIONARY_DIRECTORY + "flashmobDist.txt";
    private static final String   fbSocialDegreeFile	    = DICTIONARY_DIRECTORY + "facebookBucket100.dat";

    //private parameters
    private final String CELL_SIZE                     = "cellSize";
    private final String NUM_CELL_WINDOW               = "numberOfCellPerWindow";
    private final String MIN_FRIENDS                   = "minNumFriends";
    private final String MAX_FRIENDS                   = "maxNumFriends";
    private final String FRIEND_REJECT                 = "friendRejectRatio";
    private final String FRIEND_REACCEPT               = "friendReApproveRatio";
    private final String USER_MIN_TAGS                 = "minNumTagsPerUser";
    private final String USER_MAX_TAGS                 = "maxNumTagsPerUser";
    private final String USER_MAX_POST_MONTH           = "maxNumPostPerMonth";
    private final String MAX_COMMENT_POST              = "maxNumComments";
    private final String LIMIT_CORRELATED              = "limitProCorrelated";
    private final String BASE_CORRELATED               = "baseProbCorrelated";
    private final String MAX_EMAIL                     = "maxEmails";
    private final String MAX_COMPANIES                 = "maxCompanies";
    private final String ENGLISH_RATIO                 = "probEnglish";
    private final String SECOND_LANGUAGE_RATIO         = "probSecondLang";
    private final String OTHER_BROWSER_RATIO           = "probAnotherBrowser";
    private final String MIN_TEXT_SIZE                 = "minTextSize";
    private final String MAX_TEXT_SIZE                 = "maxTextSize";
    private final String MIN_COMMENT_SIZE              = "minCommentSize";
    private final String MAX_COMMENT_SIZE              = "maxCommentSize";
    private final String REDUCE_TEXT_RATIO             = "ratioReduceText";


    private final String MIN_LARGE_POST_SIZE                 = "minLargePostSize";
    private final String MAX_LARGE_POST_SIZE                 = "maxLargePostSize";
    private final String MIN_LARGE_COMMENT_SIZE              = "minLargeCommentSize";
    private final String MAX_LARGE_COMMENT_SIZE              = "maxLargeCommentSize";
    private final String LARGE_POST_RATIO                    = "ratioLargePost";
    private final String LARGE_COMMENT_RATIO                 = "ratioLargeComment";

    private final String MAX_PHOTOALBUM                = "maxNumPhotoAlbumsPerMonth";
    private final String MAX_PHOTO_PER_ALBUM           = "maxNumPhotoPerAlbums";
    private final String USER_MAX_GROUP                = "maxNumGroupCreatedPerUser";
    private final String MAX_GROUP_MEMBERS             = "maxNumMemberGroup";
    private final String GROUP_MODERATOR_RATIO         = "groupModeratorProb";
    private final String GROUP_MAX_POST_MONTH          = "maxNumGroupPostPerMonth";
    private final String MISSING_RATIO                 = "missingRatio";
    private final String STATUS_MISSING_RATIO          = "missingStatusRatio";
    private final String STATUS_SINGLE_RATIO           = "probSingleStatus";
    private final String SMARTHPHONE_RATIO             = "probHavingSmartPhone";
    private final String AGENT_SENT_RATIO              = "probSentFromAgent";
    private final String DIFFERENT_IP_IN_TRAVEL_RATIO  = "probDiffIPinTravelSeason";
    private final String DIFFERENT_IP_NOT_TRAVEL_RATIO = "probDiffIPnotTravelSeason";
    private final String DIFFERENT_IP_TRAVELLER_RATIO  = "probDiffIPforTraveller";
    private final String COMPANY_UNCORRELATED_RATIO    = "probUnCorrelatedCompany";
    private final String UNIVERSITY_UNCORRELATED_RATIO = "probUnCorrelatedOrganization";
    private final String BEST_UNIVERSTY_RATIO          = "probTopUniv";
    private final String MAX_POPULAR_PLACES            = "maxNumPopularPlaces";
    private final String POPULAR_PLACE_RATIO           = "probPopularPlaces";
    private final String TAG_UNCORRELATED_COUNTRY      = "tagCountryCorrProb";

    private final String FLASHMOB_TAGS_PER_MONTH                = "flashmobTagsPerMonth";
    private final String PROB_INTEREST_FLASHMOB_TAG             = "probInterestFlashmobTag";
    private final String PROB_RANDOM_PER_LEVEL                  = "probRandomPerLevel";
    private final String MAX_NUM_FLASHMOB_POST_PER_MONTH        = "maxNumFlashmobPostPerMonth";
    private final String MAX_NUM_GROUP_FLASHMOB_POST_PER_MONTH  = "maxNumGroupFlashmobPostPerMonth";
    private final String MAX_NUM_TAG_PER_FLASHMOB_POST          = "maxNumTagPerFlashmobPost";
    private final String FLASHMOB_TAG_MIN_LEVEL                 = "flashmobTagMinLevel";
    private final String FLASHMOB_TAG_MAX_LEVEL                 = "flashmobTagMaxLevel";
    private final String FLASHMOB_TAG_DIST_EXP                  = "flashmobTagDistExp";
    private final String UPDATE_PORTION                         = "updatePortion";

    /**
     * This array provides a quick way to check if any of the required parameters is missing and throw the appropriate
     * exception in the method loadParamsFromFile()
     */
    private final String[] checkParameters = {CELL_SIZE, NUM_CELL_WINDOW, MIN_FRIENDS, MAX_FRIENDS, FRIEND_REJECT,
            FRIEND_REACCEPT, USER_MIN_TAGS, USER_MAX_TAGS, USER_MAX_POST_MONTH, MAX_COMMENT_POST, LIMIT_CORRELATED,
            BASE_CORRELATED, MAX_EMAIL, MAX_COMPANIES, ENGLISH_RATIO, SECOND_LANGUAGE_RATIO, OTHER_BROWSER_RATIO,
            MIN_TEXT_SIZE, MAX_TEXT_SIZE, MIN_COMMENT_SIZE, MAX_COMMENT_SIZE, REDUCE_TEXT_RATIO,
            MIN_LARGE_POST_SIZE, MAX_LARGE_POST_SIZE, MIN_LARGE_COMMENT_SIZE, MAX_LARGE_COMMENT_SIZE, LARGE_POST_RATIO,
            LARGE_COMMENT_RATIO,MAX_PHOTOALBUM,
            MAX_PHOTO_PER_ALBUM, USER_MAX_GROUP, MAX_GROUP_MEMBERS, GROUP_MODERATOR_RATIO, GROUP_MAX_POST_MONTH,
            MISSING_RATIO, STATUS_MISSING_RATIO, STATUS_SINGLE_RATIO, SMARTHPHONE_RATIO, AGENT_SENT_RATIO,
            DIFFERENT_IP_IN_TRAVEL_RATIO, DIFFERENT_IP_NOT_TRAVEL_RATIO, DIFFERENT_IP_TRAVELLER_RATIO,
            COMPANY_UNCORRELATED_RATIO, UNIVERSITY_UNCORRELATED_RATIO, BEST_UNIVERSTY_RATIO, MAX_POPULAR_PLACES,
            POPULAR_PLACE_RATIO, TAG_UNCORRELATED_COUNTRY, FLASHMOB_TAGS_PER_MONTH,
            PROB_INTEREST_FLASHMOB_TAG, PROB_RANDOM_PER_LEVEL,MAX_NUM_FLASHMOB_POST_PER_MONTH, MAX_NUM_GROUP_FLASHMOB_POST_PER_MONTH, MAX_NUM_TAG_PER_FLASHMOB_POST, FLASHMOB_TAG_MIN_LEVEL, FLASHMOB_TAG_MAX_LEVEL,
            FLASHMOB_TAG_DIST_EXP, UPDATE_PORTION};

    //final user parameters
    private final String SCALE_FACTOR       = "scaleFactor";
    private final String SERIALIZER         = "serializer";
    private final String EXPORT_TEXT        = "exportText";
    private final String ENABLE_COMPRESSION = "enableCompression";
    private final String NUM_PERSONS        = "numPersons";
    private final String NUM_YEARS          = "numYears";
    private final String START_YEAR         = "startYear";

    /**
     * This array provides a quick way to check if any of the required parameters is missing and throw the appropriate
     * exception in the method loadParamsFromFile()
     */
    private final String[] publicCheckParameters = {SERIALIZER, ENABLE_COMPRESSION};

    // Gender string representation, both representations vector/standalone so the string is coherent.
    private final String MALE   = "male";
    private final String FEMALE = "female";
    private final String gender[] = { MALE, FEMALE};

    //Stat container
    private Statistics stats;
    
    // bookkeeping for parameter generation
    private HashMap<Long, ReducedUserProfile.Counts> factorTable;
    private HashMap<Integer, Integer> postsPerCountry;
    private HashMap<Integer, Integer> tagClassCount;
    private HashMap<String, Integer> firstNameCount;
    private HashMap<Integer, Integer> tagNameCount;
    private HashMap<Long, String> medianFirstName;
    // For blocking
    private static final int  reducerShift[] = { 26, 8, 1 };

    // For sliding window
    int 					cellSize; // Number of user in one cell
    int 					numberOfCellPerWindow;
    int	 					numTotalUser;
    int 					windowSize;

    ReducedUserProfile  	reducedUserProfiles[];
    ReducedUserProfile 		reducedUserProfilesCell[];

    // For friendship generation
    int 				friendshipNum;
    int 				minNumFriends;
    int 				maxNumFriends;
    double 				friendRejectRatio;
    double 				friendReApproveRatio;

    StorageManager		groupStoreManager;

    MRWriter			mrWriter;

    double 				baseProbCorrelated;
    double	 			limitProCorrelated;

    // For each user
    int                 minNumTagsPerUser;
    int					maxNumTagsPerUser;
    int 				maxNumPostPerMonth;
    int 				maxNumComments;

    // Random values generators
    PowerDistGenerator 	randomPowerLaw;
    PowerDistGenerator  randomTagPowerLaw;

    DateGenerator 		dateTimeGenerator;
    int					startYear;
    int                 endYear;

    // Dictionary classes
    LocationDictionary 		locationDictionary;
    LanguageDictionary      languageDictionary;
    TagDictionary 			tagDictionary;

    //For facebook-like social degree distribution
    FBSocialDegreeGenerator	fbDegreeGenerator;
    FlashmobTagDictionary   flashmobTagDictionary;
    TagTextDictionary       tagTextDictionary;
    TagMatrix	 			topicTagDictionary;
    NamesDictionary 		namesDictionary;
    UniversityDictionary    unversityDictionary;
    CompanyDictionary       companiesDictionary;
    UserAgentDictionary     userAgentDictionary;
    EmailDictionary         emailDictionary;
    BrowserDictionary       browserDictonry;
    PopularPlacesDictionary popularDictionary;
    IPAddressDictionary     ipAddDictionary;

    int                     maxNumPopularPlaces;
    double                  tagCountryCorrProb;
    double 					probUnCorrelatedOrganization;
    double 					probTopUniv; // 90% users go to top university
    double 					probUnCorrelatedCompany;
    double 					probPopularPlaces;		//probability of taking a photo at popular place

    // For generating texts of posts and comments
    int                     maxEmails;
    int                     maxCompanies;
    double                  probEnglish;
    double                  probSecondLang;

    int 					minTextSize;
    int 					maxTextSize;
    int 					minCommentSize;
    int 					maxCommentSize;
    double 					ratioReduceText; // 80% text has size less than 1/2 max size

    // This parameters configure the amount of large posts are created 
    int                     minLargePostSize;
    int                     maxLargePostSize;
    int                     minLargeCommentSize;
    int                     maxLargeCommentSize;
    double                  ratioLargePost;
    double                  ratioLargeComment;

    // For photo generator
    PhotoGenerator 			photoGenerator;
    int 					maxNumPhotoAlbumsPerMonth;
    int 					maxNumPhotoPerAlbums;

    // For generating groups
    GroupGenerator 			groupGenerator;
    int 					maxNumGroupCreatedPerUser;
    int 					maxNumMemberGroup;
    double 					groupModeratorProb;

    // For group posts
    UniformPostGenerator           uniformPostGenerator;
    FlashmobPostGenerator          flashmobPostGenerator;
    CommentGenerator        commentGenerator;
    int 					maxNumGroupPostPerMonth;


    // For serialize to RDF format
    DataExporter    dataExporter = null;
    String 			serializerType;
    String 			outUserProfileName = "userProf.ser";
    String 			outUserProfile;
    int				threadId;
    int				numThreads;
    String 			sibOutputDir;

    // For user's extra info
    double	       missingRatio;
    double         missingStatusRatio;
    double         probSingleStatus; // Status "Single" has more probability than others'


    double 		   probAnotherBrowser;
    double         probHavingSmartPhone;
    double         probSentFromAgent;

    // The probability that normal user posts from different location
    double         probDiffIPinTravelSeason; // in travel season
    double         probDiffIPnotTravelSeason; // not in travel season

    // The probability that travellers post from different location
    double         probDiffIPforTraveller;

    // Writing data for test driver
    int            thresholdPopularUser = 40;
    int            numPopularUser = 0;

    int            flashmobTagsPerMonth = 0;
    double         probInterestFlashmobTag = 0.0;
    double         probRandomPerLevel = 0.0;
    int            maxNumFlashmobPostPerMonth = 0;
    int            maxNumGroupFlashmobPostPerMonth = 0;
    int            maxNumTagPerFlashmobPost = 0;
    double         flashmobTagMinLevel = 0.0f;
    double         flashmobTagMaxLevel = 0.0f;
    double         flashmobTagDistExp  = 0.0f;

    long            deltaTime = 0;
    long            dateThreshold = 0;
    double          updatePortion = 0.0;

    // Data accessed from the hadoop jobs
    private ReducedUserProfile[] cellReducedUserProfiles;
    private int     numUserProfilesRead      = 0;
    public int     totalNumUserProfilesRead = 0;
    private int     numUserForNewCell        = 0;
    private int     mrCurCellPost            = 0;
    public static int     blockId                 = 0;
    public int     exactOutput              = 0;

    public boolean exportText = true;
    public boolean enableCompression = true;

    RandomGeneratorFarm randomFarm;
    public static final int blockSize = 10000;

    TreeMap<Integer, ScaleFactor> scaleFactors;

    /**
     * Creates the ScalableGenerator*/
    public ScalableGenerator(int threadId, Configuration conf ){
        this.conf = conf;
        this.sibOutputDir = this.conf.get("outputDir")+"/social_network";
        this.numThreads = Integer.parseInt(this.conf.get("numThreads"));
        this.threadId = threadId;
        this.deltaTime = Integer.parseInt(this.conf.get("deltaTime"));
        this.stats = new Statistics();
        this.postsPerCountry = new HashMap<Integer, Integer>();
        this.tagClassCount = new HashMap<Integer, Integer>();
        this.firstNameCount = new HashMap<String, Integer>();
        this.medianFirstName = new HashMap<Long, String>();
        this.tagNameCount = new HashMap<Integer, Integer>();
        if (threadId != -1){
            outUserProfile = "mr" + threadId + "_" + outUserProfileName;
        }
    }

    /**
     * Initializes the generator reading the private parameter file, the user parameter file
     * and initialize all the internal variables.
     */
    public void init(){
        this.factorTable = new HashMap<Long, ReducedUserProfile.Counts>();
        loadParamsFromFile();
        randomFarm = new RandomGeneratorFarm();
        windowSize = cellSize * numberOfCellPerWindow;                          // We compute the size of the window.
        mrWriter = new MRWriter(cellSize, windowSize, sibOutputDir);
        resetWindow();
        randomPowerLaw    = new PowerDistGenerator(minNumFriends,     maxNumFriends + 1,     alpha);
        randomTagPowerLaw = new PowerDistGenerator(minNumTagsPerUser, maxNumTagsPerUser + 1, alpha);

        // Initializing window memory
        reducedUserProfiles = new ReducedUserProfile[windowSize];
        cellReducedUserProfiles = new ReducedUserProfile[cellSize];

        dateTimeGenerator = new DateGenerator( new GregorianCalendar(startYear, startMonth, startDate),
                new GregorianCalendar(endYear, endMonth, endDate), alpha, deltaTime);
        if(!Boolean.parseBoolean(conf.get("updateStreams"))) updatePortion = 0.0;
        dateThreshold = dateTimeGenerator.getMaxDateTime() - (long)((dateTimeGenerator.getMaxDateTime() - dateTimeGenerator.getStartDateTime())*(updatePortion));
        SN.minDate = dateTimeGenerator.getStartDateTime();
        SN.maxDate = dateTimeGenerator.getMaxDateTime();

        System.out.println("Building location dictionary ");
        locationDictionary = new LocationDictionary(numTotalUser, countryDictionaryFile, cityDictionaryFile);
        locationDictionary.init();

        System.out.println("Building language dictionary ");
        languageDictionary = new LanguageDictionary(languageDictionaryFile, locationDictionary,
                probEnglish, probSecondLang);
        languageDictionary.init();

        System.out.println("Building Tag dictionary ");
        tagDictionary = new TagDictionary(tagNamesFile, tagDictionaryFile, tagClassFile, tagHierarchyFile,
                locationDictionary.getCountries().size(), tagCountryCorrProb);
        tagDictionary.initialize();

        System.out.println("Building Tag-text dictionary ");
        tagTextDictionary = new TagTextDictionary(tagTextFile, dateTimeGenerator, tagDictionary,
                ratioReduceText);
        tagTextDictionary.initialize();

        System.out.println("Building Tag Matrix dictionary ");
        topicTagDictionary = new TagMatrix(tagTopicDictionaryFile, tagDictionary.getNumCelebrity());
        topicTagDictionary.initMatrix();

        System.out.println("Building IP addresses dictionary ");
        ipAddDictionary = new IPAddressDictionary(countryAbbrMappingFile,
                IPZONE_DIRECTORY, locationDictionary, probDiffIPinTravelSeason, probDiffIPnotTravelSeason,
                probDiffIPforTraveller);
        ipAddDictionary.initialize();

        System.out.println("Building Names dictionary");
        namesDictionary = new NamesDictionary(surnamDictionaryFile, nameDictionaryFile,
                locationDictionary);
        namesDictionary.init();

        System.out.println("Building email dictionary");
        emailDictionary = new EmailDictionary(emailDictionaryFile);
        emailDictionary.init();

        System.out.println("Building browser dictionary");
        browserDictonry = new BrowserDictionary(browserDictonryFile, probAnotherBrowser);
        browserDictonry.init();


        System.out.println("Building companies dictionary");
        companiesDictionary = new CompanyDictionary(companiesDictionaryFile,locationDictionary,
                probUnCorrelatedCompany);
        companiesDictionary.init();

        System.out.println("Building university dictionary");
        unversityDictionary = new UniversityDictionary(universityDictionaryFile, locationDictionary,
                probUnCorrelatedOrganization, probTopUniv, companiesDictionary.getNumCompanies());
        unversityDictionary.init();

        System.out.println("Building popular places dictionary");
        popularDictionary = new PopularPlacesDictionary(popularDictionaryFile,
                locationDictionary);
        popularDictionary.init();

        System.out.println("Building user agents dictionary");
        userAgentDictionary = new UserAgentDictionary(agentFile, probSentFromAgent);
        userAgentDictionary.init();

        // Building generators.
        System.out.println("Building photo generator");
        photoGenerator = new PhotoGenerator(dateTimeGenerator,
                locationDictionary, 0, popularDictionary, probPopularPlaces, maxNumLikes, deltaTime, randomFarm);

        System.out.println("Building Group generator");
        groupGenerator = new GroupGenerator(dateTimeGenerator, locationDictionary,
                tagDictionary, numTotalUser);



        /// IMPORTANT: ratioLargeText is divided 0.083333, the probability
        /// that SetUserLargePoster returns true.
        System.out.println("Building Uniform Post Generator");
        uniformPostGenerator = new UniformPostGenerator( dateTimeGenerator,
                tagTextDictionary,
                userAgentDictionary,
                ipAddDictionary,
                browserDictonry,
                minTextSize,
                maxTextSize,
                ratioReduceText,
                minLargePostSize,
                maxLargePostSize,
                ratioLargePost/0.0833333,
                maxNumLikes,
                exportText,
                deltaTime,
                maxNumPostPerMonth,
                maxNumFriends,
                maxNumGroupPostPerMonth,
                maxNumMemberGroup
        );
        uniformPostGenerator.initialize();

        System.out.println("Building Flashmob Tag Dictionary");
        flashmobTagDictionary = new FlashmobTagDictionary( tagDictionary,
                dateTimeGenerator,
                flashmobTagsPerMonth,
                probInterestFlashmobTag,
                probRandomPerLevel,
                flashmobTagMinLevel,
                flashmobTagMaxLevel,
                flashmobTagDistExp,
                0
        );
        flashmobTagDictionary.initialize();

        stats.flashmobTags = flashmobTagDictionary.getFlashmobTags();

        /// IMPORTANT: ratioLargeText is divided 0.083333, the probability
        /// that SetUserLargePoster returns true.
        System.out.println("Building Flashmob Post Generator");
        flashmobPostGenerator = new FlashmobPostGenerator(dateTimeGenerator, tagTextDictionary,
                userAgentDictionary,
                ipAddDictionary,
                browserDictonry,
                minTextSize,
                maxTextSize,
                ratioReduceText,
                minLargePostSize,
                maxLargePostSize,
                ratioLargePost/0.0833333,
                maxNumLikes,
                exportText,
                deltaTime,
                flashmobTagDictionary,
                topicTagDictionary,
                maxNumFlashmobPostPerMonth,
                maxNumGroupFlashmobPostPerMonth,
                maxNumFriends,
                maxNumMemberGroup,
                maxNumTagPerFlashmobPost,
                flashmobDistFile
        );
        flashmobPostGenerator.initialize();

        /// IMPORTANT: ratioLargeText is divided 0.083333, the probability
        /// that SetUserLargePoster returns true.
        System.out.println("Building Comment Generator");
        commentGenerator = new CommentGenerator( tagDictionary,
                tagTextDictionary,
                topicTagDictionary,
                dateTimeGenerator,
                browserDictonry,
                ipAddDictionary,
                userAgentDictionary,
                minCommentSize,
                maxCommentSize,
                minLargeCommentSize,
                maxLargeCommentSize,
                ratioLargeComment/0.0833333,
                maxNumLikes,exportText,
                deltaTime
        );
        commentGenerator.initialize();

        System.out.println("Building Facebook-like social degree generator");
        fbDegreeGenerator = new FBSocialDegreeGenerator(numTotalUser, fbSocialDegreeFile, 0);
        fbDegreeGenerator.loadFBBuckets();
        fbDegreeGenerator.rebuildBucketRange();
    }

    /**
     * Reads and loads the private parameter file and the user parameter file.
     */
    private void loadParamsFromFile() {
        try {
            //First read the internal params.ini
            Properties properties = new Properties();
            properties.load(new InputStreamReader(getClass( ).getResourceAsStream("/"+PARAMETERS_FILE), "UTF-8"));
            for (int i = 0; i < checkParameters.length; i++) {
                if (properties.getProperty(checkParameters[i]) == null) {
                    throw new IllegalStateException("Missing " + checkParameters[i] + " parameter");
                }
            }

            cellSize = Short.parseShort(properties.getProperty(CELL_SIZE));
            numberOfCellPerWindow = Integer.parseInt(properties.getProperty(NUM_CELL_WINDOW));
            minNumFriends = Integer.parseInt(properties.getProperty(MIN_FRIENDS));
            maxNumFriends = Integer.parseInt(properties.getProperty(MAX_FRIENDS));
            thresholdPopularUser = (int) (maxNumFriends * 0.9);
            friendRejectRatio = Double.parseDouble(properties.getProperty(FRIEND_REJECT));
            friendReApproveRatio = Double.parseDouble(properties.getProperty(FRIEND_REACCEPT));
            minNumTagsPerUser= Integer.parseInt(properties.getProperty(USER_MIN_TAGS));
            maxNumTagsPerUser= Integer.parseInt(properties.getProperty(USER_MAX_TAGS));
            maxNumPostPerMonth = Integer.parseInt(properties.getProperty(USER_MAX_POST_MONTH));
            maxNumComments = Integer.parseInt(properties.getProperty(MAX_COMMENT_POST));
            limitProCorrelated = Double.parseDouble(properties.getProperty(LIMIT_CORRELATED));
            baseProbCorrelated = Double.parseDouble(properties.getProperty(BASE_CORRELATED));
            maxEmails = Integer.parseInt(properties.getProperty(MAX_EMAIL));
            maxCompanies = Integer.parseInt(properties.getProperty(MAX_EMAIL));
            probEnglish = Double.parseDouble(properties.getProperty(MAX_EMAIL));
            probSecondLang = Double.parseDouble(properties.getProperty(MAX_EMAIL));
            probAnotherBrowser = Double.parseDouble(properties.getProperty(OTHER_BROWSER_RATIO));
            minTextSize = Integer.parseInt(properties.getProperty(MIN_TEXT_SIZE));
            maxTextSize = Integer.parseInt(properties.getProperty(MAX_TEXT_SIZE));
            minCommentSize = Integer.parseInt(properties.getProperty(MIN_COMMENT_SIZE));
            maxCommentSize = Integer.parseInt(properties.getProperty(MAX_COMMENT_SIZE));
            ratioReduceText = Double.parseDouble(properties.getProperty(REDUCE_TEXT_RATIO));
            minLargePostSize = Integer.parseInt(properties.getProperty(MIN_LARGE_POST_SIZE));
            maxLargePostSize = Integer.parseInt(properties.getProperty(MAX_LARGE_POST_SIZE));
            minLargeCommentSize = Integer.parseInt(properties.getProperty(MIN_LARGE_COMMENT_SIZE));
            maxLargeCommentSize = Integer.parseInt(properties.getProperty(MAX_LARGE_COMMENT_SIZE));
            ratioLargePost = Double.parseDouble(properties.getProperty(LARGE_POST_RATIO));
            ratioLargeComment = Double.parseDouble(properties.getProperty(LARGE_COMMENT_RATIO));
            maxNumPhotoAlbumsPerMonth = Integer.parseInt(properties.getProperty(MAX_PHOTOALBUM));
            maxNumPhotoPerAlbums = Integer.parseInt(properties.getProperty(MAX_PHOTO_PER_ALBUM));
            maxNumGroupCreatedPerUser = Integer.parseInt(properties.getProperty(USER_MAX_GROUP));
            maxNumMemberGroup = Integer.parseInt(properties.getProperty(MAX_GROUP_MEMBERS));
            groupModeratorProb = Double.parseDouble(properties.getProperty(GROUP_MODERATOR_RATIO));
            maxNumGroupPostPerMonth = Integer.parseInt(properties.getProperty(GROUP_MAX_POST_MONTH));
            missingRatio = Double.parseDouble(properties.getProperty(MISSING_RATIO));
            missingStatusRatio = Double.parseDouble(properties.getProperty(STATUS_MISSING_RATIO));
            probSingleStatus = Double.parseDouble(properties.getProperty(STATUS_SINGLE_RATIO));
            probHavingSmartPhone = Double.parseDouble(properties.getProperty(SMARTHPHONE_RATIO));
            probSentFromAgent = Double.parseDouble(properties.getProperty(AGENT_SENT_RATIO));
            probDiffIPinTravelSeason = Double.parseDouble(properties.getProperty(DIFFERENT_IP_IN_TRAVEL_RATIO));
            probDiffIPnotTravelSeason = Double.parseDouble(properties.getProperty(DIFFERENT_IP_NOT_TRAVEL_RATIO));
            probDiffIPforTraveller = Double.parseDouble(properties.getProperty(DIFFERENT_IP_TRAVELLER_RATIO));
            probUnCorrelatedCompany = Double.parseDouble(properties.getProperty(COMPANY_UNCORRELATED_RATIO));
            probUnCorrelatedOrganization = Double.parseDouble(properties.getProperty(UNIVERSITY_UNCORRELATED_RATIO));
            probTopUniv = Double.parseDouble(properties.getProperty(BEST_UNIVERSTY_RATIO));
            maxNumPopularPlaces = Integer.parseInt(properties.getProperty(MAX_POPULAR_PLACES));
            probPopularPlaces = Double.parseDouble(properties.getProperty(POPULAR_PLACE_RATIO));
            tagCountryCorrProb = Double.parseDouble(properties.getProperty(TAG_UNCORRELATED_COUNTRY));
            flashmobTagsPerMonth = Integer.parseInt(properties.getProperty(FLASHMOB_TAGS_PER_MONTH));
            probInterestFlashmobTag = Double.parseDouble(properties.getProperty(PROB_INTEREST_FLASHMOB_TAG));
            probRandomPerLevel = Double.parseDouble(properties.getProperty(PROB_RANDOM_PER_LEVEL));
            maxNumFlashmobPostPerMonth = Integer.parseInt(properties.getProperty(MAX_NUM_FLASHMOB_POST_PER_MONTH));
            maxNumGroupFlashmobPostPerMonth = Integer.parseInt(properties.getProperty(MAX_NUM_GROUP_FLASHMOB_POST_PER_MONTH));
            maxNumTagPerFlashmobPost = Integer.parseInt(properties.getProperty(MAX_NUM_TAG_PER_FLASHMOB_POST));
            flashmobTagMinLevel = Double.parseDouble(properties.getProperty(FLASHMOB_TAG_MIN_LEVEL));
            flashmobTagMaxLevel = Double.parseDouble(properties.getProperty(FLASHMOB_TAG_MAX_LEVEL));
            flashmobTagDistExp = Double.parseDouble(properties.getProperty(FLASHMOB_TAG_DIST_EXP));
            updatePortion = Double.parseDouble(properties.getProperty(UPDATE_PORTION));
        } catch (Exception e) {
            System.err.println(e.getMessage());
            e.printStackTrace();
            System.exit(-1);
        }

        try {
            scaleFactors = new TreeMap<Integer,ScaleFactor>();
            DocumentBuilderFactory dbFactory = DocumentBuilderFactory.newInstance();
            DocumentBuilder dBuilder = dbFactory.newDocumentBuilder();
            Document doc = dBuilder.parse(getClass().getResourceAsStream("/"+SCALE_FACTORS_FILE));
            doc.getDocumentElement().normalize();

            System.out.println("Reading scale factors..");
            NodeList nodes = doc.getElementsByTagName("scale_factor");
            for (int i = 0; i < nodes.getLength(); i++) {
                Node node = nodes.item(i);
                if (node.getNodeType() == Node.ELEMENT_NODE) {
                    Element element = (Element) node;
                    Integer num = Integer.parseInt(element.getAttribute("number"));
                    ScaleFactor scaleFactor = new ScaleFactor();
                    NodeList files = element.getElementsByTagName("num_persons");
                    scaleFactor.numPersons = Integer.parseInt(files.item(0).getTextContent());
                    files = element.getElementsByTagName("start_year");
                    scaleFactor.startYear = Integer.parseInt(files.item(0).getTextContent());
                    files = element.getElementsByTagName("num_years");
                    scaleFactor.numYears = Integer.parseInt(files.item(0).getTextContent());
                    scaleFactors.put(num,scaleFactor);
                }
            }
        } catch (Exception e) {
            System.err.println(e.getMessage());
            System.exit(-1);
        }

        try {

            if( conf.get(NUM_PERSONS) != null && conf.get(NUM_YEARS) != null && conf.get(START_YEAR) != null) {
                numTotalUser = Integer.parseInt(conf.get(NUM_PERSONS));
                startYear = Integer.parseInt(conf.get(START_YEAR));
                int numYears = Integer.parseInt(conf.get(NUM_YEARS));
                endYear = startYear + numYears;
            } else {
                int scaleFactorId = Integer.parseInt(conf.get(SCALE_FACTOR));
                ScaleFactor scaleFactor = scaleFactors.get(scaleFactorId);
                System.out.println("Executin with scale factor " + scaleFactorId);
                System.out.println(" ... Num Persons " + scaleFactor.numPersons);
                System.out.println(" ... Start Year " + scaleFactor.startYear);
                System.out.println(" ... Num Years " + scaleFactor.numYears);
                numTotalUser = scaleFactor.numPersons;
                startYear = scaleFactor.startYear;
                int numYears = scaleFactor.numYears;
                endYear = startYear + numYears;
            }

            serializerType = conf.get(SERIALIZER);
            exportText = true;
            enableCompression = Boolean.parseBoolean(conf.get("compressed"));
            if (!serializerType.equals("ttl") && !serializerType.equals("n3") &&
                    !serializerType.equals("csv") && !serializerType.equals("none") && !serializerType.equals("csv_merge_foreign")) {
                throw new IllegalStateException("serializerType must be ttl, n3, csv, csv_merge_foreign");
            }
        } catch (Exception e) {
            System.err.println(e.getMessage());
            System.exit(-1);
        }
    }

    public void openSerializer() {
        dataExporter = getSerializer(serializerType, RDF_OUTPUT_FILE);
    }

    public void closeSerializer() {
        dataExporter.close();
        writeStatistics();
        writeFactorTable();
        System.out.println("Number of generated triples " + dataExporter.unitsGenerated());
        System.out.println("Number of popular users " + numPopularUser);
        System.out.println("Writing the data for test driver ");
    }

    public void generateUserActivity( ReducedUserProfile userProfile, Reducer<ComposedKey, ReducedUserProfile,LongWritable, ReducedUserProfile>.Context context) {
        int index = numUserProfilesRead%windowSize;
        numUserProfilesRead++;
        reducedUserProfiles[index] = userProfile;
        UserExtraInfo extraInfo = new UserExtraInfo();
        reducedUserProfiles[index].setForumWallId(SN.composeId(groupId, reducedUserProfiles[index].getCreationDate()));
        groupId++;
        setInfoFromUserProfile(reducedUserProfiles[index], extraInfo);
        UserInfo userInfo = new UserInfo();
        userInfo.user = reducedUserProfiles[index];
        userInfo.extraInfo = extraInfo;
        dataExporter.export(userInfo);
        int nameCount = firstNameCount.containsKey(extraInfo.getFirstName())? firstNameCount.get(extraInfo.getFirstName()):0;
        firstNameCount.put(extraInfo.getFirstName(), nameCount+1);
        String medianName = namesDictionary.getMedianGivenName(userProfile.getLocationId(),
                userProfile.getGender()==1, dateTimeGenerator.getBirthYear(userProfile.getBirthDay()));
        medianFirstName.put(userProfile.getAccountId(), medianName);
        if(conf.getBoolean("activity",true)) {
            Group wall = generateWall(userInfo);
            generatePosts(uniformPostGenerator, reducedUserProfiles[index], extraInfo, wall);
            generatePosts(flashmobPostGenerator, reducedUserProfiles[index], extraInfo, wall);
            generatePhotos(reducedUserProfiles[index], extraInfo);
            generateUserGroups(reducedUserProfiles[index], extraInfo);
        }
        if( numUserProfilesRead % 100 == 0) context.setStatus("Generated post and photo for "+numUserProfilesRead+" users");
        dataExporter.changePartition();
    }

    private void generateUserGroups(ReducedUserProfile userProfile, UserExtraInfo extraInfo ) {
        double moderatorProb = randomFarm.get(RandomGeneratorFarm.Aspect.GROUP_MODERATOR).nextDouble();
        if (moderatorProb <= groupModeratorProb) {
            Friend firstLevelFriends[] = userProfile.getFriendList();
            Vector<Friend> secondLevelFriends = new Vector<Friend>();
            //TODO: Include friends of friends a.k.a second level friends?
            int numGroup = randomFarm.get(RandomGeneratorFarm.Aspect.NUM_GROUP).nextInt(maxNumGroupCreatedPerUser)+1;
            for (int j = 0; j < numGroup; j++) {
                createGroupForUser(userProfile, firstLevelFriends, secondLevelFriends);
            }
        }
    }

    public void resetWindow() {
        numUserProfilesRead = 0;
        numUserForNewCell = 0;
        mrCurCellPost = 0;
    }

    public void pushUserProfile(ReducedUserProfile reduceUser, int pass, int outputDimension, Reducer<ComposedKey, ReducedUserProfile,TupleKey, ReducedUserProfile>.Context context){
        ReducedUserProfile userObject = new ReducedUserProfile();
        userObject.copyFields(reduceUser);
        totalNumUserProfilesRead++;
        if (numUserProfilesRead < windowSize) {                             // Push the user into the window if there is enought space. 
            reducedUserProfiles[numUserProfilesRead] = userObject;
            numUserProfilesRead++;
        } else {                                                            // If the window is full, push the user into the backup cell. 
            cellReducedUserProfiles[numUserForNewCell] = userObject;
            numUserForNewCell++;
            if (numUserForNewCell == cellSize) {                            // Once the backup cell is full, create friendships and slide the window.
                mr2SlideFriendShipWindow( pass,
                        mrCurCellPost,
                        context,
                        cellReducedUserProfiles,
                        outputDimension);
                mrCurCellPost++;
                numUserForNewCell = 0;
            }
        }
    }

    public void pushAllRemainingUser(int pass, int outputDimension, Reducer<ComposedKey, ReducedUserProfile,TupleKey, ReducedUserProfile>.Context context){

        // For each remianing cell in the window, we create the edges.
        for (int numLeftCell = Math.min(numberOfCellPerWindow, numUserProfilesRead/cellSize); numLeftCell > 0; --numLeftCell, ++mrCurCellPost) {
            mr2SlideLastCellsFriendShip(pass, mrCurCellPost, numLeftCell, context, outputDimension);
        }

        // We write to the context the users that might have been left into not fully filled cell.
        mrWriter.writeReducedUserProfiles(0, numUserForNewCell, outputDimension, cellReducedUserProfiles, context);
        exactOutput+=numUserForNewCell;
    }

    private void generateCellOfUsers2(int newStartIndex, ReducedUserProfile[] _cellReduceUserProfiles){
        int curIdxInWindow;
        for (int i = 0; i < cellSize; i++) {
            curIdxInWindow = newStartIndex + i;
            reducedUserProfiles[curIdxInWindow] = _cellReduceUserProfiles[i];
        }
    }

    private void mr2SlideFriendShipWindow(int pass, int cellPos, Reducer.Context context, ReducedUserProfile[] _cellReduceUserProfiles,
                                          int outputDimension){

        int cellPosInWindow = (cellPos) % numberOfCellPerWindow;
        int startIndex = cellPosInWindow * cellSize;
        for (int i = 0; i < cellSize; i++) {
            int curIdxInWindow = startIndex + i;
            for (int j = i + 1; (j < windowSize - 1) &&
                    reducedUserProfiles[curIdxInWindow].getNumFriendsAdded() < reducedUserProfiles[curIdxInWindow].getNumPassFriends(pass);
                 j++) {
                int checkFriendIdx = (curIdxInWindow + j) % windowSize;
                if ( !(reducedUserProfiles[checkFriendIdx].getNumFriendsAdded() == reducedUserProfiles[checkFriendIdx].getNumPassFriends(pass) ||
                        reducedUserProfiles[curIdxInWindow].isExistFriend(reducedUserProfiles[checkFriendIdx].getAccountId()))) {
                    double randProb = randomFarm.get(RandomGeneratorFarm.Aspect.UNIFORM).nextDouble();
                    double prob = getFriendCreatePro(curIdxInWindow, checkFriendIdx, pass);
                    if ((randProb < prob) || (randProb < limitProCorrelated)) {
                        createFriendShip(reducedUserProfiles[curIdxInWindow], reducedUserProfiles[checkFriendIdx], (byte) pass);
                    }
                }
            }
        }
        updateLastPassFriendAdded(startIndex, startIndex + cellSize, pass);
        mrWriter.writeReducedUserProfiles(startIndex, startIndex + cellSize, outputDimension, reducedUserProfiles, context);
        generateCellOfUsers2(startIndex, _cellReduceUserProfiles);
        exactOutput = exactOutput + cellSize;
    }

    private void mr2SlideLastCellsFriendShip(int pass, int cellPos,	int numleftCell, Reducer.Context context, int outputDimension) {

        int startIndex = (cellPos % numberOfCellPerWindow) * cellSize;
        for (int i = 0; i < cellSize; i++) {
            int curIdxInWindow = startIndex + i;
            // From this user, check all the user in the window to create friendship
            for (int j = i + 1; (j < numleftCell * cellSize )
                    && reducedUserProfiles[curIdxInWindow].getNumFriendsAdded()
                    < reducedUserProfiles[curIdxInWindow].getNumPassFriends(pass);
                 j++) {
                int checkFriendIdx = (curIdxInWindow + j) % windowSize;
                if ( !(reducedUserProfiles[checkFriendIdx].getNumFriendsAdded() ==
                        reducedUserProfiles[checkFriendIdx].getNumPassFriends(pass) ||
                        reducedUserProfiles[curIdxInWindow].isExistFriend(reducedUserProfiles[checkFriendIdx].getAccountId()))) {
                    double randProb = randomFarm.get(RandomGeneratorFarm.Aspect.UNIFORM).nextDouble();
                    double prob = getFriendCreatePro(curIdxInWindow, checkFriendIdx, pass);
                    if ((randProb < prob) || (randProb < limitProCorrelated)) {
                        createFriendShip(reducedUserProfiles[curIdxInWindow], reducedUserProfiles[checkFriendIdx],
                                (byte) pass);
                    }
                }
            }
        }
        updateLastPassFriendAdded(startIndex, startIndex + cellSize, pass);
        mrWriter.writeReducedUserProfiles(startIndex, startIndex + cellSize, outputDimension, reducedUserProfiles, context);
        exactOutput = exactOutput + cellSize;
    }



    public void resetState(int seed) {
        blockId = seed;
        randomFarm.resetRandomGenerators((long)blockId);
        postId = 0;
        groupId = 0;
        SN.setMachineNumber(blockId, (int)Math.ceil(numTotalUser / (double) (blockSize)) );
        fbDegreeGenerator.resetState(blockId);
        resetWindow();
        if( dataExporter != null ) {
            dataExporter.resetState(seed);
        }
    }


    /**
     * Generates the users. The user generation process is divided in blocks of size four times the
     * size of the window. At the beginning of each block, the seeds of the random number generators
     * are reset. This is for the sake of determinism, so independently of the mapper that receives a
     * block, the seeds are set deterministically, and therefore, we make this generation phase 
     * deterministic. This implies that the different mappers have to process blocks of full size,
     * that is, a block have to be fully processed by a mapper.
     *
     * @param pass The pass identifying the current pass.
     * @param context The map-reduce context.
     */
    public void mrGenerateUserInfo(int pass, Context context){

        if (numTotalUser % cellSize != 0) {
            System.err.println("Number of users should be a multiple of the cellsize");
            System.exit(-1);
        }

        // Here we determine the blocks in the "block space" that this mapper is responsible for.
        int numBlocks = (int) (Math.ceil(numTotalUser / (double)blockSize));
        int initBlock = (int) (Math.ceil((numBlocks / (double)numThreads) * threadId ));
        int endBlock  = (int) (Math.ceil((numBlocks / (double)numThreads) * (threadId+1)));

        int numUsersToGenerate = 0;
        for( int i = initBlock; i < endBlock;++i) {
            // Setting the state for the block
            resetState(i);
            locationDictionary.advanceToUser(i*blockSize);      // We location dictionary has to be advanced to the specific point.
            for (int j = i*blockSize; j < (i+1)*blockSize && j < numTotalUser; ++j) {
                ReducedUserProfile reduceUserProf = generateGeneralInformation(j);
                ++numUsersToGenerate;
                try {
                    //context.write(new LongWritable(reduceUserProf.getDicElementId(pass)), reduceUserProf);
                    context.write(new TupleKey(reduceUserProf.getDicElementId(pass),reduceUserProf.getAccountId()), reduceUserProf);
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
        System.out.println("Number of generated users: "+numUsersToGenerate);
    }


    private Group generateWall(UserInfo userInfo) {
        Group group = new Group();
        group.setCreatedDate(userInfo.user.getCreationDate()+deltaTime);
        group.setGroupName("Wall of " + userInfo.extraInfo.getFirstName() + " " + userInfo.extraInfo.getLastName());
        group.setGroupId(userInfo.user.getForumWallId());
        group.setModeratorId(userInfo.user.getAccountId());

        Iterator<Integer> itTags = userInfo.user.getSetOfTags().iterator();
        Integer tags[] = new Integer[userInfo.user.getSetOfTags().size()];
        int index = 0;
        while (itTags.hasNext()){
            tags[index] = itTags.next();
            index++;
        }
        group.setTags(tags);
        Friend[] friends = userInfo.user.getFriendList();
        group.initAllMemberships(friends.length);
        for (int i = 0; i < friends.length; i ++){
            if (friends[i] != null && friends[i].getCreatedTime() != -1){
                GroupMemberShip membership = new GroupMemberShip();
                membership.setGroupId(group.getGroupId());
                membership.setJoinDate(friends[i].getCreatedTime()+deltaTime);
                membership.setUserId(friends[i].getFriendAcc());
                membership.setIP(friends[i].getSourceIp());
                membership.userCreationDate = friends[i].toCreationDate;
                membership.setBrowserIdx(friends[i].getBrowserIdx());
                membership.setAgentIdx(friends[i].getAgentIdx());
                membership.setFrequentChange(friends[i].isFrequentChange());
                membership.setHaveSmartPhone(friends[i].isHaveSmartPhone());
                membership.setLargePoster(friends[i].isLargePoster());
                group.addMember(membership);
            }
        }
        dataExporter.export(group,userInfo.user.getCreationDate());
        return group;
    }

    private void generatePosts(PostGenerator postGenerator, ReducedUserProfile user, UserExtraInfo extraInfo, Group group){
        Vector<Post> createdPosts = postGenerator.createPosts( randomFarm, user, extraInfo, postId );
        //Vector<Post> createdPosts = postGenerator.createPosts(randomFarm,group,postId);
        postId+=createdPosts.size();
        Iterator<Post> it = createdPosts.iterator();
        
        while(it.hasNext()) {
            Post post = it.next();
            if (post.getTags() != null) {
            	for (Integer t: post.getTags()){
            		Integer tagClass = tagDictionary.getTagClass(t);
                	Integer tagCount = tagClassCount.containsKey(tagClass) ? tagClassCount.get(tagClass) : 0;
                	tagClassCount.put(tagClass, tagCount+1);
                    tagCount = tagNameCount.containsKey(t) ? tagNameCount.get(t) : 0;
                    tagNameCount.put(t, tagCount + 1);
            	}
            }
            
            int locationID = ipAddDictionary.getLocation(post.getIpAddress());
            String countryName = locationDictionary.getLocationName(locationID);
            stats.countries.add(countryName);
                       
            int postCount = postsPerCountry.containsKey(locationID) ? postsPerCountry.get(locationID) : 0;
            postsPerCountry.put(locationID, postCount+1);

            GregorianCalendar date = new GregorianCalendar();
            date.setTimeInMillis(post.getCreationDate());
            String strCreationDate = DateGenerator.formatYear(date);
            
            if (stats.maxPostCreationDate == null) {
                stats.maxPostCreationDate = strCreationDate;
                stats.minPostCreationDate = strCreationDate;
            } else {
                if (stats.maxPostCreationDate.compareTo(strCreationDate) < 0) {
                    stats.maxPostCreationDate = strCreationDate;
                }
                if (stats.minPostCreationDate.compareTo(strCreationDate) > 0) {
                    stats.minPostCreationDate = strCreationDate;
                }
            }
            dataExporter.export(post,post.getAuthorCreationDate());
            // Generate comments
            int numComment = randomFarm.get(RandomGeneratorFarm.Aspect.NUM_COMMENT).nextInt(maxNumComments+1);
            ArrayList<Message> replyCandidates = new ArrayList<Message>();
            replyCandidates.add(post);
            for (int l = 0; l < numComment; l++) {
                int replyIndex = randomFarm.get(RandomGeneratorFarm.Aspect.REPLY_TO).nextInt(replyCandidates.size());
                Message replyTo = replyCandidates.get(replyIndex);
                Comment comment = commentGenerator.createComment(randomFarm, postId, post, replyTo, user);
                
                if ( comment!=null ) {
                    if (!factorTable.containsKey(replyTo.getAuthorId())) {
                    	factorTable.put(replyTo.getAuthorId(), new ReducedUserProfile.Counts());
                    }
                    factorTable.get(replyTo.getAuthorId()).numberOfPostReplies++;
                	
                   	if (comment.getTags() != null) {
                    	for (Integer t: comment.getTags()){
                    		Integer tagClass = tagDictionary.getTagClass(t);
                        	Integer tagCount = tagClassCount.containsKey(tagClass) ? tagClassCount.get(tagClass) : 0;
                        	tagClassCount.put(tagClass, tagCount+1);
                        	tagCount = tagNameCount.containsKey(t) ? tagNameCount.get(t) : 0;
                        	tagNameCount.put(t, tagCount + 1);
                    	}
                	}
                	locationID = ipAddDictionary.getLocation(comment.getIpAddress());
                	countryName = locationDictionary.getLocationName(locationID);
                	postCount = postsPerCountry.containsKey(locationID) ? postsPerCountry.get(locationID) : 0;
                	postsPerCountry.put(locationID, postCount+1);
                    stats.countries.add(countryName);
                    dataExporter.export(comment,comment.getAuthorCreationDate());
                    if( comment.getTextSize() > 10 ) replyCandidates.add(comment);
                    postId++;
                }
            }
        }
    }

    private void generatePhotos(ReducedUserProfile user, UserExtraInfo extraInfo){
        // Generate photo Album and photos
        int numOfmonths = (int) dateTimeGenerator.numberOfMonths(user);
        int numPhotoAlbums = randomFarm.get(RandomGeneratorFarm.Aspect.NUM_PHOTO_ALBUM).nextInt(maxNumPhotoAlbumsPerMonth+1);
        if (numOfmonths != 0) {
            numPhotoAlbums = numOfmonths * numPhotoAlbums;
        }

        for (int m = 0; m < numPhotoAlbums; m++) {
            Group album = groupGenerator.createAlbum(randomFarm, groupId, user, extraInfo, m, joinProbs[0]);
            if( album != null ) {
                groupId++;
                dataExporter.export(album,user.getCreationDate());

                // Generate photos for this album
                int numPhotos = randomFarm.get(RandomGeneratorFarm.Aspect.NUM_PHOTO).nextInt(maxNumPhotoPerAlbums+1);
                
                for (int l = 0; l < numPhotos; l++) {
                    Photo photo = photoGenerator.generatePhoto(user, album, l, postId);
                    if( photo != null ) {
                        postId++;
                        photo.setUserAgent(userAgentDictionary.getUserAgentName(randomFarm.get(RandomGeneratorFarm.Aspect.USER_AGENT_SENT),user.isHaveSmartPhone(), user.getAgentId()));
                        photo.setBrowserIdx(browserDictonry.getPostBrowserId(randomFarm.get(RandomGeneratorFarm.Aspect.DIFF_BROWSER),randomFarm.get(RandomGeneratorFarm.Aspect.BROWSER), user.getBrowserIdx()));
                        photo.setIpAddress(ipAddDictionary.getIP(randomFarm.get(RandomGeneratorFarm.Aspect.IP), randomFarm.get(RandomGeneratorFarm.Aspect.DIFF_IP), randomFarm.get(RandomGeneratorFarm.Aspect.DIFF_IP_FOR_TRAVELER), user.getIpAddress(),
                                user.isFrequentChange(), photo.getCreationDate(), photo.getLocationId()));
                        int locationID = ipAddDictionary.getLocation(photo.getIpAddress());
                        String countryName = locationDictionary.getLocationName(locationID);
                        int postCount = postsPerCountry.containsKey(locationID) ? postsPerCountry.get(locationID) : 0;
                        postsPerCountry.put(locationID, postCount+1);
                        stats.countries.add(countryName);
                        dataExporter.export(photo,photo.getAuthorCreationDate());
                        
                    	if (photo.getTags() != null) {
                        	for (Integer t: photo.getTags()){
                        		Integer tagClass = tagDictionary.getTagClass(t);
                            		Integer tagCount = tagClassCount.containsKey(tagClass) ? tagClassCount.get(tagClass) : 0;
                            		tagClassCount.put(tagClass, tagCount+1);
		                        tagCount = tagNameCount.containsKey(t) ? tagNameCount.get(t) : 0;
                		        tagNameCount.put(t, tagCount + 1);
                        	}
                    	}
                    }
                }
            }
        }
    }

    private void createGroupForUser(ReducedUserProfile user,
                                   Friend firstLevelFriends[], Vector<Friend> secondLevelFriends) {
        double randLevelProb;
        double randMemberProb;

        Group group = groupGenerator.createGroup(randomFarm,groupId,user);

        if( group != null ) {
            groupId++;
            TreeSet<Long> memberIds = new TreeSet<Long>();

            int numGroupMember = randomFarm.get(RandomGeneratorFarm.Aspect.NUM_USERS_PER_GROUP).nextInt(maxNumMemberGroup);
            group.initAllMemberships(numGroupMember);

            int numLoop = 0;
            while ((group.getNumMemberAdded() < numGroupMember) && (numLoop < windowSize)) {
                randLevelProb = randomFarm.get(RandomGeneratorFarm.Aspect.FRIEND_LEVEL).nextDouble();
                // Select the appropriate friend level
                if (randLevelProb < levelProbs[0] && user.getNumFriendsAdded() > 0 ) { // ==> level 1
                    int friendIdx = randomFarm.get(RandomGeneratorFarm.Aspect.MEMBERSHIP_INDEX).nextInt(user.getNumFriendsAdded());
                    long potentialMemberAcc = firstLevelFriends[friendIdx].getFriendAcc();
                    randMemberProb = randomFarm.get(RandomGeneratorFarm.Aspect.MEMBERSHIP).nextDouble();
                    if (randMemberProb < joinProbs[0]) {
                        // Check whether this user has been added and then add to the group
                        if (!memberIds.contains(potentialMemberAcc)) {
                            memberIds.add(potentialMemberAcc);
                            // Assume the earliest membership date is the friendship created date
                            GroupMemberShip memberShip = groupGenerator.createGroupMember(randomFarm.get(RandomGeneratorFarm.Aspect.MEMBERSHIP_INDEX),
                                    potentialMemberAcc, group.getCreatedDate(),
                                    firstLevelFriends[friendIdx]);
                            if( memberShip != null ) {
                                memberShip.userCreationDate = firstLevelFriends[friendIdx].fromCreationDate;
                                memberShip.setGroupId(group.getGroupId());
                                group.addMember(memberShip);
                            }
                        }
                    }
                } else if (randLevelProb < levelProbs[1]) { // ==> level 2
                    if (secondLevelFriends.size() != 0) {
                        int friendIdx = randomFarm.get(RandomGeneratorFarm.Aspect.MEMBERSHIP_INDEX).nextInt(secondLevelFriends.size());
                        long potentialMemberAcc = secondLevelFriends.get(friendIdx).getFriendAcc();
                        randMemberProb = randomFarm.get(RandomGeneratorFarm.Aspect.MEMBERSHIP).nextDouble();
                        if (randMemberProb < joinProbs[1]) {
                            // Check whether this user has been added and then add to the group
                            if (!memberIds.contains(potentialMemberAcc)) {
                                memberIds.add(potentialMemberAcc);
                                // Assume the earliest membership date is the friendship created date
                                GroupMemberShip memberShip = groupGenerator.createGroupMember(randomFarm.get(RandomGeneratorFarm.Aspect.MEMBERSHIP_INDEX),
                                        potentialMemberAcc, group.getCreatedDate(),
                                        secondLevelFriends.get(friendIdx));
                                if( memberShip != null ) {
                                    memberShip.userCreationDate = secondLevelFriends.get(friendIdx).fromCreationDate;
                                    memberShip.setGroupId(group.getGroupId());
                                    group.addMember(memberShip);
                                }
                            }
                        }
                    }
                } else { // ==> random users
                    // Select a user from window
                    int friendIdx = randomFarm.get(RandomGeneratorFarm.Aspect.MEMBERSHIP_INDEX).nextInt(Math.min(numUserProfilesRead,windowSize));
                    long potentialMemberAcc = reducedUserProfiles[friendIdx].getAccountId();
                    randMemberProb = randomFarm.get(RandomGeneratorFarm.Aspect.MEMBERSHIP).nextDouble();
                    if (randMemberProb < joinProbs[2]) {
                        // Check whether this user has been added and then add to the group
                        if (!memberIds.contains(potentialMemberAcc)) {
                            memberIds.add(potentialMemberAcc);
                            GroupMemberShip memberShip = groupGenerator.createGroupMember(randomFarm.get(RandomGeneratorFarm.Aspect.MEMBERSHIP_INDEX),
                                    potentialMemberAcc, group.getCreatedDate(),
                                    reducedUserProfiles[friendIdx]);
                            if( memberShip != null ) {
                                memberShip.userCreationDate = reducedUserProfiles[friendIdx].getCreationDate();
                                memberShip.setGroupId(group.getGroupId());
                                group.addMember(memberShip);
                            }
                        }
                    }
                }
                numLoop++;
            }

            dataExporter.export(group,user.getCreationDate());
            generatePostForGroup(uniformPostGenerator,group);
            generatePostForGroup(flashmobPostGenerator,group);
        }
    }

    private void generatePostForGroup(PostGenerator postGenerator, Group group) {
        Vector<Post> createdPosts =  postGenerator.createPosts( randomFarm, group, postId);
        postId+=createdPosts.size();
        Iterator<Post> it = createdPosts.iterator();

        while(it.hasNext()) {
            Post groupPost = it.next();
            if (groupPost.getTags() != null) {
            	for (Integer t: groupPost.getTags()){
            		Integer tagClass = tagDictionary.getTagClass(t);
                	Integer tagCount = tagClassCount.containsKey(tagClass) ? tagClassCount.get(tagClass) : 0;
                	tagClassCount.put(tagClass, tagCount+1);
                        tagCount = tagNameCount.containsKey(t) ? tagNameCount.get(t) : 0;
                        tagNameCount.put(t, tagCount + 1);
            	}
            }

            int locationID = ipAddDictionary.getLocation(groupPost.getIpAddress());
            String countryName = locationDictionary.getLocationName(locationID);
            int postCount = postsPerCountry.containsKey(locationID) ? postsPerCountry.get(locationID) : 0;
            postsPerCountry.put(locationID, postCount+1);
            stats.countries.add(countryName);
            dataExporter.export(groupPost,groupPost.getAuthorCreationDate());

            int numComment = randomFarm.get(RandomGeneratorFarm.Aspect.NUM_COMMENT).nextInt(maxNumComments+1);
            ArrayList<Message> replyCandidates = new ArrayList<Message>();
            replyCandidates.add(groupPost);
            for (int j = 0; j < numComment; j++) {
                int replyIndex = randomFarm.get(RandomGeneratorFarm.Aspect.REPLY_TO).nextInt(replyCandidates.size());
                Message replyTo = replyCandidates.get(replyIndex);
                Comment comment = commentGenerator.createComment(randomFarm, postId, groupPost, replyTo, group);
                if ( comment!=null ) {
                    if (!factorTable.containsKey(replyTo.getAuthorId())) {
                    	factorTable.put(replyTo.getAuthorId(), new ReducedUserProfile.Counts());
                    }
                    factorTable.get(replyTo.getAuthorId()).numberOfPostReplies++;

                	locationID = ipAddDictionary.getLocation(comment.getIpAddress());
                	countryName = locationDictionary.getLocationName(locationID);
                	postCount = postsPerCountry.containsKey(locationID) ? postsPerCountry.get(locationID) : 0;
                	postsPerCountry.put(locationID, postCount+1);
                    stats.countries.add(countryName);
                    dataExporter.export(comment,comment.getAuthorCreationDate());
                    if( comment.getTextSize() > 10 ) replyCandidates.add(comment);
                                    	
                	if (comment.getTags() != null) {
                    		for (Integer t: comment.getTags()){
                    			Integer tagClass = tagDictionary.getTagClass(t);
                        		Integer tagCount = tagClassCount.containsKey(tagClass) ? tagClassCount.get(tagClass) : 0;
                        		tagClassCount.put(tagClass, tagCount+1);
                		        tagCount = tagNameCount.containsKey(t) ? tagNameCount.get(t) : 0;
		                        tagNameCount.put(t, tagCount + 1);
                    		}
                	}

                    postId++;
                }
            }
        }
    }

    private long composeUserId( long id, long date, long spid ) {
        long spidMask = ~(0xFFFFFFFFFFFFFFFFL << 7);
        long idMask = ~(0xFFFFFFFFFFFFFFFFL << 33);
//        long dateMask = ~(0xFFFFFFFFFFFFFFFFL << 20);
        //return (((date >> 20) & dateMask) << 40) | ((id & idMask) << 7) | (spid & spidMask);
        long bucket = (long)(256*(date - SN.minDate) / (double)SN.maxDate);
        return (  bucket << 40) | ((id & idMask) << 7) | (spid & spidMask);
    }

    private ReducedUserProfile generateGeneralInformation(int accountId) {
        // User Creation
        long creationDate = dateTimeGenerator.randomDateInMillis( randomFarm.get(RandomGeneratorFarm.Aspect.DATE) );
        int locationId = locationDictionary.getLocationForUser(accountId);
        ReducedUserProfile userProf = new ReducedUserProfile();
        userProf.setCreationDate(creationDate);
        userProf.setGender((randomFarm.get(RandomGeneratorFarm.Aspect.GENDER).nextDouble() > 0.5) ? (byte) 1 : (byte) 0);
        userProf.setBirthDay(dateTimeGenerator.getBirthDay(randomFarm.get(RandomGeneratorFarm.Aspect.BIRTH_DAY), creationDate));
        userProf.setBrowserId(browserDictonry.getRandomBrowserId(randomFarm.get(RandomGeneratorFarm.Aspect.BROWSER)));
        userProf.setLocationId(locationId);
        userProf.setCityId(locationDictionary.getRandomCity(randomFarm.get(RandomGeneratorFarm.Aspect.CITY), locationId));
        userProf.setIpAddress(ipAddDictionary.getRandomIPFromLocation(randomFarm.get(RandomGeneratorFarm.Aspect.IP), locationId));
        userProf.setNumFriends(fbDegreeGenerator.getSocialDegree());
        userProf.setSdpId(fbDegreeGenerator.getIDByPercentile());  	//Generate Id from its percentile in the social degree distribution
        userProf.setNumDimensions(NUM_FRIENDSHIP_HADOOP_JOBS);
        userProf.setAccountId(composeUserId(accountId, creationDate, userProf.getSdpId()));

        // Setting the number of friends and friends per pass
        short totalFriendSet = 0;
        for (int i = 0; i < NUM_FRIENDSHIP_HADOOP_JOBS-1; i++){
            short numPassFriend = (short) Math.floor(friendRatioPerJob[i] * userProf.getNumFriends());
            totalFriendSet = (short) (totalFriendSet + numPassFriend);
            userProf.setNumPassFriends(i,totalFriendSet);
        }
        userProf.setNumPassFriends(NUM_FRIENDSHIP_HADOOP_JOBS-1,userProf.getNumFriends());
        // Setting tags
        int userMainTag = tagDictionary.getaTagByCountry(randomFarm.get(RandomGeneratorFarm.Aspect.TAG_OTHER_COUNTRY), randomFarm.get(RandomGeneratorFarm.Aspect.TAG),userProf.getLocationId());
        userProf.setMainTag(userMainTag);
        short numTags = ((short) randomTagPowerLaw.getValue(randomFarm.get(RandomGeneratorFarm.Aspect.NUM_TAG)));
        userProf.setSetOfTags(topicTagDictionary.getSetofTags(randomFarm.get(RandomGeneratorFarm.Aspect.TOPIC), randomFarm.get(RandomGeneratorFarm.Aspect.TAG_OTHER_COUNTRY), userMainTag, numTags));


        userProf.setUniversityLocationId(unversityDictionary.getRandomUniversity(randomFarm, userProf.getLocationId()));

        // Set whether the user has a smartphone or not.
        userProf.setHaveSmartPhone(randomFarm.get(RandomGeneratorFarm.Aspect.USER_AGENT).nextDouble() > probHavingSmartPhone);
        if (userProf.isHaveSmartPhone()) {
            userProf.setAgentId(userAgentDictionary.getRandomUserAgentIdx(randomFarm.get(RandomGeneratorFarm.Aspect.USER_AGENT)));
        }

        // Compute the popular places the user uses to visit.
        byte numPopularPlaces = (byte) randomFarm.get(RandomGeneratorFarm.Aspect.NUM_POPULAR).nextInt(maxNumPopularPlaces + 1);
        Vector<Short> auxPopularPlaces = new Vector<Short>();
        for (int i = 0; i < numPopularPlaces; i++){
            short aux = popularDictionary.getPopularPlace(randomFarm.get(RandomGeneratorFarm.Aspect.POPULAR),userProf.getLocationId());
            if(aux != -1) {
                auxPopularPlaces.add(aux);
            }
        }

        // Setting popular places
        short popularPlaces[] = new short[auxPopularPlaces.size()];
        Iterator<Short> it = auxPopularPlaces.iterator();
        int i = 0;
        while(it.hasNext()) {
            popularPlaces[i] = it.next();
            ++i;
        }
        userProf.setPopularPlaceIds(popularPlaces);

        // Set random Index used to sort users randomly
        userProf.setRandomId(randomFarm.get(RandomGeneratorFarm.Aspect.RANDOM).nextInt(USER_RANDOM_ID_LIMIT));

        // Set whether the user is a large poster or not.
        userProf.setLargePoster(IsUserALargePoster(userProf));
        return userProf;
    }

    private boolean IsUserALargePoster(ReducedUserProfile user) {
        if(dateTimeGenerator.getBirthMonth(user.getBirthDay()) == GregorianCalendar.JANUARY) {
            return true;
        }
        return false;
    }


    private void setInfoFromUserProfile(ReducedUserProfile user,
                                       UserExtraInfo userExtraInfo) {

        int locationId = (user.getCityId() != -1) ? user.getCityId() : user.getLocationId();
        userExtraInfo.setLocationId(locationId);
        userExtraInfo.setLocation(locationDictionary.getLocationName(locationId));
        double distance = randomFarm.get(RandomGeneratorFarm.Aspect.EXACT_LONG_LAT).nextDouble() * 2;
        userExtraInfo.setLatt(locationDictionary.getLatt(user.getLocationId()) + distance);
        userExtraInfo.setLongt(locationDictionary.getLongt(user.getLocationId()) + distance);
        userExtraInfo.setUniversity(unversityDictionary.getUniversityFromLocation(user.getUniversityLocationId()));

        boolean isMale;
        if (user.getGender() == 1) {
            isMale = true;
            userExtraInfo.setGender(gender[0]); // male
        } else {
            isMale = false;
            userExtraInfo.setGender(gender[1]); // female
        }

        userExtraInfo.setFirstName(namesDictionary.getRandomGivenName(randomFarm.get(RandomGeneratorFarm.Aspect.NAME),
                user.getLocationId(),isMale, dateTimeGenerator.getBirthYear(user.getBirthDay())));
        userExtraInfo.setLastName(namesDictionary.getRandomSurname(randomFarm.get(RandomGeneratorFarm.Aspect.SURNAME),user.getLocationId()));

        // email is created by using the user's first name + userId
        int numEmails = randomFarm.get(RandomGeneratorFarm.Aspect.EXTRA_INFO).nextInt(maxEmails) + 1;
        double prob = randomFarm.get(RandomGeneratorFarm.Aspect.EXTRA_INFO).nextDouble();
        if (prob >= missingRatio) {
            String base = userExtraInfo.getFirstName();
            base = Normalizer.normalize(base,Normalizer.Form.NFD);
            base = base.replaceAll("\\p{InCombiningDiacriticalMarks}+", "");
            base = base.replaceAll(" ", ".");
            base = base.replaceAll("[.]+", ".");
            for (int i = 0; i < numEmails; i++) {
                String email = base + "" + user.getAccountId() + "@" + emailDictionary.getRandomEmail(randomFarm.get(RandomGeneratorFarm.Aspect.TOP_EMAIL),randomFarm.get(RandomGeneratorFarm.Aspect.EMAIL));
                userExtraInfo.addEmail(email);
            }
        }

        // Set class year
        prob = randomFarm.get(RandomGeneratorFarm.Aspect.EXTRA_INFO).nextDouble();
        if ((prob < missingRatio) || userExtraInfo.getUniversity() == -1 ) {
            userExtraInfo.setClassYear(-1);
        } else {
            userExtraInfo.setClassYear(dateTimeGenerator.getClassYear(randomFarm.get(RandomGeneratorFarm.Aspect.DATE),
                    user.getCreationDate(), user.getBirthDay()));
        }

        // Set company and workFrom
        int numCompanies = randomFarm.get(RandomGeneratorFarm.Aspect.EXTRA_INFO).nextInt(maxCompanies) + 1;
        prob = randomFarm.get(RandomGeneratorFarm.Aspect.EXTRA_INFO).nextDouble();
        if (prob >= missingRatio) {
            for (int i = 0; i < numCompanies; i++) {
                long workFrom;
                if (userExtraInfo.getClassYear() != -1) {
                    workFrom = dateTimeGenerator.getWorkFromYear( randomFarm.get(RandomGeneratorFarm.Aspect.DATE),
                            user.getCreationDate(),
                            user.getBirthDay());
                } else {
                    workFrom = dateTimeGenerator.getWorkFromYear(randomFarm.get(RandomGeneratorFarm.Aspect.DATE), userExtraInfo.getClassYear());
                }
                long company = companiesDictionary.getRandomCompany(randomFarm, user.getLocationId());
                userExtraInfo.addCompany(company, workFrom);
                user.addNumOfWorkPlaces(1);
                String countryName = locationDictionary.getLocationName(companiesDictionary.getCountry(company));
                stats.countries.add(countryName);

                GregorianCalendar date = new GregorianCalendar();
                date.setTimeInMillis(workFrom);
                String strWorkFrom = DateGenerator.formatYear(date);
                if (stats.maxWorkFrom == null) {
                    stats.maxWorkFrom = strWorkFrom;
                    stats.minWorkFrom = strWorkFrom;
                } else {
                    if (stats.maxWorkFrom.compareTo(strWorkFrom) < 0) {
                        stats.maxWorkFrom = strWorkFrom;
                    }
                    if (stats.minWorkFrom.compareTo(strWorkFrom) > 0) {
                        stats.minWorkFrom = strWorkFrom;
                    }
                }
            }
        }
        Vector<Integer> userLanguages = languageDictionary.getLanguages(randomFarm.get(RandomGeneratorFarm.Aspect.LANGUAGE),user.getLocationId());
        int internationalLang = languageDictionary.getInternationlLanguage(randomFarm.get(RandomGeneratorFarm.Aspect.LANGUAGE));
        if (internationalLang != -1 && userLanguages.indexOf(internationalLang) == -1) {
            userLanguages.add(internationalLang);
        }
        userExtraInfo.setLanguages(userLanguages);

        stats.maxPersonId = Math.max(stats.maxPersonId, user.getAccountId());
        stats.minPersonId = Math.min(stats.minPersonId, user.getAccountId());
        stats.firstNames.add(userExtraInfo.getFirstName());
        String countryName = locationDictionary.getLocationName(user.getLocationId());
        stats.countries.add(countryName);

        TreeSet<Integer> tags = user.getSetOfTags();
        for (Integer tagID : tags) {
            stats.tagNames.add(tagDictionary.getName(tagID));
            Integer parent = tagDictionary.getTagClass(tagID);
            while (parent != -1) {
                stats.tagClasses.add(tagDictionary.getClassName(parent));
                parent = tagDictionary.getClassParent(parent);
            }
        }
    }


    public double getFriendCreatePro(int i, int j, int pass){
        double prob;
        if (j > i){
            prob = Math.pow(baseProbCorrelated, (j- i));
        } else{
            prob =  Math.pow(baseProbCorrelated, (j + windowSize - i));
        }
        return prob;
    }

    private void createFriendShip(ReducedUserProfile user1, ReducedUserProfile user2, byte pass) {

        long requestedTime = dateTimeGenerator.randomFriendRequestedDate(randomFarm.get(RandomGeneratorFarm.Aspect.DATE),user1, user2);
        byte initiator = (byte) randomFarm.get(RandomGeneratorFarm.Aspect.INITIATOR).nextInt(2);
        long createdTime = -1;
        long declinedTime = -1;
        if (randomFarm.get(RandomGeneratorFarm.Aspect.FRIEND_REJECT).nextDouble() > friendRejectRatio) {
            createdTime = dateTimeGenerator.randomFriendApprovedDate(randomFarm.get(RandomGeneratorFarm.Aspect.DATE),requestedTime);
        } else {
            declinedTime = dateTimeGenerator.randomFriendDeclinedDate(randomFarm.get(RandomGeneratorFarm.Aspect.DATE),requestedTime);
            if (randomFarm.get(RandomGeneratorFarm.Aspect.FRIEND_APROVAL).nextDouble() < friendReApproveRatio) {
                createdTime = dateTimeGenerator.randomFriendReapprovedDate(randomFarm.get(RandomGeneratorFarm.Aspect.DATE),declinedTime);
            }
        }
        if(createdTime != -1) {
            createdTime = createdTime - user1.getCreationDate() >= deltaTime ? createdTime : createdTime + deltaTime;
            createdTime = createdTime - user2.getCreationDate() >= deltaTime ? createdTime : createdTime + deltaTime;
            if (createdTime <= dateTimeGenerator.getEndDateTime()) {
                user2.addNewFriend(new Friend(user2, user1, requestedTime, declinedTime,
                        createdTime, pass, initiator));
                user1.addNewFriend(new Friend(user1, user2, requestedTime, declinedTime,
                        createdTime, pass, initiator));
                friendshipNum++;
            }
        }
    }


    private void updateLastPassFriendAdded(int from, int to, int pass) {
        if (to > windowSize) {
            for (int i = from; i < windowSize; i++) {
                reducedUserProfiles[i].setNumPassFriendsAdded(pass, reducedUserProfiles[i].getNumFriendsAdded());
            }
            for (int i = 0; i < to - windowSize; i++) {
                reducedUserProfiles[i].setNumPassFriendsAdded(pass, reducedUserProfiles[i].getNumFriendsAdded());
            }
        } else {
            for (int i = from; i < to; i++) {
                reducedUserProfiles[i].setNumPassFriendsAdded(pass, reducedUserProfiles[i].getNumFriendsAdded());
            }
        }
    }

    private DataExporter getSerializer(String type, String outputFileName) {
        String t = type.toLowerCase();
        DataExporter.DataFormat format;
        String configFile = new String("");
        if (t.equals("ttl")) {
            format = DataExporter.DataFormat.TURTLE;
        } else if (t.equals("n3")) {
            format = DataExporter.DataFormat.N3;
        } else if (t.equals("csv")) {
            format = DataExporter.DataFormat.CSV;
 //           configFile = new String(CSV_DIRECTORY+"/csv.xml");
        } else if (t.equals("csv_merge_foreign")) {
            format = DataExporter.DataFormat.CSV_MERGE_FOREIGN;
//            configFile = new String(CSV_DIRECTORY+"/csvMergeForeign.xml");
        } else if (t.equals("none")) {
            format = DataExporter.DataFormat.NONE;
        } else {
            System.err.println("Unexpected Serializer - Aborting");
            System.exit(-1);
            return null;
        }
        return new DataExporter(format,sibOutputDir, threadId, dateThreshold,
                exportText,enableCompression,conf.getInt("numUpdatePartitions",1),deltaTime,conf.getBoolean("activity",true),tagDictionary,browserDictonry,companiesDictionary,
                unversityDictionary,ipAddDictionary,locationDictionary,languageDictionary, configFile, factorTable, startMonth, startYear, stats);
    }

    private void writeFactorTable(){
        Configuration conf = new Configuration();
        try {            
        	FileSystem fs = FileSystem.get(conf);
            OutputStream writer = fs.create(new Path(sibOutputDir+"/"+ "m" + threadId + PARAM_COUNT_FILE));
            writer.write(Integer.toString(factorTable.size()).getBytes());
            writer.write("\n".getBytes());

            for (Map.Entry<Long, ReducedUserProfile.Counts> c: factorTable.entrySet()){
            	ReducedUserProfile.Counts count = c.getValue();
            	// correct the group counts
            	//count.numberOfGroups += count.numberOfFriends;
                StringBuffer strbuf = new StringBuffer();
                strbuf.append(c.getKey()); strbuf.append(",");
                String name = medianFirstName.get(c.getKey());
                strbuf.append(name); strbuf.append(",");
                strbuf.append(count.numberOfFriends); 		strbuf.append(",");
                strbuf.append(count.numberOfPosts); 		strbuf.append(",");
                strbuf.append(count.numberOfLikes); 		strbuf.append(",");
                strbuf.append(count.numberOfTagsOfPosts); 	strbuf.append(",");
                strbuf.append(count.numberOfGroups); 		strbuf.append(",");
                strbuf.append(count.numberOfWorkPlaces); 	strbuf.append(",");
                strbuf.append(count.numberOfPostReplies); 	strbuf.append(",");

            	int numBuckets = count.numberOfPostsPerMonth.length;
            	for (int i = 0; i < numBuckets; i++){
            		strbuf.append(count.numberOfPostsPerMonth[i]);
            		strbuf.append(",");
            	}
            	numBuckets = count.numberOfGroupsPerMonth.length;
            	int total = count.numberOfGroupsPerMonth[0];
            	strbuf.append(count.numberOfGroupsPerMonth[0]);
            	for (int i = 1; i < numBuckets; i++){
            		strbuf.append(",");
            		strbuf.append(count.numberOfGroupsPerMonth[i]);
            		total += count.numberOfGroupsPerMonth[i];
            	}
            	strbuf.append("\n");
            	writer.write(strbuf.toString().getBytes());
            }
            writer.write(Integer.toString(postsPerCountry.size()).getBytes());
            writer.write("\n".getBytes());
            for (Map.Entry<Integer, Integer> c: postsPerCountry.entrySet()){
            	StringBuffer strbuf = new StringBuffer();
            	strbuf.append(locationDictionary.getLocationName(c.getKey()));
            	strbuf.append(",");
            	strbuf.append(c.getValue());
            	strbuf.append("\n");
            	writer.write(strbuf.toString().getBytes());
            }
            
            writer.write(Integer.toString(tagClassCount.size()).getBytes());
            writer.write("\n".getBytes());
            for (Map.Entry<Integer, Integer> c: tagClassCount.entrySet()){
            	StringBuffer strbuf = new StringBuffer();
            	strbuf.append(tagDictionary.getClassName(c.getKey()));
            	strbuf.append(",");
            	strbuf.append(tagDictionary.getClassName(c.getKey()));
            	strbuf.append(",");
            	strbuf.append(c.getValue());
            	strbuf.append("\n");
            	writer.write(strbuf.toString().getBytes());
            }
            writer.write(Integer.toString(tagNameCount.size()).getBytes());
            writer.write("\n".getBytes());
            for (Map.Entry<Integer, Integer> c: tagNameCount.entrySet()){
                StringBuffer strbuf = new StringBuffer();
                strbuf.append(tagDictionary.getName(c.getKey()));
                strbuf.append(",");
                //strbuf.append(tagDictionary.getClassName(c.getKey()));
                //strbuf.append(",");
                strbuf.append(c.getValue());
                strbuf.append("\n");
                writer.write(strbuf.toString().getBytes());
            }

            writer.write(Integer.toString(firstNameCount.size()).getBytes());
            writer.write("\n".getBytes());
            for (Map.Entry<String, Integer> c: firstNameCount.entrySet()){
            	StringBuffer strbuf = new StringBuffer();
            	strbuf.append(c.getKey());
            	strbuf.append(",");
            	strbuf.append(c.getValue());
            	strbuf.append("\n");
            	writer.write(strbuf.toString().getBytes());
            }
	    StringBuffer strbuf = new StringBuffer();
	    strbuf.append(startMonth);	strbuf.append("\n");
	    strbuf.append(startYear);   strbuf.append("\n");
	    strbuf.append(stats.minWorkFrom); strbuf.append("\n");
	    strbuf.append(stats.maxWorkFrom); strbuf.append("\n");
            writer.write(strbuf.toString().getBytes());
            writer.flush();
            writer.close();
        } catch (IOException e) {
            System.err.println("Unable to write parameter counts");
            System.err.println(e.getMessage());
            e.printStackTrace();
        }
    }
    
    private void writeStatistics() {
        Gson gson = new GsonBuilder().setExclusionStrategies(stats.getExclusion()).disableHtmlEscaping().create();
        Configuration conf = new Configuration();
        try {
            FileSystem fs = FileSystem.get(conf);
            stats.makeCountryPairs(locationDictionary);
            stats.deltaTime = deltaTime;
            OutputStream writer = fs.create(new Path(sibOutputDir+"/"+ "m" + threadId + STATS_FILE));
            writer.write(gson.toJson(stats).getBytes("UTF8"));
            writer.flush();
            writer.close();
        } catch (IOException e) {
            System.err.println("Unable to write stastistics");
            System.err.println(e.getMessage());
            e.printStackTrace();
        }
    }

} public class TagTextDictionary {
    private static final String SEPARATOR = "  ";
    
    String dicFileName;
    DateGenerator dateGen;
    
    TagDictionary tagDic;
	HashMap<Integer, String> tagText;
	
    double reducedTextRatio;
	
	public TagTextDictionary(String dicFileName, DateGenerator dateGen, TagDictionary tagDic, 
	        double reduceTextRatio){
		this.dicFileName = dicFileName;
		this.tagText = new HashMap<Integer, String>();
		this.dateGen = dateGen;
		this.tagDic = tagDic;
		this.reducedTextRatio = reducedTextRatio;
	}
	
	public void initialize() {
	    try {
	        BufferedReader dictionary = new BufferedReader(new InputStreamReader(getClass( ).getResourceAsStream(dicFileName), "UTF-8"));
	        String line;
	        while ((line = dictionary.readLine()) != null){
	            String[] data = line.split(SEPARATOR);
	            Integer id = Integer.valueOf(data[0]);
	            tagText.put(id, data[1]);
	        }
	        dictionary.close();
	    } catch (Exception e) {
	        e.printStackTrace();
	    }
	}
	
	public String getTagText(int id) {
	    return tagText.get(id);
	}

    public int getRandomTextSize(Random randomTextSize, Random randomReducedText, int minSize, int maxSize ) {

        if (randomReducedText.nextDouble() > reducedTextRatio){
            return randomTextSize.nextInt(maxSize - minSize) + minSize;
        }
        return randomTextSize.nextInt((maxSize >> 1) - minSize) + minSize;
    }

    public int getRandomLargeTextSize( Random randomTextSize, int minSize, int maxSize ) {
        return randomTextSize.nextInt(maxSize - minSize) + minSize;
    }

    public String generateText(Random randomTextSize, TreeSet<Integer> tags, int textSize ) {
        String returnString = "";
        int textSizePerTag = (int)Math.ceil(textSize / (double)tags.size());
        while( returnString.length() < textSize ) {
            Iterator<Integer> it = tags.iterator();
            while (it.hasNext() && returnString.length() < textSize) {
                Integer tag = it.next();
                String content = getTagText(tag);
                int thisTagTextSize = Math.min(textSizePerTag,textSize - returnString.length());
                if (thisTagTextSize >= content.length()) {
                    returnString += content;
                } else {
                    int startingPos = randomTextSize.nextInt(content.length() - thisTagTextSize);
                    String finalString = content.substring(startingPos, startingPos + thisTagTextSize);
//                    String finalString = content.substring(0, thisTagTextSize );
                    String tagName = tagDic.getName(tag).replace("_", " ");
                    tagName = tagName.replace("\"", "\\\"");
                    String prefix = "About " +tagName+ ", ";

                    int posSpace = finalString.indexOf(" ");
                    finalString = (posSpace != -1) ? prefix + finalString.substring(posSpace).trim() : prefix + finalString;
                    posSpace = finalString.lastIndexOf(" ");
                    if (posSpace != -1){
                        finalString = finalString.substring(0, posSpace);
                    }
                    finalString = finalString.substring(0,Math.min(thisTagTextSize,finalString.length()));
                    returnString+=finalString;
                }
                if (!returnString.endsWith(".")) {
                    if( returnString.length() == 1 ) {
                        returnString = ".";
                    } else {
                        returnString = returnString.substring(0,returnString.length() - 1) + ".";
                    }
                }
                if (returnString.length() < textSize-1) {
                    returnString += " ";
                }
            }
        }
        return returnString.replace("|", " ");
    }
}
 public class FlashmobTagDictionary {

    private static final String SEPARATOR = "\t";
    
    DateGenerator dateGen;                              /**< @brief The date generator used to generate dates.*/
	PowerDistGenerator levelGenerator;                  /**< @brief The powerlaw distribution generator used to generate the levels.*/
	Random random;                                      /**< @brief A uniform random genereator.*/
	TagDictionary tagDictionary;                        /**< @brief The tag dictionary used to create the flashmob tags.*/
	HashMap<Integer,Vector<FlashmobTag> > flashmobTags; /**< @brief A map of identifiers of tags to flashmob tag instances.*/
	FlashmobTag[] flashmobTagCumDist;                   /**< @brief The cumulative distribution of flashmob tags sorted by date.*/
	double flashmobTagsPerMonth;                        /**< @brief The number of flashmob tags per month.*/
	double probInterestFlashmobTag;                     /**< @brief The probability to take an interest flashmob tag.*/
    double probRandomPerLevel;                          /**< @brief The probability per level to take a flashmob tag.*/

	public FlashmobTagDictionary( TagDictionary tagDictionary, 
								  DateGenerator dateGen,
								  int flashmobTagsPerMonth,
								  double probInterestFlashmobTag,
								  double probRandomPerLevel,
								  double flashmobTagMinLevel,
								  double flashmobTagMaxLevel,
								  double flashmobTagDistExp,
								  long seed ) {

		this.tagDictionary = tagDictionary;	    
		this.dateGen = dateGen;
		this.levelGenerator  = new PowerDistGenerator(flashmobTagMinLevel, flashmobTagMaxLevel, flashmobTagDistExp); 
		this.random = new Random(seed);
		this.flashmobTags = new HashMap<Integer,Vector<FlashmobTag> >();
		this.flashmobTagsPerMonth = flashmobTagsPerMonth;
		this.probInterestFlashmobTag = probInterestFlashmobTag;
        this.probRandomPerLevel = probRandomPerLevel;
	}

    /** @brief Initializes the flashmob tag dictionary, by selecting a set of tags as flashmob tags.*/
    public void initialize() {
    	int numFlashmobTags = (int)(flashmobTagsPerMonth * dateGen.numberOfMonths(dateGen.getStartDateTime()));
    	Integer[] tags = tagDictionary.getRandomTags(random,numFlashmobTags);
    	flashmobTagCumDist = new FlashmobTag[numFlashmobTags];
    	double sumLevels = 0;
    	for( int i = 0; i < numFlashmobTags; ++i ){
    		Vector<FlashmobTag> instances = flashmobTags.get(tags[i]);
    		if( instances == null ) {
    			instances = new Vector<FlashmobTag>();	
    			flashmobTags.put(tags[i],instances);
    		}
   			FlashmobTag flashmobTag = new FlashmobTag();
   			flashmobTag.date = dateGen.randomDateInMillis(random,dateGen.getStartDateTime(), dateGen.getEndDateTime());
   			flashmobTag.level = levelGenerator.getValue(random);
   			sumLevels += flashmobTag.level;
   			flashmobTag.tag = tags[i];
    		instances.add(flashmobTag);
    		flashmobTagCumDist[i] = flashmobTag;
    	}
    	Arrays.sort(flashmobTagCumDist);
    	int size = flashmobTagCumDist.length;
    	double currentProb = 0.0;
    	for( int i = 0; i < size; ++i ) {
    		flashmobTagCumDist[i].prob = currentProb;
    		currentProb += (double)(flashmobTagCumDist[i].level) / (double)(sumLevels);
    	}
        System.out.println("Number of flashmob tags generated: "+numFlashmobTags);
    }

    /** @brief Selects the earliest flashmob tag index from a given date.
     *  @param[in] fromDate The minimum date to consider.
     *  @return The index to the earliest flashmob tag.*/
    private int searchEarliestIndex( long fromDate ) {
            int lowerBound = 0;
            int upperBound = flashmobTagCumDist.length;
            int midPoint = (upperBound + lowerBound)  / 2;
            while (upperBound > (lowerBound+1)){
                if (flashmobTagCumDist[midPoint].date > fromDate ){
                    upperBound = midPoint;
                } else{
                    lowerBound = midPoint; 
                }
                midPoint = (upperBound + lowerBound)  / 2;
            }
            return midPoint;
    }

    /** @brief Makes a decision of selecting or not a flashmob tag.
     *  @param[in] index The index of the flashmob tag to select.
     *  @return true if the flashmob tag is selected. false otherwise.*/
    private boolean selectFlashmobTag(Random rand,  int index ) {
            return rand.nextDouble() > (1-probRandomPerLevel*flashmobTagCumDist[index].level);
    }

    /** @brief Given a set of interests and a date, generates a set of flashmob tags.
     *  @param[in] interests The set of interests.
     *  @param[in] fromDate The date from which to consider the flashmob tags.
     *  @return A vector containing the selected flashmob tags.*/
	public Vector<FlashmobTag> generateFlashmobTags( Random rand, TreeSet<Integer> interests, long fromDate ) {
		Vector<FlashmobTag> result = new Vector<FlashmobTag>();
		Iterator<Integer> it = interests.iterator();
		while(it.hasNext()) {
			Integer tag = it.next();
			Vector<FlashmobTag> instances = flashmobTags.get(tag);
			if( instances != null ) {
				Iterator<FlashmobTag> it2 = instances.iterator();
				while( it2.hasNext()){
					FlashmobTag instance = it2.next();
					if( instance.date >= fromDate ) {
						if(rand.nextDouble() > 1 - probInterestFlashmobTag){
                            result.add(instance);
						}
					} 
				}
			}
		}
		int numberOfMonths = (int)(dateGen.numberOfMonths(fromDate));	
		int earliestIndex = searchEarliestIndex(fromDate);
		for( int i = earliestIndex; i < flashmobTagCumDist.length; ++i ) {
				if(selectFlashmobTag(rand, i)) {
                    result.add(flashmobTagCumDist[i]); 
                }
		}
		return result;
	}

    public FlashmobTag[] getFlashmobTags() {
        return flashmobTagCumDist;
    }
}
 public class UserAgentDictionary {
    
	String fileName;
	
	Vector<String> userAgents;
	double probSentFromAgent; 
	
	public UserAgentDictionary(String fileName, double probSentFromAgent){
		this.fileName = fileName; 
		this.probSentFromAgent = probSentFromAgent;
	}
	
	public void init(){
		try {
		    userAgents = new Vector<String>();
			BufferedReader agentFile = new BufferedReader(new InputStreamReader(getClass().getResourceAsStream(fileName), "UTF-8"));
			String line; 
			while ((line = agentFile.readLine()) != null) {
			    userAgents.add(line.trim());
            }
			agentFile.close();
			System.out.println("Done ... " + userAgents.size() + " agents have been extracted ");
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
	public String getUserAgentName(Random randomSent, boolean hasSmathPhone, byte agentId){
		return (hasSmathPhone && (randomSent.nextDouble() > probSentFromAgent)) ? userAgents.get(agentId) : "";
	}
	
	public byte getRandomUserAgentIdx(Random random){
		return (byte)random.nextInt(userAgents.size());
	}	
} public class IPAddressDictionary {
    
    private static final String SEPARATOR_COUNTRY = "   ";
    private static final String SEPARATOR_IP = "[.]";
    private static final String SEPARATOR_MASK = "/";
    private static final int MAX_IP_COUNTRY = 100;
    
    
    
	HashMap<Integer, Vector<IP>> ipsByCountry;
	HashMap<Integer, Integer> ipCountry;
	
	LocationDictionary locationDic;
	
	String 	mappingFileName;
	String 	baseIPdir;
	
	double probDiffIPinTravelSeason;
    double probDiffIPnotTravelSeason;
    double probDiffIPforTraveller;
	
	public IPAddressDictionary(String _mappingFileName, String _baseIPdir, LocationDictionary locationDic, 
								double _probDiffIPinTravelSeason, 
								double _probDiffIPnotTravelSeason, double _probDiffIPforTraveller){
		this.mappingFileName = _mappingFileName;
		this.baseIPdir = _baseIPdir;
		
		this.locationDic = locationDic;
		ipCountry = new HashMap<Integer, Integer>();
		ipsByCountry = new HashMap<Integer, Vector<IP>>();
		
		probDiffIPinTravelSeason = _probDiffIPinTravelSeason; 
		probDiffIPnotTravelSeason = _probDiffIPnotTravelSeason;
		probDiffIPforTraveller = _probDiffIPforTraveller;
	}
	
	public void initialize() {
	    String line;
	    HashMap<String, String> countryAbbreMap = new HashMap<String, String>();
	    try {
	        BufferedReader mappingFile = new BufferedReader(new InputStreamReader(getClass().getResourceAsStream(mappingFileName), "UTF-8"));
	        while ((line = mappingFile.readLine()) != null){
	            String data[] = line.split(SEPARATOR_COUNTRY);
	            String abbr = data[0];
	            String countryName = data[1].trim().replace(" ", "_");
	            countryAbbreMap.put(countryName, abbr);
	        }
	        mappingFile.close();

	        Vector<Integer> countries = locationDic.getCountries();
	        for (int i = 0; i < countries.size(); i ++) {
	            ipsByCountry.put(countries.get(i), new Vector<IP>());

	            //Get the name of file
	            String fileName = countryAbbreMap.get(locationDic.getLocationName(countries.get(i)));
	            fileName = baseIPdir + "/" + fileName + ".zone";
	            BufferedReader ipZoneFile = new BufferedReader(new InputStreamReader(getClass( ).getResourceAsStream(fileName), "UTF-8"));

	            int j = 0;
	            while ((line = ipZoneFile.readLine()) != null && (j < MAX_IP_COUNTRY)) {
	                String data[] = line.split(SEPARATOR_IP);
	                String maskData[] = data[3].split(SEPARATOR_MASK);
	                int byte1 = Integer.valueOf(data[0]);
	                int byte2 = Integer.valueOf(data[1]);
	                int byte3 = Integer.valueOf(data[2]);
	                int byte4 = Integer.valueOf(maskData[0]);
	                int maskNum = Integer.valueOf(maskData[1]);

	                IP ip = new IP(byte1, byte2, byte3, byte4, maskNum);

	                ipsByCountry.get(i).add(ip);
	                ipCountry.put(ip.getIp() & ~ip.getMask(), i);
	                j++;
	            }
	            ipZoneFile.close();
	        } 
	    } catch (IOException e) {
	        e.printStackTrace();
	    }
	}

	public int getLocation(IP ip) {
	    int network = ip.getIp() & ~ip.getMask();
	    return (ipCountry.containsKey(network)) ? ipCountry.get(network) : -1;
	}
	
	public IP getRandomIPFromLocation(Random random, int locationIdx) {
        while(locationDic.getType(locationIdx) != "country") {
            locationIdx = locationDic.belongsTo(locationIdx);
        }
		Vector<IP> countryIPs = ipsByCountry.get(locationIdx);
		int idx = random.nextInt(countryIPs.size());
		
		IP networkIp = countryIPs.get(idx);
		
		int formattedIP = 0;
		int ip = networkIp.getIp();
		int mask = networkIp.getMask();
		
		for (int i = 0; i < IP.IP4_SIZE_BYTES; i++) {
		    int randomRange = ((mask >>> (IP.BYTE_SIZE * i)) & 0xFF) + 1;
		    int base = ((ip >>> (IP.BYTE_SIZE * i)) & 0xFF) + random.nextInt(randomRange);
		    formattedIP = formattedIP | (base << IP.BYTE_SIZE * i);
        }
		
		return new IP(formattedIP, mask);
	}
	
	public IP getRandomIP(Random random) {
	    Vector<Integer> countries = locationDic.getCountries();
        int randomLocationIdx = random.nextInt(countries.size());
		return getRandomIPFromLocation(random, randomLocationIdx);
	}
	
	private boolean changeUsualIp(Random randomDiffIP, Random randomDiffIPForTravelers, boolean isFrequentChange, long date) {
        double diffIpProb = randomDiffIP.nextDouble();
        double diffIpForTravelersProb = randomDiffIPForTravelers.nextDouble();
        boolean isTravelSeason = DateGenerator.isTravelSeason(date);
        if ( (isFrequentChange && diffIpProb < probDiffIPforTraveller ) ||
             (!isFrequentChange && ((isTravelSeason && diffIpForTravelersProb < probDiffIPinTravelSeason) ||
                                   (!isTravelSeason && diffIpForTravelersProb < probDiffIPnotTravelSeason)))) {
                 return true;
        }
        return false;
	}
	
	public IP getIP(Random randomIP, Random randomDiffIP, Random randomDiffIPForTravelers, IP ip, boolean isFrequentChange, long date) {
	    return (changeUsualIp(randomDiffIP, randomDiffIPForTravelers, isFrequentChange, date)) ? new IP(ip.getIp(), ip.getMask()) : getRandomIP(randomIP);
	}
	
	public IP getIP(Random randomIP, Random randomDiffIP, Random randomDiffIPForTravelers, IP ip, boolean isFrequentChange, long date, int countryId) {
        return (changeUsualIp(randomDiffIP, randomDiffIPForTravelers, isFrequentChange, date)) ? new IP(ip.getIp(), ip.getMask()) : getRandomIPFromLocation(randomIP, countryId);
    }
} public class BrowserDictionary {
	
    private static final String SEPARATOR = "  ";
    
    Vector<String>      vBrowser;
	Vector<Double>  	vBrowserCummulative; 

	String fileName;
	
	double probAnotherBrowser;
	/**
	 * Creator.
	 * 
	 * @param fileName: The file which contains the browser data.
	 * @param probAnotherBrowser: Probability of the user using another browser.
	 */
	public BrowserDictionary(String fileName, double probAnotherBrowser){
		this.fileName = fileName;
		this.probAnotherBrowser = probAnotherBrowser;
	}
	
	/**
	 * Initializes the dictionary extracting the data from the file.
	 */
	public void init(){
		try {
		    BufferedReader dictionary = new BufferedReader(
		            new InputStreamReader(getClass( ).getResourceAsStream(fileName), "UTF-8"));
			vBrowser = new Vector<String>();
			vBrowserCummulative = new Vector<Double>();
			
			String line; 
			double cummulativeDist = 0.0;

			while ((line = dictionary.readLine()) != null){
			    String data[] = line.split(SEPARATOR);
			    String browser = data[0];
			    cummulativeDist += Double.parseDouble(data[1]);
			    vBrowser.add(browser);
			    vBrowserCummulative.add(cummulativeDist);
			}			
			dictionary.close();
			System.out.println("Done ... " + vBrowser.size() + " browsers were extracted");
			
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
	/**
	 * Gets the browser name.
	 */
	public String getName(byte id) {
        return vBrowser.get(id);
    }
	
	/**
	 * Gets a random browser id.
	 */
	public byte getRandomBrowserId(Random random) {
	    double prob = random.nextDouble();
		int minIdx = 0;
		int maxIdx = (prob < vBrowserCummulative.get(minIdx)) ? minIdx : vBrowserCummulative.size() - 1;
		while ((maxIdx - minIdx) > 1) {
			
		    int middlePoint = minIdx + (maxIdx - minIdx) / 2;
			if (prob > vBrowserCummulative.get(middlePoint)) {
				minIdx = middlePoint;
			} else {
				maxIdx = middlePoint;
			}
		}
		return (byte)maxIdx;
	}

	/**
	 * Gets the post browser. There is a chance of being different from the user preferred browser
	 * @param userBrowserId: The user preferred browser.
	 */
	public byte getPostBrowserId(Random randomDiffBrowser, Random randomBrowser, byte userBrowserId){
		double prob = randomDiffBrowser.nextDouble();
		return (prob < probAnotherBrowser) ? getRandomBrowserId(randomBrowser) : userBrowserId;
	}
} public class TagMatrix {
    
    private static final String SEPARATOR = " ";
    
    String dicFileName;
    
    ArrayList<ArrayList<Double>> vecCumulative;
    ArrayList<ArrayList<Integer>> vecTopicID;
    TreeMap<Integer,ArrayList<Integer>>  auxMatrix;
    ArrayList<Integer> tagList;
    
    public TagMatrix(String dicFileName, int numCelebrities){
        this.dicFileName = dicFileName;
        vecCumulative = new ArrayList<ArrayList<Double>>(numCelebrities);
        vecTopicID    = new ArrayList<ArrayList<Integer>>(numCelebrities);
        for (int i =  0; i < numCelebrities; i++){
            vecCumulative.add(new ArrayList<Double>());
            vecTopicID.add(new ArrayList<Integer>());
        }
        auxMatrix = new TreeMap<Integer,ArrayList<Integer>>();
        tagList = new ArrayList<Integer>();
    }
    
    public void initMatrix() {
        try {
            BufferedReader dictionary = new BufferedReader(new InputStreamReader(getClass().getResourceAsStream(dicFileName), "UTF-8"));
            String line;
            while ((line = dictionary.readLine()) != null) {
                String data[] = line.split(SEPARATOR);
                int celebrityId = Integer.parseInt(data[0]);
                int topicId = Integer.parseInt(data[1]);
                double cumuluative = Double.parseDouble(data[2]);
                
                vecCumulative.get(celebrityId).add(cumuluative);
                vecTopicID.get(celebrityId).add(topicId);
                Insert(celebrityId,topicId);
            }
            dictionary.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void Insert(Integer tag1, Integer tag2) {
        ArrayList<Integer> vect = auxMatrix.get(tag1);
        if( vect == null ) {
           vect = new ArrayList<Integer> ();
           auxMatrix.put(tag1,vect);
           tagList.add(tag1);
        }
        vect.add(tag2);

        vect = auxMatrix.get(tag2);
        if( vect == null ) {
            vect = new ArrayList<Integer> ();
            auxMatrix.put(tag2,vect);
            tagList.add(tag1);
        }
        vect.add(tag1);
    }

    public Integer getRandomRelated( Random randomTopic, Random randomTag, int tag) {
        ArrayList<Integer> vect = auxMatrix.get(tag);
        if( vect != null ) {
            int index = randomTag.nextInt(vect.size());
            return vect.get(index);
        } else {
            return tagList.get(randomTag.nextInt(tagList.size()));
        }
    }
    
    // Combine the main tag and related tags
    public TreeSet<Integer> getSetofTags(Random randomTopic, Random randomTag, int celebrityId, int numTags){
        TreeSet<Integer> resultTags = new TreeSet<Integer>();
        resultTags.add(celebrityId);
        while (resultTags.size() < numTags) {
            int tagId; 
            tagId = celebrityId; 
            
            while (vecTopicID.get(tagId).size() == 0) {
                tagId = randomTopic.nextInt(vecTopicID.size());
            }

            // Doing binary search for finding the tag
            double randomDis = randomTag.nextDouble();
            int lowerBound = 0;
            int upperBound = vecTopicID.get(tagId).size();
            int midPoint = (upperBound + lowerBound)  / 2;

            while (upperBound > (lowerBound+1)){
                if (vecCumulative.get(tagId).get(midPoint) > randomDis ){
                    upperBound = midPoint;
                } else{
                    lowerBound = midPoint; 
                }
                midPoint = (upperBound + lowerBound)  / 2;
            }
            resultTags.add(vecTopicID.get(tagId).get(midPoint));
        }
        return resultTags;    
    }
}
 public class Comment extends Message {

	private long postId;			/**< @brief The post identifier of the replied post.**/
	private	long replyOf;			/**< @brief The id of the parent post/comment of this comment.**/

	public Comment( long commentId,
                 String content,
                 int textSize,
                 long creationDate,
                 long authorId,
                 long authorCreationDate,
                 long groupId,
                 TreeSet<Integer> tags,
                 IP ipAddress,
                 String userAgent,
                 byte browserIdx,
                 int locationId,
                 long postId,
                 long replyOf
    ) {

        super(commentId, content, textSize, creationDate, authorId, authorCreationDate, groupId, tags, ipAddress, userAgent, browserIdx, locationId);
        this.postId = postId;
        this.replyOf = replyOf;
	}

	public long getReplyOf() {
		return replyOf;
	}
	public void setReplyOf(long replyOf) {
		this.replyOf = replyOf;
	}
	public long getPostId() {
		return postId;
	}
	public void setPostId(long postId) {
		this.postId = postId;
	}
} public class Friend implements Serializable{
    private static final long serialVersionUID = 3657773293974543890L;
    long from;
	long to;
    public long dependantDate;
    public long fromCreationDate;
    public long toCreationDate;
	long createdTime;			//approved Time 
	long requestTime;			 
	long declinedTime; 
	byte passIdx;
	byte initiator; 			// 0: if user with smaller Id initiate the relationship, 1: if else
	
	
	//For user's agent information
	boolean				isHaveSmartPhone; 		// Use for providing the user agent information
	byte 				agentIdx; 				// Index of user agent in the dictionary, e.g., 0 for iPhone, 1 for HTC
	byte				browserIdx;				// Index of web browser, e.g., 0 for Internet Explorer
	
	boolean 			isFrequentChange;		
	IP	 				sourceIp; 				// Source IP address of the friend

	boolean 			isLargePoster;			// True if friend is a large poster.
	
	
/*	public Friend(long from, long to, long _requestedTime, long _declinedTime, long _createdTime, byte passidx, byte initiator, boolean isLargePoster){
        this.from = from;
		this.to = to;
		this.requestTime = _requestedTime;
		this.declinedTime = _declinedTime;
		this.createdTime = _createdTime; 
		this.passIdx = passidx; 
		this.initiator = initiator;
		this.isLargePoster = isLargePoster;
	}
	*/
	public Friend(){}
	
	public void readFields(DataInput arg0) throws IOException{
        from = arg0.readLong();
		to = arg0.readLong();
        dependantDate = arg0.readLong();
        fromCreationDate = arg0.readLong();
        toCreationDate = arg0.readLong();
		createdTime = arg0.readLong();
		requestTime = arg0.readLong();
		declinedTime = arg0.readLong(); 
		passIdx = arg0.readByte(); 
		initiator = arg0.readByte(); 
		
		isHaveSmartPhone = arg0.readBoolean();
		agentIdx = arg0.readByte();
		browserIdx = arg0.readByte(); 
		isFrequentChange = arg0.readBoolean();
		int ip = arg0.readInt();
		int mask = arg0.readInt();
		this.sourceIp = new IP(ip, mask); 
		this.isLargePoster = arg0.readBoolean();
	}
	public void write(DataOutput arg0) throws IOException {
		arg0.writeLong(from);
        arg0.writeLong(to);
        arg0.writeLong(dependantDate);
        arg0.writeLong(fromCreationDate);
        arg0.writeLong(toCreationDate);
		arg0.writeLong(createdTime);
		arg0.writeLong(requestTime);
		arg0.writeLong(declinedTime);
		arg0.writeByte(passIdx);
		arg0.writeByte(initiator);
		
		arg0.writeBoolean(isHaveSmartPhone);
		arg0.writeByte(agentIdx);
		arg0.writeByte(browserIdx); 
		arg0.writeBoolean(isFrequentChange);
		arg0.writeInt(sourceIp.getIp());
		arg0.writeInt(sourceIp.getMask());
		arg0.writeBoolean(isLargePoster);
	}

	public Friend(ReducedUserProfile from, ReducedUserProfile to, long _requestedTime, long _declinedTime, long _createdTime,
	        byte passidx, byte initiator){
	    this.from = from.getAccountId();
        this.to = to.getAccountId();
        this.dependantDate = Math.max(from.getCreationDate(), to.getCreationDate());
        this.fromCreationDate = from.getCreationDate();
        this.toCreationDate = to.getCreationDate();
	    this.requestTime = _requestedTime;
	    this.declinedTime = _declinedTime;
	    this.createdTime = _createdTime; 
	    this.passIdx = passidx; 
	    this.initiator = initiator;

	    this.isHaveSmartPhone = to.isHaveSmartPhone;
	    this.agentIdx = to.getAgentId();
	    this.browserIdx = to.getBrowserIdx();
	    this.isFrequentChange = to.isFrequentChange;
	    this.setSourceIp(to.getIpAddress());
	    this.isLargePoster = to.isLargePoster();
	}	
	
	
	public long getFriendAcc() {
		return to;
	}
	public void setFriendAcc(long friendAcc) {
		this.to = friendAcc;
	}

    public long getUserAcc() {
        return from;
    }

    public void setUserAcc(long from) {
        this.from = from;
    }
	public long getCreatedTime() {
		return createdTime;
	}
	public void setCreatedTime(long createdTime) {
		this.createdTime = createdTime;
	}
	public int getPassIdx() {
		return passIdx;
	}
	public void setPassIdx(byte passIdx) {
		this.passIdx = passIdx;
	} 
	public long getRequestTime() {
		return requestTime;
	}
	public void setRequestTime(long requestTime) {
		this.requestTime = requestTime;
	}
	public long getDeclinedTime() {
		return declinedTime;
	}
	public void setDeclinedTime(long declinedTime) {
		this.declinedTime = declinedTime;
	}
	public byte getInitiator() {
		return initiator;
	}
	public void setInitiator(byte initiator) {
		this.initiator = initiator;
	}
	public IP getSourceIp() {
		return sourceIp;
	}

	public void setSourceIp(IP sourceIp) {
		this.sourceIp = sourceIp;
	}

	public boolean isHaveSmartPhone() {
		return isHaveSmartPhone;
	}

	public void setHaveSmartPhone(boolean isHaveSmartPhone) {
		this.isHaveSmartPhone = isHaveSmartPhone;
	}

	public byte getAgentIdx() {
		return agentIdx;
	}

	public void setAgentIdx(byte agentIdx) {
		this.agentIdx = agentIdx;
	}

	public byte getBrowserIdx() {
		return browserIdx;
	}

	public void setBrowserIdx(byte browserIdx) {
		this.browserIdx = browserIdx;
	}

	public boolean isFrequentChange() {
		return isFrequentChange;
	}

	public void setFrequentChange(boolean isFrequentChange) {
		this.isFrequentChange = isFrequentChange;
	}

	public boolean isLargePoster() {
		return this.isLargePoster;
	}

	public void setLargePoster(boolean isLargePoster) {
		this.isLargePoster = isLargePoster;
	}

} public class Group {
	long groupId;
	long moderatorId; 		//creator Id
	long createdDate;
//    int forumWallId;

	//int forumStatusId;
	
	String groupName; 

	Integer[] tags;
	
	int locationIdx; 			// Each group is for one location which is the creator's location
	
	GroupMemberShip memberShips[]; 
	int numMemberAdded = 0; 

	public void initAllMemberships(int numMembers){
		memberShips = new GroupMemberShip[numMembers];
	}
	public void addMember(GroupMemberShip member){
		memberShips[numMemberAdded] = member;
		numMemberAdded++;
	}
	public long getGroupId() {
		return groupId;
	}

	public void setGroupId(long groupId) {
		this.groupId = groupId;
	}
	public long getModeratorId() {
		return moderatorId;
	}
	public void setModeratorId(long moderatorId) {
		this.moderatorId = moderatorId;
	}
	public long getCreatedDate() {
		return createdDate;
	}
	public void setCreatedDate(long createdDate) {
		this.createdDate = createdDate;
	} 
	public Integer[] getTags() {
		return tags;
	}
	public void setTags(Integer[] tags) {
		this.tags = tags;
	}
	public String getGroupName() {
		return groupName;
	}
	public void setGroupName(String groupName) {
		this.groupName = groupName;
	}
	public GroupMemberShip[] getMemberShips() {
		return memberShips;
	}
	public void setMemberShips(GroupMemberShip[] memberShips) {
		this.memberShips = memberShips;
	}
/*	public int getForumWallId() {
		return forumWallId;
	}
	public void setForumWallId(int forumWallId) {
		this.forumWallId = forumWallId;
	}
	*/
	/*public int getForumStatusId() {
		return forumStatusId;
	}
	public void setForumStatusId(int forumStatusId) {
		this.forumStatusId = forumStatusId;
	}
	*/
	public int getNumMemberAdded() {
		return numMemberAdded;
	}
	public void setNumMemberAdded(int numMemberAdded) {
		this.numMemberAdded = numMemberAdded;
	}
	public int getLocationIdx() {
		return locationIdx;
	}
	public void setLocationIdx(int locationIdx) {
		this.locationIdx = locationIdx;
	}
} public class GroupMemberShip {
    long    groupId;
	long	userId;
	long 	joinDate;
    public long    userCreationDate;
	IP      ip;
	byte    browserIdx;
	byte    agentIdx;
    boolean isLargePoster = false;       //This is used for creating large posts.
	boolean isFrequentChange;
	boolean isHaveSmartPhone;

    public long getGroupId() {
        return groupId;
    }

    public void setGroupId( long groupId ) {
       this.groupId = groupId;
    }

	public long getUserId() {
		return userId;
	}
	public void setUserId(long userId) {
		this.userId = userId;
	}
	
	public byte getAgentIdx() {
        return agentIdx;
    }

    public void setAgentIdx(byte agentIdx) {
        this.agentIdx = agentIdx;
    }
	
	public byte getBrowserIdx() {
        return browserIdx;
    }

    public void setBrowserIdx(byte browserIdx) {
        this.browserIdx = browserIdx;
    }
	
	public IP getIP() {
	    return ip;
	}
	
	public void setIP(IP ip) {
	    this.ip = ip;
	}
	
	public boolean isFrequentChange() {
        return isFrequentChange;
    }

    public void setFrequentChange(boolean isFrequentChange) {
        this.isFrequentChange = isFrequentChange;
    }
    
    public boolean isHaveSmartPhone() {
        return isHaveSmartPhone;
    }

    public void setHaveSmartPhone(boolean isHaveSmartPhone) {
        this.isHaveSmartPhone = isHaveSmartPhone;
    }
	
	public long getJoinDate() {
		return joinDate;
	}
	public void setJoinDate(long joinDate) {
		this.joinDate = joinDate;
	}

    public boolean isLargePoster() {
        return this.isLargePoster;
    }

    public void setLargePoster( boolean isLargePoster ) {
        this.isLargePoster = isLargePoster;
    } 
} public class Post extends  Message {

    private int language;                       /**< @brief The language used in the post.*/

    public Post( long postId,
                 String content,
                 int textSize,
                 long creationDate,
                 long authorId,
                 long authorCreationDate,
                 long groupId,
                 TreeSet<Integer> tags,
                 IP ipAddress,
                 String userAgent,
                 byte browserIdx,
                 int locationId,
                 int language
                ) {
        super(postId, content, textSize, creationDate, authorId, authorCreationDate, groupId, tags, ipAddress, userAgent, browserIdx,locationId );
        this.language = language;
    }
    public int getLanguage() {
        return this.language;
    }
    public void setLanguage(int language) {
        this.language = language;
    }
} public class ReducedUserProfile implements Serializable, Writable{
	private static final long serialVersionUID = 3657773293974543890L;
	long 				accountId;
    int					sdpId;
	long	 			creationDate;
	public short 		numFriends;
	public short 		numFriendsAdded;
	public byte			numCorDimensions; 
	public short 		numPassFriends[];
	public short 		numPassFriendsAdded[];
	Friend 				friendList[];
	TreeSet<Long>	    friendIds;
	int					dicElementIds[];	// Id of an element in a dictionary, e.g., locationId
										// interestId
	
	//For user's agent information
	boolean				isHaveSmartPhone; 		// Use for providing the user agent information
	byte 				agentIdx; 				// Index of user agent in the dictionary, e.g., 0 for iPhone, 1 for HTC
	byte				browserIdx;				// Index of web browser, e.g., 0 for Internet Explorer
	
	//For IP address
	boolean 			isFrequentChange;		// About 1% of users frequently change their location
	IP					ipAddress;				// IP address
	
	
	// Store redundant info
	int 				locationId;
	int                 cityIdx;
	long 				forumWallId;
	TreeSet<Integer> 	setOfTags;
    int                 mainTag;
	
	short				popularPlaceIds[]; 
	byte				numPopularPlace;
	
	// For organization dimension
	int 				universityLocationId;
    byte				gender; 
	long				birthDay;

	// For posting
	boolean 			isLargePoster;
	
	static public class Counts {
		public int numberOfPosts;
		public int[] numberOfPostsPerMonth;
		public int[] numberOfGroupsPerMonth;
		public int numberOfLikes;
		public int numberOfGroups;
		public int numberOfWorkPlaces;
		public int numberOfTagsOfPosts;
		public int numberOfFriends;
		public int numberOfPostReplies;
		
		public Counts(Counts other){
			this.numberOfPosts = other.numberOfPosts;
			this.numberOfPostsPerMonth = other.numberOfPostsPerMonth;
			this.numberOfLikes = other.numberOfLikes;
			this.numberOfGroups = other.numberOfGroups;
			this.numberOfWorkPlaces = other.numberOfWorkPlaces;
			this.numberOfTagsOfPosts = other.numberOfTagsOfPosts;
			this.numberOfFriends = other.numberOfFriends;
		}
		public Counts(){
			this.numberOfPosts = 0;
			this.numberOfPostsPerMonth = new int[36+1];
			this.numberOfGroupsPerMonth = new int[36+1];
			this.numberOfLikes = 0;
			this.numberOfGroups = 0;
			this.numberOfWorkPlaces = 0;
			this.numberOfTagsOfPosts = 0;
			this.numberOfFriends = 0;
		}
	}

	Counts stats;
	
	public void clear(){
		Arrays.fill(friendList,null);
		friendList = null;
		friendIds.clear();
		friendIds = null;
		numPassFriends = null; 
		numPassFriendsAdded = null; 
		dicElementIds = null; 
		setOfTags.clear();
		setOfTags = null;
		popularPlaceIds = null; 
		stats = null;
	}
	
	private void readObject(java.io.ObjectInputStream stream)
			 throws IOException, ClassNotFoundException{
			accountId = stream.readLong();
            sdpId = stream.readInt();
			creationDate = stream.readLong();
			numFriends = stream.readShort();
			numFriendsAdded = stream.readShort();
			numCorDimensions = stream.readByte();
			numPassFriends = new short[numCorDimensions];
			for (int i = 0; i < numCorDimensions; i ++){
				numPassFriends[i] = stream.readShort();
			}
			numPassFriendsAdded = new short[numCorDimensions];
			for (int i = 0; i < numCorDimensions; i ++){
				numPassFriendsAdded[i] = stream.readShort();
			}
			friendList = new Friend[numFriends];
			friendIds = new TreeSet<Long>();
			for (int i = 0; i < numFriendsAdded; i++){
				Friend fr = new Friend(); 
				fr.readFields(stream);
				friendList[i] = fr; 
			}
			//Read the size of Treeset first
			int size = stream.readInt(); 
			for (int i = 0; i < size; i++){
				friendIds.add(stream.readLong());
			}
			dicElementIds = new int[numCorDimensions];
			for (int i = 0; i < numCorDimensions; i++){
				dicElementIds[i] = stream.readInt();
			}
			
			isHaveSmartPhone = stream.readBoolean();
			agentIdx = stream.readByte();
			browserIdx = stream.readByte();
			isFrequentChange = stream.readBoolean();

			int ip = stream.readInt();
	        int mask = stream.readInt();
	        ipAddress = new IP(ip, mask); 
			
			locationId = stream.readInt();
			cityIdx = stream.readInt();
			forumWallId = stream.readLong();
			//forumStatusId = stream.readInt();
			
			byte numOfTags = stream.readByte();
			setOfTags = new TreeSet<Integer>();
			for (byte i = 0; i < numOfTags;i++){
				setOfTags.add(stream.readInt());
			}
            mainTag = stream.readInt();
			
			numPopularPlace = stream.readByte(); 
			popularPlaceIds = new short[numPopularPlace];
			for (byte i=0; i < numPopularPlace; i++){
				popularPlaceIds[i] = stream.readShort();
			}
			
			universityLocationId = stream.readInt();
			gender = stream.readByte();
			birthDay = stream.readLong();
			isLargePoster = stream.readBoolean();
			
			stats = new Counts();
			stats.numberOfPosts = stream.readInt();
			stats.numberOfLikes = stream.readInt();
			stats.numberOfGroups = stream.readInt();
			stats.numberOfWorkPlaces = stream.readInt();
			stats.numberOfTagsOfPosts = stream.readInt();
			stats.numberOfFriends = stream.readInt();
	 }
	
	private void writeObject(java.io.ObjectOutputStream stream)
	throws IOException{
		 	stream.writeLong(accountId);
            stream.writeInt(sdpId);
			stream.writeLong(creationDate);
			stream.writeShort(numFriends);
			stream.writeShort(numFriendsAdded);
			stream.writeByte(numCorDimensions);
			for (int i = 0; i < numCorDimensions; i ++){
				stream.writeShort(numPassFriends[i]);
			}
			for (int i = 0; i < numCorDimensions; i ++){
				stream.writeShort(numPassFriendsAdded[i]);
			}
			
			for (int i = 0; i < numFriendsAdded; i++){
				friendList[i].write(stream);
			}
			//Read the size of Treeset first
			stream.writeInt(friendIds.size()); 
			Iterator<Long> it = friendIds.iterator();
			while (it.hasNext()){
				stream.writeLong(it.next());
			}
			
			for (int i = 0; i < numCorDimensions; i++){
				stream.writeInt(dicElementIds[i]);
			}
			
			stream.writeBoolean(isHaveSmartPhone);
			stream.writeByte(agentIdx);
			stream.writeByte(browserIdx);
			stream.writeBoolean(isFrequentChange);
			
			stream.writeInt(ipAddress.getIp());
			stream.writeInt(ipAddress.getMask());

			stream.writeInt(locationId);
			stream.writeInt(cityIdx);
			stream.writeLong(forumWallId);

			stream.writeByte((byte)setOfTags.size());
			Iterator<Integer> iter2 = setOfTags.iterator();
			while (iter2.hasNext()){
				stream.writeInt(iter2.next());
			}
            stream.writeInt(mainTag);

			stream.writeByte(numPopularPlace); 
			for (byte i=0; i < numPopularPlace; i++){
				stream.writeShort(popularPlaceIds[i]);
			}
			
			stream.writeInt(universityLocationId);
			stream.writeByte(gender);
			stream.writeLong(birthDay);
			stream.writeBoolean(isLargePoster);
			
			stream.writeInt(stats.numberOfPosts);
			stream.writeInt(stats.numberOfLikes);
			stream.writeInt(stats.numberOfGroups);
			stream.writeInt(stats.numberOfWorkPlaces);
			stream.writeInt(stats.numberOfTagsOfPosts);
			stream.writeInt(stats.numberOfFriends);
	 }
	
	public void readFields(DataInput arg0) throws IOException {
		accountId = arg0.readLong();
        sdpId = arg0.readInt();
		creationDate = arg0.readLong();
		numFriends = arg0.readShort();
		numFriendsAdded = arg0.readShort();
		numCorDimensions = arg0.readByte();
		numPassFriends = new short[numCorDimensions];
		for (int i = 0; i < numCorDimensions; i ++){
			numPassFriends[i] = arg0.readShort();
		}
		numPassFriendsAdded = new short[numCorDimensions];
		for (int i = 0; i < numCorDimensions; i ++){
			numPassFriendsAdded[i] = arg0.readShort();
		}
		friendList = new Friend[numFriends];
		friendIds = new TreeSet<Long>();
		for (int i = 0; i < numFriendsAdded; i++){
			Friend fr = new Friend(); 
			fr.readFields(arg0);
			friendList[i] = fr; 
		}
		//Read the size of Treeset first
		int size = arg0.readInt(); 
		for (int i = 0; i < size; i++){
			friendIds.add(arg0.readLong());
		}
		dicElementIds = new int[numCorDimensions];
		for (int i = 0; i < numCorDimensions; i++){
			dicElementIds[i] = arg0.readInt();
		}
		
		isHaveSmartPhone = arg0.readBoolean();
		agentIdx = arg0.readByte();
		browserIdx = arg0.readByte();
		isFrequentChange = arg0.readBoolean();

		int ip = arg0.readInt();
		int mask = arg0.readInt();
		ipAddress = new IP(ip, mask); 
		
		locationId = arg0.readInt();
		cityIdx = arg0.readInt();
		forumWallId = arg0.readLong();
		//forumStatusId = arg0.readInt();
		
		byte numTags = arg0.readByte(); 
		setOfTags = new TreeSet<Integer>();
		for (byte i = 0; i < numTags;i++){
			setOfTags.add(arg0.readInt());
		}
        mainTag = arg0.readInt();
		numPopularPlace = arg0.readByte();
		popularPlaceIds = new short[numPopularPlace];
		for (byte i=0; i < numPopularPlace; i++){
			popularPlaceIds[i] = arg0.readShort();
		}
		
		universityLocationId = arg0.readInt();
		gender = arg0.readByte();
		birthDay = arg0.readLong();
		isLargePoster = arg0.readBoolean();
		
		stats = new Counts();
		stats.numberOfPosts = arg0.readInt();
		stats.numberOfLikes = arg0.readInt();
		stats.numberOfGroups = arg0.readInt();
		stats.numberOfWorkPlaces = arg0.readInt();
		stats.numberOfTagsOfPosts = arg0.readInt();		
		stats.numberOfFriends = arg0.readInt();
	}
	
	public void copyFields(ReducedUserProfile user){
		accountId = user.accountId;
        sdpId = user.sdpId;
		creationDate = user.creationDate;
		numFriends = user.numFriends;
		numFriendsAdded = user.numFriendsAdded;
        numCorDimensions = user.numCorDimensions;
        numPassFriends = user.numPassFriends;
		numPassFriendsAdded = user.numPassFriendsAdded;
		friendList = user.friendList;
		friendIds = user.friendIds;
		dicElementIds = user.dicElementIds;
		isHaveSmartPhone = user.isHaveSmartPhone;
		agentIdx = user.agentIdx;
		browserIdx = user.browserIdx;
		isFrequentChange = user.isFrequentChange;
		ipAddress = user.ipAddress;
		locationId = user.locationId;
		cityIdx = user.cityIdx;
		forumWallId = user.forumWallId;
		setOfTags = user.setOfTags;
        mainTag = user.mainTag;
		numPopularPlace = user.numPopularPlace;
		popularPlaceIds = user.popularPlaceIds;
		universityLocationId = user.universityLocationId;
		gender = user.gender;
		birthDay = user.birthDay;
		isLargePoster = user.isLargePoster;
		stats = user.stats;
		stats.numberOfWorkPlaces = user.stats.numberOfWorkPlaces;
	}
	
	public void write(DataOutput arg0) throws IOException {
		arg0.writeLong(accountId);
        arg0.writeInt(sdpId);
		arg0.writeLong(creationDate);
		arg0.writeShort(numFriends);
		arg0.writeShort(numFriendsAdded);
		arg0.writeByte(numCorDimensions);
		for (int i = 0; i < numCorDimensions; i ++){
			arg0.writeShort(numPassFriends[i]);
		}
		for (int i = 0; i < numCorDimensions; i ++){
			 arg0.writeShort(numPassFriendsAdded[i]);
		}
		
		for (int i = 0; i < numFriendsAdded; i++){
			friendList[i].write(arg0);
		}
		//Read the size of Treeset first
		arg0.writeInt(friendIds.size()); 
		Iterator<Long> it = friendIds.iterator();
		while (it.hasNext()){
			arg0.writeLong(it.next());
		}
		
		for (int i = 0; i < numCorDimensions; i++){
			arg0.writeInt(dicElementIds[i]);
		}
		
		arg0.writeBoolean(isHaveSmartPhone);
		arg0.writeByte(agentIdx);
		arg0.writeByte(browserIdx);
		arg0.writeBoolean(isFrequentChange);

		arg0.writeInt(ipAddress.getIp());
		arg0.writeInt(ipAddress.getMask());
		
		
		arg0.writeInt(locationId);
		arg0.writeInt(cityIdx);
		arg0.writeLong(forumWallId);

		arg0.writeByte((byte)setOfTags.size()); 
		Iterator<Integer> iter2 = setOfTags.iterator();
		while (iter2.hasNext()){
			arg0.writeInt(iter2.next());
		}
        arg0.writeInt(mainTag);

		
		arg0.writeByte(numPopularPlace); 
		for (byte i=0; i < numPopularPlace; i++){
			arg0.writeShort(popularPlaceIds[i]);
		}
		
		arg0.writeInt(universityLocationId);
		arg0.writeByte(gender);
		arg0.writeLong(birthDay);
		arg0.writeBoolean(isLargePoster);
		
		arg0.writeInt(stats.numberOfPosts);
		arg0.writeInt(stats.numberOfLikes);
		arg0.writeInt(stats.numberOfGroups);
		arg0.writeInt(stats.numberOfWorkPlaces);
		arg0.writeInt(stats.numberOfTagsOfPosts);		
		arg0.writeInt(stats.numberOfFriends);
	}

	public ReducedUserProfile(){
		stats = new Counts();
	}
	
/*	public ReducedUserProfile(UserProfile user, int numCorrDimensions){
		this.setAccountId(user.getAccountId());
        this.setSdpId(user.getSdpId());
		this.setCreationDate(user.getCreationDate());
		this.setNumFriends(user.getNumFriends());
		this.setNumFriendsAdded((short)0);
		this.setUniversityLocationId(user.getUniversityLocationId());
        this.numCorDimensions = (byte)numCorrDimensions;
		dicElementIds = new int[numCorrDimensions];
		this.setGender(user.getGender());
		this.setBirthDay(user.getBirthDay());
		GregorianCalendar date = new GregorianCalendar();
		date.setTimeInMillis(birthDay);
		int birthYear = date.get(GregorianCalendar.YEAR);
		int organizationDimension = universityLocationId | (birthYear << 1) | gender;
		this.setDicElementId(organizationDimension,0);
		this.setDicElementId(user.getMainTagId(), 1);
		this.setDicElementId(user.getRandomIdx(),2);
		this.allocateFriendListMemory();
		this.setHaveSmartPhone(user.isHaveSmartPhone());
		this.setAgentIdx(user.getAgentId());
		this.setBrowserId(user.getBrowserId());
		this.setIpAddress(user.getIpAddress());
		this.setNumPassFriends(user.getNumPassFriends());
		this.setLocationId(user.getLocationId());
		this.setCityId(user.getCityId());
		this.setForumWallId(user.getForumWallId());
		this.setSetOfTags(user.getSetOfTags());
		this.setPopularPlaceIds(user.getPopularPlaceIds());
		this.setNumPopularPlace(user.getNumPopularPlace());
		
		this.numPassFriendsAdded = new short[numCorrDimensions];
		this.isLargePoster = user.isLargePoster();
	}
	*/
	
	public int getDicElementId(int index) {
		return dicElementIds[index];
	}

/*	public void setDicElementId(int dicElementId, int index) {
		this.dicElementIds[index] = dicElementId;
	}
	*/

	public void setNumPassFriendsAdded(int pass, short numPassFriendAdded) {
		numPassFriendsAdded[pass] = numPassFriendAdded;
	}

    public short getNumPassFriendsAdded(int pass) {
        return numPassFriendsAdded[pass];
    }

    public void setNumPassFriends(int pass, short numPassFriendAdded) {
        numPassFriends[pass] = numPassFriendAdded;
    }

    public short getNumPassFriends(int pass) {
        return numPassFriends[pass];
    }

	public short getNumFriendsAdded() {
		return numFriendsAdded;
	}
	
	public void addNewFriend(Friend friend) {
	    if (friend != null && !friendIds.contains(friend.getFriendAcc())) {
	        friendList[numFriendsAdded] = friend;
	        friendIds.add(friend.getFriendAcc());
	        numFriendsAdded++;
	    }
	}
	
	public boolean isExistFriend(long friendId){
		return friendIds.contains(friendId);
	}
	

    public void setNumDimensions( int numDimensions )	 {
        this.numCorDimensions = (byte)numDimensions;
        this.dicElementIds = new int[numDimensions];
        this.numPassFriendsAdded = new short[numDimensions];
        this.numPassFriends = new short[numDimensions];
    }
	public void allocateFriendListMemory(){
		friendList = new Friend[numFriends];
		friendIds = new TreeSet<Long>();
	}

	public Friend[] getFriendList() {
		return friendList;
	}
/*	public short getNumFriends(int pass) {
		return numPassFriends[pass];
	}
	*/
	public void setNumFriends(short numFriends) {
		this.numFriends = numFriends;
        this.allocateFriendListMemory();
	}
	public long getCreationDate() {
		return creationDate;
	}
	public void setCreationDate(long creationDate) {
		this.creationDate = creationDate;
	}
	public long getAccountId() {
		return accountId;
	}
	public void setAccountId(long accountId) {
		this.accountId = accountId;
	}
	public int getSdpId() {
		return sdpId;
	}
	public void setSdpId(int sdpId) {
		this.sdpId = sdpId;
	}
	public boolean isHaveSmartPhone() {
		return isHaveSmartPhone;
	}
	public void setHaveSmartPhone(boolean isHaveSmartPhone) {
		this.isHaveSmartPhone = isHaveSmartPhone;
	}
	public byte getAgentId() {
		return agentIdx;
	}
	public void setAgentId(byte agentIdx) {
		this.agentIdx = agentIdx;
	}
	public byte getBrowserIdx() {
		return browserIdx;
	}
	public void setBrowserId(byte browserIdx) {
		this.browserIdx = browserIdx;
	}
	public boolean isFrequentChange() {
		return isFrequentChange;
	}
	public void setFrequentChange(boolean isFrequentChange) {
		this.isFrequentChange = isFrequentChange;
	}
	public IP getIpAddress() {
		return ipAddress;
	}
	public void setIpAddress(IP ipAddress) {
		this.ipAddress = ipAddress;
	}

	public int getLocationId() {
		return locationId;
	}
	public void setLocationId(int locationId) {
		this.locationId = locationId;
	}
	public int getCityId() {
        return cityIdx;
    }
    public void setCityId(int cityIdx) {
        this.cityIdx = cityIdx;
    }
	public long getForumWallId() {
		return forumWallId;
	}
	public void setForumWallId(long forumWallId) {
		this.forumWallId = forumWallId;
	}
	public TreeSet<Integer> getSetOfTags() {
		return setOfTags;
	}
	public void setSetOfTags(TreeSet<Integer> setOfTags) {
		this.setOfTags = setOfTags;
	}
	public byte getNumPopularPlace() {
		return numPopularPlace;
	}
	public void setNumPopularPlace(byte numPopularPlace) {
		this.numPopularPlace = numPopularPlace;
	}
	public short getPopularId(int index){
		return popularPlaceIds[index];
	}
	public short[] getPopularPlaceIds() {
		return popularPlaceIds;
	}
	public void setPopularPlaceIds(short[] popularPlaceIds) {
		this.popularPlaceIds = popularPlaceIds;
	}
	public short getNumFriends() {
		return numFriends;
	}
	public TreeSet<Long> getFriendIds() {
		return friendIds;
	}
	public int[] getDicElementIds() {
		return dicElementIds;
	}
	public int getUniversityLocationId() {
		return universityLocationId;
	}
	public void setUniversityLocationId(int universityLocatonId) {
		this.universityLocationId = universityLocatonId;
        GregorianCalendar date = new GregorianCalendar();
        date.setTimeInMillis(birthDay);
        int birthYear = date.get(GregorianCalendar.YEAR);
        int organizationDimension = (int) (universityLocationId | (birthYear << 1) | gender);
        dicElementIds[0] = organizationDimension;
	}
	public byte getGender() {
		return gender;
	}
	public void setGender(byte gender) {
		this.gender = gender;
	}
	public long getBirthDay() {
		return birthDay;
	}
	public void setBirthDay(long birthDay) {
		this.birthDay = birthDay;
	}
	public boolean isLargePoster() {
		return this.isLargePoster;
	}
	public void setLargePoster(boolean isLargePoster) {
		this.isLargePoster = isLargePoster;
	}
	
	public int getNumOfPosts(){
		return stats.numberOfPosts;
	}
	
	public int getNumOfLikes(){
		return stats.numberOfLikes;
	}
	
	public int getNumOfGroups(){
		return stats.numberOfGroups;
	}
	
	public int getNumOfWorkPlaces(){
		return stats.numberOfWorkPlaces;
	}
	
	public int getNumOfTagsOfPosts(){
		return stats.numberOfTagsOfPosts;
	}
	
	public void addNumOfPosts(int num){
		stats.numberOfPosts += num;
	}
	
	public void addNumOfGroups(int num){
		stats.numberOfGroups += num;
	}
	
	public void addNumOfWorkPlaces(int num){
		stats.numberOfWorkPlaces += num;
	}
	
	public void addNumOfTagsOfPosts(int num){
		stats.numberOfTagsOfPosts += num;
	}
	
	public void addNumOfLikesToPosts(int num){
		stats.numberOfLikes += num;
	}
    public void setMainTag( int mainTag ) {
        this.mainTag = mainTag;
        dicElementIds[1] = this.mainTag;
    }

    public void setRandomId( int randomId ) {
        dicElementIds[2] = randomId;
    }
} public class FlashmobTag implements Comparable<FlashmobTag> {
	public int level;
	public long date;
	public double prob;
	public int tag;
	public int compareTo( FlashmobTag t ) {
		if( this.date - t.date < 0) return -1;
		if( this.date - t.date > 0) return 1;
		if( this.date - t.date == 0) return 0;
		return 0;
	}

    public void copyTo( FlashmobTag t ) {
        t.level = this.level;
        t.date = this.date;
        t.prob = this.prob;
        t.tag = this.tag;
    }
}
 public class Distribution {
    

    double[] distribution;
    String distributionFile;
    
    public Distribution( String distributionFile ){
        this.distributionFile = distributionFile;
    }

    public void initialize() {
        try{
		    BufferedReader distributionBuffer = new BufferedReader(new InputStreamReader(getClass().getResourceAsStream(distributionFile), "UTF-8"));
            ArrayList<Double> temp = new ArrayList<Double>();
		    String line;
		    while ((line = distributionBuffer.readLine()) != null){
                Double prob = Double.valueOf(line);
                temp.add(prob);
            }
            distribution = new double[temp.size()];
            int index = 0;
            Iterator<Double> it = temp.iterator();
            while(it.hasNext()) {
                distribution[index] = it.next();
                ++index;
            }
		} catch (IOException e) {
			e.printStackTrace();
		}
    }

    private int binarySearch( double prob ) {
        int upperBound = distribution.length - 1;
        int lowerBound = 0;
        int midPoint = (upperBound + lowerBound)  / 2;
        while (upperBound > (lowerBound+1)){
            if (distribution[midPoint] > prob ){
                upperBound = midPoint;
            } else{
                lowerBound = midPoint; 
            }
            midPoint = (upperBound + lowerBound)  / 2;
        }
        return midPoint;
    }

    public double nextDouble( Random random) {
        return (double)binarySearch(random.nextDouble())/(double)distribution.length;
    }
}
import java.util.*;
import ldbc.socialnet.dbgen.generator.DateGenerator;
import ldbc.socialnet.dbgen.generator.ScalableGenerator;
import ldbc.socialnet.dbgen.dictionary.TagTextDictionary;
import ldbc.socialnet.dbgen.dictionary.FlashmobTagDictionary;
import ldbc.socialnet.dbgen.dictionary.UserAgentDictionary;
import ldbc.socialnet.dbgen.dictionary.IPAddressDictionary;
import ldbc.socialnet.dbgen.dictionary.BrowserDictionary;
import ldbc.socialnet.dbgen.dictionary.TagMatrix;
import ldbc.socialnet.dbgen.objects.Comment;
import ldbc.socialnet.dbgen.objects.Friend;
import ldbc.socialnet.dbgen.objects.Group;
import ldbc.socialnet.dbgen.objects.GroupMemberShip;
import ldbc.socialnet.dbgen.objects.Post;
import ldbc.socialnet.dbgen.objects.ReducedUserProfile;
import ldbc.socialnet.dbgen.objects.FlashmobTag;
import ldbc.socialnet.dbgen.util.Distribution;

/*
 * Copyright (c) 2013 LDBC
 * Linked Data Benchmark Council (http://ldbc.eu)
 *
 * This file is part of ldbc_socialnet_dbgen.
 *
 * ldbc_socialnet_dbgen is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 * 
 * ldbc_socialnet_dbgen is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License
 * along with ldbc_socialnet_dbgen.  If not, see <http://www.gnu.org/licenses/>.
 *
 * Copyright (C) 2011 OpenLink Software <bdsmt@openlinksw.com>
 * All Rights Reserved.
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation;  only Version 2 of the License dated
 * June 1991.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 */

package ldbc.socialnet.dbgen.generator;





public class FlashmobPostGenerator extends PostGenerator {

    private static final String SEPARATOR = "  ";

    private DateGenerator dateGen;                          /**< @brief the date generator.*/

    private FlashmobTagDictionary flashmobTagDictionary;    /**< @brief The flashmobTagDictionary used to get the tags of the posts.*/
