// Path: stormcv/src/main/java/nl/tno/stormcv/model/CVParticle.java
// public abstract class CVParticle implements Comparable<CVParticle>{
// 	
// 	private Tuple tuple;
// 	private long requestId = -1;
// 	private String streamId;
// 	private long sequenceNr;
// 	private HashMap<String, Object> metadata = new HashMap<String, Object>();
// 	
// 	/**
// 	 * Constructs a generic type based on the provided tuple. The tuple must contain streamID and sequenceNR
// 	 * values. 
// 	 * @param tuple
// 	 */
// 	@SuppressWarnings("unchecked")
// 	public CVParticle(Tuple tuple){
// 		this(tuple.getStringByField(CVParticleSerializer.STREAMID), 
// 				tuple.getLongByField(CVParticleSerializer.SEQUENCENR));
// 		this.tuple = tuple;
// 		this.setRequestId(tuple.getLongByField(CVParticleSerializer.REQUESTID));
// 		this.setMetadata((HashMap<String, Object>)tuple.getValueByField(CVParticleSerializer.METADATA));
// 	}
// 	
// 	/**
// 	 * Constructs a GenericType object for some piece of information regarding a stream
// 	 * @param streamId the id of the stream
// 	 * @param sequenceNr the sequence number of the the information (used for ordering)
// 	 */
// 	public CVParticle(String streamId, long sequenceNr){
// 		this.streamId = streamId;
// 		this.sequenceNr = sequenceNr;
// 	}
// 	
// 	public String getStreamId(){
// 		return streamId;
// 	}
// 	
// 	public long getSequenceNr(){
// 		return sequenceNr;
// 	}
// 
// 	public Tuple getTuple() {
// 		return tuple;
// 	}
// 
// 	public HashMap<String, Object> getMetadata() {
// 		return metadata;
// 	}
// 	
// 	public void setMetadata(HashMap<String, Object> metadata) {
// 		if(metadata != null){
// 			this.metadata = metadata;
// 		}
// 	}
// 	
// 	public void setRequestId(long id){
// 		this.requestId = id;
// 	}
// 	
// 	public long getRequestId() {
// 		return requestId;
// 	}
// 	
// 	/**
// 	 * Compares one generictype to another based on their sequence number.
// 	 * @return -1 if this.sequenceNr < other.sequenceNr, 0 if this.sequenceNr == other.sequenceNr else 1
// 	 */
// 	@Override
// 	public int compareTo(CVParticle other){
// 		return (int)(getSequenceNr() - other.getSequenceNr());
// 	}
// 
// 
// }
// 
// Path: stormcv/src/main/java/nl/tno/stormcv/operation/ISingleInputOperation.java
// public interface ISingleInputOperation <Output extends CVParticle> extends IOperation<Output>{
// 
// 	public List<Output> execute(CVParticle particle) throws Exception;
// 	
// }


import java.util.List;
import java.util.Map;
import nl.tno.stormcv.model.CVParticle;
import nl.tno.stormcv.operation.ISingleInputOperation;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
package nl.tno.stormcv.bolt;



/**
 * A basic {@link CVParticleBolt} implementation that works with single items received (hence maintains no
 * history of items received). The bolt will ask the provided {@link ISingleInputOperation} implementation to
 * work on the input it received and produce zero or more results which will be emitted by the bolt. 
 * If an operation throws an exception the input will be failed in all other situations the input will be acked.
 * 
 * @author Corne Versloot
 *
 */
public class SingleInputBolt extends CVParticleBolt {
	
	private static final long serialVersionUID = 8954087163234223475L;

	private ISingleInputOperation<? extends CVParticle> operation;

	/**
	 * Constructs a SingleInputOperation 
	 * @param operation the operation to be performed
	 */
// Path: stormcv/src/main/java/nl/tno/stormcv/model/CVParticle.java
// public abstract class CVParticle implements Comparable<CVParticle>{
// 	
// 	private Tuple tuple;
// 	private long requestId = -1;
// 	private String streamId;
// 	private long sequenceNr;
// 	private HashMap<String, Object> metadata = new HashMap<String, Object>();
// 	
// 	/**
// 	 * Constructs a generic type based on the provided tuple. The tuple must contain streamID and sequenceNR
// 	 * values. 
// 	 * @param tuple
// 	 */
// 	@SuppressWarnings("unchecked")
// 	public CVParticle(Tuple tuple){
// 		this(tuple.getStringByField(CVParticleSerializer.STREAMID), 
// 				tuple.getLongByField(CVParticleSerializer.SEQUENCENR));
// 		this.tuple = tuple;
// 		this.setRequestId(tuple.getLongByField(CVParticleSerializer.REQUESTID));
// 		this.setMetadata((HashMap<String, Object>)tuple.getValueByField(CVParticleSerializer.METADATA));
// 	}
// 	
// 	/**
// 	 * Constructs a GenericType object for some piece of information regarding a stream
// 	 * @param streamId the id of the stream
// 	 * @param sequenceNr the sequence number of the the information (used for ordering)
// 	 */
// 	public CVParticle(String streamId, long sequenceNr){
// 		this.streamId = streamId;
// 		this.sequenceNr = sequenceNr;
// 	}
// 	
// 	public String getStreamId(){
// 		return streamId;
// 	}
// 	
// 	public long getSequenceNr(){
// 		return sequenceNr;
// 	}
// 
// 	public Tuple getTuple() {
// 		return tuple;
// 	}
// 
// 	public HashMap<String, Object> getMetadata() {
// 		return metadata;
// 	}
// 	
// 	public void setMetadata(HashMap<String, Object> metadata) {
// 		if(metadata != null){
// 			this.metadata = metadata;
// 		}
// 	}
// 	
// 	public void setRequestId(long id){
// 		this.requestId = id;
// 	}
// 	
// 	public long getRequestId() {
// 		return requestId;
// 	}
// 	
// 	/**
// 	 * Compares one generictype to another based on their sequence number.
// 	 * @return -1 if this.sequenceNr < other.sequenceNr, 0 if this.sequenceNr == other.sequenceNr else 1
// 	 */
// 	@Override
// 	public int compareTo(CVParticle other){
// 		return (int)(getSequenceNr() - other.getSequenceNr());
// 	}
// 
// 
// }
// 
// Path: stormcv/src/main/java/nl/tno/stormcv/operation/ISingleInputOperation.java
// public interface ISingleInputOperation <Output extends CVParticle> extends IOperation<Output>{
// 
// 	public List<Output> execute(CVParticle particle) throws Exception;
// 	
// }


// Path: stormcv/src/main/java/nl/tno/stormcv/bolt/SingleInputBolt.java
import java.util.List;
import java.util.Map;
import nl.tno.stormcv.model.CVParticle;
import nl.tno.stormcv.operation.ISingleInputOperation;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;

package nl.tno.stormcv.bolt;



/**
 * A basic {@link CVParticleBolt} implementation that works with single items received (hence maintains no
 * history of items received). The bolt will ask the provided {@link ISingleInputOperation} implementation to
 * work on the input it received and produce zero or more results which will be emitted by the bolt. 
 * If an operation throws an exception the input will be failed in all other situations the input will be acked.
 * 
 * @author Corne Versloot
 *
 */
public class SingleInputBolt extends CVParticleBolt {
	
	private static final long serialVersionUID = 8954087163234223475L;

	private ISingleInputOperation<? extends CVParticle> operation;

	/**
	 * Constructs a SingleInputOperation 
	 * @param operation the operation to be performed
	 */
	public SingleInputBolt(ISingleInputOperation<? extends CVParticle> operation){