<commit_before>package ch.uzh.ddis.katts.bolts.source;

import java.io.InputStream;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;

import org.joda.time.format.DateTimeFormatter;
import org.joda.time.format.ISODateTimeFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import clojure.pprint.pretty_writer__init;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import ch.uzh.ddis.katts.bolts.source.file.CSVSource;
import ch.uzh.ddis.katts.bolts.source.file.GzipSourceWrapper;
import ch.uzh.ddis.katts.bolts.source.file.Source;
import ch.uzh.ddis.katts.bolts.source.file.ZipSourceWrapper;
import ch.uzh.ddis.katts.monitoring.StarterMonitor;
import ch.uzh.ddis.katts.query.source.File;
import ch.uzh.ddis.katts.spouts.file.HeartBeatSpout;

/**
 * This Bolt reads in triples from files. By the abstraction of the {@link Source} different type of files can be read
 * with this bolt.
 * 
 * Internally a thread is run to read in the data. Storm uses normally Spouts for read in data. Because of the
 * heartbeat, this must be implement as a Bolt because the heartbeat is provided centrally and hence the reader must
 * listen to this stream. Spouts are not able to listen to streams. Therefore a Bolt is used for this task.
 * 
 * To improve the parallelization of reading this bolt accepts multiple files. For each file a separate instance is
 * created.
 * 
 * TODO: Checkout how reliability can be achieved with this architecture.
 * 
 * @author <NAME>
 * 
 */
public class FileTripleReader implements IRichBolt {

	/** This formatter is used to parse dateTime string values */
	private transient DateTimeFormatter isoFormat = ISODateTimeFormat.dateTimeParser();

	private static final long serialVersionUID = 1L;
	private OutputCollector collector;
	private FileTripleReaderConfiguration configuration;
	private Source source = null;
	private StarterMonitor starterMonitor;

	private Thread thread = null;

	private Date currentRealTimeDate;

	private Logger logger = LoggerFactory.getLogger(FileTripleReader.class);

	private long numberRead = 0;

	/**
	 * Converts a string into a proper Java object.
	 * <p/>
	 * The order in which we try to convert the values is: Long - Double - Date - String.
	 * 
	 * @param value
	 *            the string value to convert into an object.
	 * @return the created object.
	 */
	public Object convertStringToObject(String value) {
		Object result = value;

		try {
			result = this.isoFormat.parseDateTime(value).toDate();
		} catch (IllegalArgumentException e) {
			// so it's also not a date either
		}

		try {
			result = Double.valueOf(value);
		} catch (NumberFormatException e) {
			// so it's not a double
		}

		try {
			result = Long.valueOf(value);
		} catch (NumberFormatException e) {
			// so it's not a long
		}

		return result;
	}
	
	/**
	 * This method reads the next tuple from the file source.
	 * 
	 * @return True, when a tuple was emitted.
	 */
	public boolean nextTuple() {
		final String dateStringValue;
		final Date date;
		// TODO Add support for end date

		List<String> triple = null;
		try {
			triple = source.getNextTuple();
		} catch (Exception e) {
			throw new RuntimeException(String.format("Unable to read next triple because: %1s", e.getMessage()), e);
		}

		if (triple == null) {
			logger.info(String.format("End of file is reached in component %1s at date %2s. Line: %3s", this
					.getConfiguration().getId(), currentRealTimeDate.toString(), numberRead));
			return false;
		}<commit_msg>workaround for necessity to set bogus date at the end of the file<commit_after>package ch.uzh.ddis.katts.bolts.source;

import java.io.InputStream;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;

import org.joda.time.format.DateTimeFormatter;
import org.joda.time.format.ISODateTimeFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import clojure.pprint.pretty_writer__init;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import ch.uzh.ddis.katts.bolts.source.file.CSVSource;
import ch.uzh.ddis.katts.bolts.source.file.GzipSourceWrapper;
import ch.uzh.ddis.katts.bolts.source.file.Source;
import ch.uzh.ddis.katts.bolts.source.file.ZipSourceWrapper;
import ch.uzh.ddis.katts.monitoring.StarterMonitor;
import ch.uzh.ddis.katts.query.source.File;
import ch.uzh.ddis.katts.spouts.file.HeartBeatSpout;

/**
 * This Bolt reads in triples from files. By the abstraction of the {@link Source} different type of files can be read
 * with this bolt.
 * 
 * Internally a thread is run to read in the data. Storm uses normally Spouts for read in data. Because of the
 * heartbeat, this must be implement as a Bolt because the heartbeat is provided centrally and hence the reader must
 * listen to this stream. Spouts are not able to listen to streams. Therefore a Bolt is used for this task.
 * 
 * To improve the parallelization of reading this bolt accepts multiple files. For each file a separate instance is
 * created.
 * 
 * TODO: Checkout how reliability can be achieved with this architecture.
 * 
 * @author <NAME>
 * 
 */
public class FileTripleReader implements IRichBolt {

	/** This formatter is used to parse dateTime string values */
	private transient DateTimeFormatter isoFormat = ISODateTimeFormat.dateTimeParser();

	private static final long serialVersionUID = 1L;
	private OutputCollector collector;
	private FileTripleReaderConfiguration configuration;
	private Source source = null;
	private StarterMonitor starterMonitor;

	private Thread thread = null;

	private Date currentRealTimeDate;

	private Logger logger = LoggerFactory.getLogger(FileTripleReader.class);

	private long numberRead = 0;

	/**
	 * Converts a string into a proper Java object.
	 * <p/>
	 * The order in which we try to convert the values is: Long - Double - Date - String.
	 * 
	 * @param value
	 *            the string value to convert into an object.
	 * @return the created object.
	 */
	public Object convertStringToObject(String value) {
		Object result = value;

		try {
			result = this.isoFormat.parseDateTime(value).toDate();
		} catch (IllegalArgumentException e) {
			// so it's also not a date either
		}

		try {
			result = Double.valueOf(value);
		} catch (NumberFormatException e) {
			// so it's not a double
		}

		try {
			result = Long.valueOf(value);
		} catch (NumberFormatException e) {
			// so it's not a long
		}

		return result;
	}
	
	/**
	 * This method reads the next tuple from the file source.
	 * 
	 * @return True, when a tuple was emitted.
	 */
	public boolean nextTuple() {
		final String dateStringValue;
		final Date date;
		// TODO Add support for end date

		List<String> triple = null;
		try {
			triple = source.getNextTuple();
		} catch (Exception e) {
			throw new RuntimeException(String.format("Unable to read next triple because: %1s", e.getMessage()), e);
		}

		if (triple == null) {
			logger.info(String.format("End of file is reached in component %1s at date %2s. Line: %3s", this
					.getConfiguration().getId(), currentRealTimeDate.toString(), numberRead));
			
			/*
			 * TODO: this is a dirty hack: by setting the process date to the year 292278994, the next
			 * heartbeat that gets sent down the processing chain will tell the TerminationBolt, that we
			 * reached the end of the file.
			 */
			currentRealTimeDate = new Date(Long.MAX_VALUE); // set the current date 
			
			return false;
		}
{'fraction_non_alphanumeric': 0.09020356234096692, 'fraction_numerical': 0.0031806615776081423, 'mean_word_length': 3.8614718614718613, 'pattern_counts': {'":': 0, '<': 9, '<?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': '6272378', 'n_tokens_mistral': 2537, 'n_tokens_neox': 2308, 'n_words': 800}