<commit_before>package big.marketing.controller;

import java.io.IOException;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.EnumMap;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Semaphore;

import org.apache.log4j.Logger;

import big.marketing.Application;
import big.marketing.Settings;
import big.marketing.data.DataType;
import big.marketing.data.FlowMessage;
import big.marketing.data.HealthMessage;
import big.marketing.data.IPSMessage;
import big.marketing.data.Node;

import com.mongodb.BasicDBObject;
import com.mongodb.DB;
import com.mongodb.DBCollection;
import com.mongodb.DBCursor;
import com.mongodb.DBObject;
import com.mongodb.MongoClient;

/*
 * Tutorial on Mongo with java:
 * http://docs.mongodb.org/ecosystem/tutorial/getting-started-with-java-driver/#getting-started-with-java-driver
 */
public class MongoController implements Runnable {

	static Logger logger = Logger.getLogger(MongoController.class);

	/*
	 * writerThread and buffer for storing write requests. handling read requests
	 * in a seperate thread doesn't make sense, since reads are always
	 * synchronous (as a value has to be returned)
	 */
	Thread writer;

	EnumMap<DataType, CollectionHandler> collections;

	private static MongoController instance;
	private static MongoClient mongo;
	private static DB database;

	public static String HOST_NAME = "localhost", DB_NAME = "eyeNet", FLOW_COLLECTION_NAME = "flow", IPS_COLLECTION_NAME = "ips",
	      HEALTH_COLLECTION_NAME = "health", DESCRIPTION_COLLECTION_NAME = "nodes";
	public static int BUFFER_SIZE = 1000;
	public static int MAX_TRIES = 3;

	public static MongoController getInstance() {
		if (instance == null)
			instance = new MongoController();
	   return instance;
   }
	
	private MongoController() {
		loadSettings();

		boolean isConnected = connectToDatabase();
		for (int i = 1; i <= MAX_TRIES && !isConnected; i++) {
			logger.warn("Could not connect to MongoDB in try " + i + " of " + MAX_TRIES);
			MongoExecutor.startMongoProcess();
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				logger.warn("Interrupted: "+e.getLocalizedMessage());
			}
			isConnected = connectToDatabase();
		}

		if (!isConnected) {
			logger.fatal("Failed to start MongoDB! Exiting now ...");
			Application.quit();
		}

		collections = new EnumMap<DataType, CollectionHandler>(DataType.class);

		for (DataType t : DataType.values()) {
			collections.put(t, new CollectionHandler(t));
		}

		writer = new Thread(this, "DB-Writer");
		writer.start();
	}

	private void loadSettings() {
		HOST_NAME = Settings.get("mongo.hostname");
		DB_NAME = Settings.get("mongo.databasename");
		FLOW_COLLECTION_NAME = Settings.get("mongo.collections.flow");
		IPS_COLLECTION_NAME = Settings.get("mongo.collections.ips");
		HEALTH_COLLECTION_NAME = Settings.get("mongo.collections.health");
		DESCRIPTION_COLLECTION_NAME = Settings.get("mongo.collections.network");
		BUFFER_SIZE = Settings.getInt("mongo.writeBuffer.size");
		MAX_TRIES = Settings.getInt("mongo.exe.maxtries");

	}

	public boolean connectToDatabase() {

		try {
			mongo = new MongoClient(HOST_NAME);
			mongo.getConnector().getDBPortPool(mongo.getAddress()).get().ensureOpen();
			database = mongo.getDB(DB_NAME);
			logger.info("Connection to MongoDB established");
			return true;
		} catch (UnknownHostException e) {
			logger.error(e.getMessage());
		} catch (IOException e) {
			logger.warn("No MongoDB server running!");
		}
		return false;
	}

	private BlockingQueue<DBObject> getBuffer(DataType t) {
		return collections.get(t).buffer;
	}

	DBCollection getCollection(DataType t) {
		return collections.get(t).collection;
	}

	public void setCollection(DataType t, String newCollectionName, boolean dropOld) {

		// stop the writer thread and wait for it to finish
		int permits = sem.drainPermits();
		while (!sem.hasQueuedThreads()) {
			try {
				Thread.sleep(100);
			} catch (InterruptedException e) {
				logger.error(e.getLocalizedMessage());
			}
		}

		collections.get(t).flushBuffer();
		DBCollection old = getCollection(t);
		DBCollection newColl = database.getCollection(newCollectionName);

		if (dropOld) {
			newColl.rename(getCollectionName(t), true);
		} else {
			old.rename("tmp" + newCollectionName);
			newColl.rename(getCollectionName(t));
			database.getCollection("tmp" + newCollectionName).rename(newCollectionName);
		}
		collections.put(t, new CollectionHandler(t));
		sem.release(permits);
	}

	public void flushBuffers(){
		for (CollectionHandler handler : collections.values()){
			handler.flushBuffer();
		}
	}
	
	private String getCollectionName(DataType t) {
		switch (t) {
		case FLOW:
			return FLOW_COLLECTION_NAME;
		case HEALTH:<commit_msg>rename unprocessed collection to old...<commit_after>package big.marketing.controller;

import java.io.IOException;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.EnumMap;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Semaphore;

import org.apache.log4j.Logger;

import big.marketing.Application;
import big.marketing.Settings;
import big.marketing.data.DataType;
import big.marketing.data.FlowMessage;
import big.marketing.data.HealthMessage;
import big.marketing.data.IPSMessage;
import big.marketing.data.Node;

import com.mongodb.BasicDBObject;
import com.mongodb.DB;
import com.mongodb.DBCollection;
import com.mongodb.DBCursor;
import com.mongodb.DBObject;
import com.mongodb.MongoClient;

/*
 * Tutorial on Mongo with java:
 * http://docs.mongodb.org/ecosystem/tutorial/getting-started-with-java-driver/#getting-started-with-java-driver
 */
public class MongoController implements Runnable {

	static Logger logger = Logger.getLogger(MongoController.class);

	/*
	 * writerThread and buffer for storing write requests. handling read requests
	 * in a seperate thread doesn't make sense, since reads are always
	 * synchronous (as a value has to be returned)
	 */
	Thread writer;

	EnumMap<DataType, CollectionHandler> collections;

	private static MongoController instance;
	private static MongoClient mongo;
	private static DB database;

	public static String HOST_NAME = "localhost", DB_NAME = "eyeNet", FLOW_COLLECTION_NAME = "flow", IPS_COLLECTION_NAME = "ips",
	      HEALTH_COLLECTION_NAME = "health", DESCRIPTION_COLLECTION_NAME = "nodes";
	public static int BUFFER_SIZE = 1000;
	public static int MAX_TRIES = 3;

	public static MongoController getInstance() {
		if (instance == null)
			instance = new MongoController();
	   return instance;
   }
	
	private MongoController() {
		loadSettings();

		boolean isConnected = connectToDatabase();
		for (int i = 1; i <= MAX_TRIES && !isConnected; i++) {
			logger.warn("Could not connect to MongoDB in try " + i + " of " + MAX_TRIES);
			MongoExecutor.startMongoProcess();
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				logger.warn("Interrupted: "+e.getLocalizedMessage());
			}
			isConnected = connectToDatabase();
		}

		if (!isConnected) {
			logger.fatal("Failed to start MongoDB! Exiting now ...");
			Application.quit();
		}

		collections = new EnumMap<DataType, CollectionHandler>(DataType.class);

		for (DataType t : DataType.values()) {
			collections.put(t, new CollectionHandler(t));
		}

		writer = new Thread(this, "DB-Writer");
		writer.start();
	}

	private void loadSettings() {
		HOST_NAME = Settings.get("mongo.hostname");
		DB_NAME = Settings.get("mongo.databasename");
		FLOW_COLLECTION_NAME = Settings.get("mongo.collections.flow");
		IPS_COLLECTION_NAME = Settings.get("mongo.collections.ips");
		HEALTH_COLLECTION_NAME = Settings.get("mongo.collections.health");
		DESCRIPTION_COLLECTION_NAME = Settings.get("mongo.collections.network");
		BUFFER_SIZE = Settings.getInt("mongo.writeBuffer.size");
		MAX_TRIES = Settings.getInt("mongo.exe.maxtries");

	}

	public boolean connectToDatabase() {

		try {
			mongo = new MongoClient(HOST_NAME);
			mongo.getConnector().getDBPortPool(mongo.getAddress()).get().ensureOpen();
			database = mongo.getDB(DB_NAME);
			logger.info("Connection to MongoDB established");
			return true;
		} catch (UnknownHostException e) {
			logger.error(e.getMessage());
		} catch (IOException e) {
			logger.warn("No MongoDB server running!");
		}
		return false;
	}

	private BlockingQueue<DBObject> getBuffer(DataType t) {
		return collections.get(t).buffer;
	}

	DBCollection getCollection(DataType t) {
		return collections.get(t).collection;
	}

	public void setCollection(DataType t, String newCollectionName, boolean dropOld) {

		// stop the writer thread and wait for it to finish
		int permits = sem.drainPermits();
		while (!sem.hasQueuedThreads()) {
			try {
				Thread.sleep(100);
			} catch (InterruptedException e) {
				logger.error(e.getLocalizedMessage());
			}
		}

		collections.get(t).flushBuffer();
		DBCollection old = getCollection(t);
		DBCollection newColl = database.getCollection(newCollectionName);

		if (dropOld) {
			newColl.rename(getCollectionName(t), true);
		} else {
			old.rename("old" + getCollectionName(t));
			newColl.rename(getCollectionName(t));
		}
		collections.put(t, new CollectionHandler(t));
		sem.release(permits);
	}

	public void flushBuffers(){
		for (CollectionHandler handler : collections.values()){
			handler.flushBuffer();
		}
	}
	
	private String getCollectionName(DataType t) {
		switch (t) {
		case FLOW:
			return FLOW_COLLECTION_NAME;
		case HEALTH:
{'fraction_non_alphanumeric': 0.11873514518962489, 'fraction_numerical': 0.0028934587165443835, 'mean_word_length': 4.733412322274882, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1556024', 'n_tokens_mistral': 3334, 'n_tokens_neox': 2998, 'n_words': 669}