package ch.uzh.ddis.katts.query.processor;

import java.util.ArrayList;
import java.util.List;

import javax.xml.bind.annotation.XmlElement;
import javax.xml.bind.annotation.XmlElementWrapper;
import javax.xml.bind.annotation.XmlTransient;

import ch.uzh.ddis.katts.query.AbstractNode;
import ch.uzh.ddis.katts.query.stream.Producers;
import ch.uzh.ddis.katts.query.stream.Stream;
import ch.uzh.ddis.katts.query.stream.StreamConsumer;

/**
 * This class implements an abstract processor. It provides convenient configuration
 * methods for consuming, processing and emitting data.
 * 
 * These are primarily the handling of the stream configurations.
 * 
 * @author Thomas Hunziker
 *
 */
public abstract class AbstractProcessor extends AbstractNode implements Processor{

	private static final long serialVersionUID = 1L;

	@XmlTransient
	private List<StreamConsumer> consumers = new ArrayList<StreamConsumer>();
	
	@XmlTransient
	private List<Stream> producers = new Producers(this);
	
	@XmlElementWrapper(name="consumes")
	@XmlElement(name="stream")
	@Override
	public List<StreamConsumer> getConsumers() {
		for (StreamConsumer consumer : consumers) {
			consumer.setNode(this);
		}
		
		return consumers;
	}
	
	@XmlElementWrapper(name="produces")
	@XmlElement(name="stream")
	@Override
	public List<Stream> getProducers() {
		for (Stream producer : producers) {
			producer.setNode(this);
		}
		
		return producers;
	}

	public void setConsumers(List<StreamConsumer> consumers) {
		this.consumers = consumers;
	}

	public void setProducers(List<Stream> producers) {
		this.producers = producers;
	}
	
	public void appendConsumer(StreamConsumer consumer) {
		this.getConsumers().add(consumer);
	}
	
	public void appendProducer(Stream producer) {
		this.getProducers().add(producer);
	}
	
	@Override
	@XmlTransient
	public int getParallelism() {
		// We assume that a regular processor is fully parallelizable  
		return 0;
	}
	
}

@@ -85,24 +85,71 @@
 util.List;%0A%0A
+import javax.xml.bind.annotation.XmlAttribute;%0A
 import javax
@@ -952,24 +952,135 @@
 sumer%3E();%0A%09%0A
+%09// We assume that a regular processor is fully parallelizable %0A%09@XmlTransient%0A%09private int parallelism = 0;%0A%09%0A
 %09@XmlTransie
@@ -1959,25 +1959,27 @@
 de%0A%09@Xml
-Transient
+Attribute()
 %0A%09public
@@ -2008,87 +2008,118 @@
 %7B%0A%09%09
-// We assume that a regular processor is fully parallelizable  %0A%09%09return 0
+return this.parallelism;%0A%09%7D%0A%09%0A%09public void setParallelism(int paralleism) %7B%0A%09%09this.parallelism = paralleism
 ;%0A%09%7D%0A
-%09%0A
 %7D%0A

Add option to override the parallelization for processing nodes (joins, functions, filter and aggregates).