// Path: SensorStorm/src/nl/tno/sensorstorm/api/particles/MetaParticle.java
// public interface MetaParticle extends Particle {
// 
// 	/**
// 	 * @return originId, an unique identifier for a spout or bolt instance
// 	 */
// 	String getOriginId();
// 
// 	/**
// 	 * Set the originId of the {@link MetaParticle}. This method should only be
// 	 * called by the {@link SensorStormSpout} or {@link SensorStormBolt}.
// 	 * 
// 	 * @param originId
// 	 *            unique identifier for a spout or bolt instance
// 	 */
// 	void setOriginId(String originId);
// 
// 	/**
// 	 * Check if this MetaParticle is equal to another {@link Particle}.
// 	 * {@link MetaParticle}s need to implement their own equals method, which
// 	 * ignores the originId. This method is used by the {@link SyncBuffer} in
// 	 * order to remove duplicate {@link MetaParticle}s.
// 	 * 
// 	 * @param other
// 	 *            Other MetaParticle used for comparison
// 	 * @return True if this object is equal to another instance of a
// 	 *         MetaParticle, with the exception of the originId.
// 	 */
// 	boolean equalMetaParticle(MetaParticle other);
// }
// 
// Path: SensorStorm/src/nl/tno/sensorstorm/api/processing/Operation.java
// public interface Operation extends Serializable {
// 
// 	/**
// 	 * Initialize this operation.
// 	 * 
// 	 * @param fieldGrouperValue
// 	 *            Value for the grouping field if any, otherwise null
// 	 * @param startTimeStamp
// 	 *            The time of the first particle this operation will get. Or -1
// 	 *            if a single operation is created for all particles (in the
// 	 *            SensorStormBolt constructor the fieldgrouperid == null)
// 	 * @param stormNativeConfig
// 	 *            A reference to the storm config object
// 	 * @param zookeeperStormConfiguration
// 	 *            A reference to the {@link ExternalStormConfiguration}
// 	 * @throws OperationException
// 	 *             If an error occurs in the {@link Operation}
// 	 * 
// 	 */
// 	void init(String fieldGrouperValue, long startTimeStamp,
// 			@SuppressWarnings("rawtypes") Map stormNativeConfig,
// 			ExternalStormConfiguration zookeeperStormConfiguration)
// 			throws OperationException;
// 
// }


import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.TreeSet;
import nl.tno.sensorstorm.api.particles.MetaParticle;
import nl.tno.sensorstorm.api.particles.Particle;
import nl.tno.sensorstorm.api.processing.Operation;
package nl.tno.sensorstorm.impl;



/**
 * Class that synchronizes {@link Particle}s coming from multiple (probably
 * parallel executing) sources and filters out duplicate {@link MetaParticle}s.
 * This buffer introduces a configurable delay.
 * <p>
 * When the delay is 0 the buffer will always immediately return the same
 * particle. Warning: When the size is 0, duplicate MetaParticles will not be
 * filtered out!
 */
public class SyncBuffer {

	private static final int INITIAL_CAPACITY_RETURN_LIST = 5;

	/**
	 * Special {@link Comparator} used by the {@link SyncBuffer}. It sorts
	 * {@link Particle}s based on their timestamp. Additionally, duplicate
	 * {@link MetaParticle}s (with possibly a different originId) will return 0,
	 * which results in removing the duplicates from the TreeSet.
	 */
	private static class ParticleComparator implements Comparator<Particle> {

		@Override
		public int compare(Particle p1, Particle p2) {
			int comp = Long.compare(p1.getTimestamp(), p2.getTimestamp());
			if (comp == 0) {
				// Same timestamp
				if ((p1 instanceof MetaParticle)
// Path: SensorStorm/src/nl/tno/sensorstorm/api/particles/MetaParticle.java
// public interface MetaParticle extends Particle {
// 
// 	/**
// 	 * @return originId, an unique identifier for a spout or bolt instance
// 	 */
// 	String getOriginId();
// 
// 	/**
// 	 * Set the originId of the {@link MetaParticle}. This method should only be
// 	 * called by the {@link SensorStormSpout} or {@link SensorStormBolt}.
// 	 * 
// 	 * @param originId
// 	 *            unique identifier for a spout or bolt instance
// 	 */
// 	void setOriginId(String originId);
// 
// 	/**
// 	 * Check if this MetaParticle is equal to another {@link Particle}.
// 	 * {@link MetaParticle}s need to implement their own equals method, which
// 	 * ignores the originId. This method is used by the {@link SyncBuffer} in
// 	 * order to remove duplicate {@link MetaParticle}s.
// 	 * 
// 	 * @param other
// 	 *            Other MetaParticle used for comparison
// 	 * @return True if this object is equal to another instance of a
// 	 *         MetaParticle, with the exception of the originId.
// 	 */
// 	boolean equalMetaParticle(MetaParticle other);
// }
// 
// Path: SensorStorm/src/nl/tno/sensorstorm/api/processing/Operation.java
// public interface Operation extends Serializable {
// 
// 	/**
// 	 * Initialize this operation.
// 	 * 
// 	 * @param fieldGrouperValue
// 	 *            Value for the grouping field if any, otherwise null
// 	 * @param startTimeStamp
// 	 *            The time of the first particle this operation will get. Or -1
// 	 *            if a single operation is created for all particles (in the
// 	 *            SensorStormBolt constructor the fieldgrouperid == null)
// 	 * @param stormNativeConfig
// 	 *            A reference to the storm config object
// 	 * @param zookeeperStormConfiguration
// 	 *            A reference to the {@link ExternalStormConfiguration}
// 	 * @throws OperationException
// 	 *             If an error occurs in the {@link Operation}
// 	 * 
// 	 */
// 	void init(String fieldGrouperValue, long startTimeStamp,
// 			@SuppressWarnings("rawtypes") Map stormNativeConfig,
// 			ExternalStormConfiguration zookeeperStormConfiguration)
// 			throws OperationException;
// 
// }


// Path: SensorStorm/src/nl/tno/sensorstorm/impl/SyncBuffer.java
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.TreeSet;
import nl.tno.sensorstorm.api.particles.MetaParticle;
import nl.tno.sensorstorm.api.particles.Particle;
import nl.tno.sensorstorm.api.processing.Operation;

package nl.tno.sensorstorm.impl;



/**
 * Class that synchronizes {@link Particle}s coming from multiple (probably
 * parallel executing) sources and filters out duplicate {@link MetaParticle}s.
 * This buffer introduces a configurable delay.
 * <p>
 * When the delay is 0 the buffer will always immediately return the same
 * particle. Warning: When the size is 0, duplicate MetaParticles will not be
 * filtered out!
 */
public class SyncBuffer {

	private static final int INITIAL_CAPACITY_RETURN_LIST = 5;

	/**
	 * Special {@link Comparator} used by the {@link SyncBuffer}. It sorts
	 * {@link Particle}s based on their timestamp. Additionally, duplicate
	 * {@link MetaParticle}s (with possibly a different originId) will return 0,
	 * which results in removing the duplicates from the TreeSet.
	 */
	private static class ParticleComparator implements Comparator<Particle> {

		@Override
		public int compare(Particle p1, Particle p2) {
			int comp = Long.compare(p1.getTimestamp(), p2.getTimestamp());
			if (comp == 0) {
				// Same timestamp
				if ((p1 instanceof MetaParticle)
						&& (p2 instanceof MetaParticle)