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

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.PriorityQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.expression.Expression;
import org.springframework.expression.ExpressionParser;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.expression.spel.support.StandardEvaluationContext;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import ch.uzh.ddis.katts.query.stream.StreamConsumer;

/**
 * This bolt implements a generic synchronization between all receiving streams of the bolt. The synchronization of the
 * streams provides a single stream of events in a correct temporal order.
 * 
 * @author <NAME>
 * 
 */
public abstract class AbstractStreamSynchronizedBolt extends AbstractSynchronizedBolt {

	private static final long serialVersionUID = 1L;

	// TODO: Move this buffer into the storage
	private PriorityQueue<StreamSynchronizedEventWrapper> buffer = new PriorityQueue<StreamSynchronizedEventWrapper>();

	// TODO: Move this lastDatePerStream into the storage
	private Map<StreamConsumer, Date> lastDatePerStream = new HashMap<StreamConsumer, Date>();

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

	private Expression eventSynchronizationExpression;

	@Override
	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		super.prepare(stormConf, context, collector);
		// try {
		// String boltId = this.getId();
		// lastDatePerStream = StorageFactory.createDefaultStorage(boltId + "_last_date_per_stream");
		//
		// } catch (InstantiationException e) {
		// logger.error("Could not load storage object.", e);
		// } catch (IllegalAccessException e) {
		// logger.error("Could not load storage object.", e);
		// }

		// for (StreamConsumer stream : this.getStreamConsumer()) {
		// buffers.put(stream, new PriorityQueue<Event>(stream.getMaxBufferSize(), new EventTimeComparator()));
		// }

	}

	/**
	 * This method is called when the event is synchronized with in the stream. We override this method to provide
	 * stream synchronization.
	 */
	@Override
	protected void executeSynchronizedEvent(Event event) {

		StreamSynchronizedEventWrapper syncEvent = new StreamSynchronizedEventWrapper(event,
				this.getSynchronizationDate(event));
		buffer.add(syncEvent);
		lastDatePerStream.put(event.getEmittedOn(), syncEvent.getSynchronizationDate());
		event.ack();
		executeEventsInBuffer();

	}

	/**
	 * This method executes all events in the buffer, with the correct temporal order.
	 */
	private void executeEventsInBuffer() {

		StreamSynchronizedEventWrapper next = buffer.peek();

		boolean bufferTimeout = false;
		long timeout = next.getEmittedOn().getRealBufferTimout();
		if (timeout > 0) {
			Date lastDate = lastDatePerStream.get(next.getEmittedOn());

			if (Math.abs((lastDate.getTime() - next.getSynchronizationDate().getTime())) > timeout) {
				bufferTimeout = true;
			}
		}

		if (next != null && (isInTemporalOrder(next) || bufferTimeout)) {
			executeSynchronizedStreamEvent(next);
		}

	}

	@Override
	public void ack(Event event) {
		super.ack(event);

		buffer.remove(event);

		// Since we found an event with correct temporal order, try to find
		// other events and execute them.
		executeEventsInBuffer();
	}

	/**
	 * This method checks if the given event is in the correct temporal order corresponding to the all streams. An event
	 * is in temporal order, when it has the smallest event date overall streams.
	 * 
	 * @param event
	 *            The event to check for ordering.
	 * @return True when the event is in Order, False when the event is not in the corresponding order.
	 */
	private boolean isInTemporalOrder(StreamSynchronizedEventWrapper event) {

		for (Entry<StreamConsumer, Date> dateEntry : lastDatePerStream.entrySet()) {
			if (dateEntry.getValue().before(event.getSynchronizationDate())) {
				return false;
			}<commit_msg>avoid NullPointerException<commit_after>package ch.uzh.ddis.katts.bolts;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.PriorityQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.expression.Expression;
import org.springframework.expression.ExpressionParser;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.expression.spel.support.StandardEvaluationContext;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import ch.uzh.ddis.katts.query.stream.StreamConsumer;

/**
 * This bolt implements a generic synchronization between all receiving streams of the bolt. The synchronization of the
 * streams provides a single stream of events in a correct temporal order.
 * 
 * @author <NAME>
 * 
 */
public abstract class AbstractStreamSynchronizedBolt extends AbstractSynchronizedBolt {

	private static final long serialVersionUID = 1L;

	// TODO: Move this buffer into the storage
	private PriorityQueue<StreamSynchronizedEventWrapper> buffer = new PriorityQueue<StreamSynchronizedEventWrapper>();

	// TODO: Move this lastDatePerStream into the storage
	private Map<StreamConsumer, Date> lastDatePerStream = new HashMap<StreamConsumer, Date>();

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

	private Expression eventSynchronizationExpression;

	@Override
	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		super.prepare(stormConf, context, collector);
		// try {
		// String boltId = this.getId();
		// lastDatePerStream = StorageFactory.createDefaultStorage(boltId + "_last_date_per_stream");
		//
		// } catch (InstantiationException e) {
		// logger.error("Could not load storage object.", e);
		// } catch (IllegalAccessException e) {
		// logger.error("Could not load storage object.", e);
		// }

		// for (StreamConsumer stream : this.getStreamConsumer()) {
		// buffers.put(stream, new PriorityQueue<Event>(stream.getMaxBufferSize(), new EventTimeComparator()));
		// }

	}

	/**
	 * This method is called when the event is synchronized with in the stream. We override this method to provide
	 * stream synchronization.
	 */
	@Override
	protected void executeSynchronizedEvent(Event event) {

		StreamSynchronizedEventWrapper syncEvent = new StreamSynchronizedEventWrapper(event,
				this.getSynchronizationDate(event));
		buffer.add(syncEvent);
		lastDatePerStream.put(event.getEmittedOn(), syncEvent.getSynchronizationDate());
		event.ack();
		executeEventsInBuffer();

	}

	/**
	 * This method executes all events in the buffer, with the correct temporal order.
	 */
	private void executeEventsInBuffer() {
		StreamSynchronizedEventWrapper next = buffer.peek();
		if (next != null) {
			boolean bufferTimeout = false;
			long timeout = next.getEmittedOn().getRealBufferTimout();
			if (timeout > 0) {
				Date lastDate = lastDatePerStream.get(next.getEmittedOn());

				if (Math.abs((lastDate.getTime() - next.getSynchronizationDate().getTime())) > timeout) {
					bufferTimeout = true;
				}
			}

			if ((isInTemporalOrder(next) || bufferTimeout)) {
				executeSynchronizedStreamEvent(next);
			}
		}
	}

	@Override
	public void ack(Event event) {
		super.ack(event);

		buffer.remove(event);

		// Since we found an event with correct temporal order, try to find
		// other events and execute them.
		executeEventsInBuffer();
	}

	/**
	 * This method checks if the given event is in the correct temporal order corresponding to the all streams. An event
	 * is in temporal order, when it has the smallest event date overall streams.
	 * 
	 * @param event
	 *            The event to check for ordering.
	 * @return True when the event is in Order, False when the event is not in the corresponding order.
	 */
	private boolean isInTemporalOrder(StreamSynchronizedEventWrapper event) {

		for (Entry<StreamConsumer, Date> dateEntry : lastDatePerStream.entrySet()) {
			if (dateEntry.getValue().before(event.getSynchronizationDate())) {
				return false;
			}
{'fraction_non_alphanumeric': 0.09790886014746766, 'fraction_numerical': 0.0009670010878762239, 'mean_word_length': 4.6209239130434785, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 21, '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': '3439647', 'n_tokens_mistral': 2460, 'n_tokens_neox': 2245, 'n_words': 658}