/**
 * 
 */
package ch.uzh.ddis.katts.bolts.aggregate;

import java.util.Date;
import java.util.Map;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import ch.uzh.ddis.katts.bolts.AbstractSynchronizedBolt;
import ch.uzh.ddis.katts.bolts.Event;
import ch.uzh.ddis.katts.bolts.VariableBindings;
import ch.uzh.ddis.katts.bolts.join.SimpleVariableBindings;
import ch.uzh.ddis.katts.query.processor.aggregate.AggregateConfiguration;
import ch.uzh.ddis.katts.query.processor.aggregate.AggregatorConfiguration;
import ch.uzh.ddis.katts.query.stream.Stream;
import ch.uzh.ddis.katts.query.stream.Variable;

import com.google.common.collect.ImmutableList;
import com.google.common.collect.Table;

// TODO: make sure data gets output even when only the heartbeat is advanced

/**
 * This is the implementation of the AggregateBolt as documented in {@link AggregateConfiguration}.
 * 
 * @author "Lorenz Fischer" <lfischer@ifi.uzh.ch>
 */
public class AggregateBolt extends AbstractSynchronizedBolt {

	/** The object containing all confguration details of this bolt. */
	private AggregateConfiguration configuration;

	/** The aggregator manager takes care of all the aggregates we compute. */
	private AggregatorManager aggregatorManager;

	/**
	 * This list contains all field names over which we need to group
	 */
	private ImmutableList<String> groupByFieldNames;

	/**
	 * This key will be used to get the default list of aggregators from teh aggregators map in the case that no
	 * group-by attribute has been specified in the configuration.
	 */
	private final ImmutableList<Object> nonGroupedAggegateKey = ImmutableList.of();

	public AggregateBolt(AggregateConfiguration configuration) {
		this.configuration = configuration;
	}

	@Override
	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		super.prepare(stormConf, context, collector);

		AggregatorConfiguration<?>[] configs;
		configs = new AggregatorConfiguration[configuration.getAggregators().size()];

		this.aggregatorManager = new AggregatorManager(this.configuration.getWindowSize(),
				this.configuration.getOutputInterval(), new AggregatorManager.Callback() {

					@Override
					public void callback(Table<ImmutableList<Object>, String, Object> aggregateValues, Date startDate,
							Date endDate) {

						for (ImmutableList<Object> groupingKeyValues : aggregateValues.rowKeySet()) {
							for (String aggregateName : aggregateValues.columnKeySet()) {
								// create the simple variable bindings object where we copy the values from
								SimpleVariableBindings bindings = new SimpleVariableBindings();

								bindings.put("startDate", startDate);
								bindings.put("endDate", endDate);

								// create the values for the group-by fields
								for (int i = 0; i < AggregateBolt.this.groupByFieldNames.size(); i++) {
									bindings.put(AggregateBolt.this.groupByFieldNames.get(i), groupingKeyValues.get(i));
								}

								// now add teh actual aggregate value.. finally!
								bindings.put(aggregateName, aggregateValues.get(groupingKeyValues, aggregateName));

								// we loop over each stream and create the bindings object to emit
								for (Stream stream : getStreams()) {
									// TODO: remove that anchor event stuff. for now using null
									VariableBindings bindingsToEmit = getEmitter().createVariableBindings(stream, null);

									/*
									 * Copy all values from the bindings object into the outgoing binding. This includes
									 * the inherited variables.
									 */
									for (Variable variable : stream.getAllVariables()) {
										if (bindings.get(variable.getReferencesTo()) == null) {
											throw new NullPointerException("Could not find a value for variable \""
													+ variable.getReferencesTo() + "\" in the bindings object.");
										}
										bindingsToEmit.add(variable, bindings.get(variable.getReferencesTo()));
									}
									bindingsToEmit.setStartDate((Date) bindings.get("startDate"));
									bindingsToEmit.setEndDate((Date) bindings.get("endDate"));

									bindingsToEmit.emit();
								}

							}
						}

						// set the date that will be propagated by the heartbeat
						setLastDateProcessed(endDate);
					}
				}, this.configuration.isOnlyIfChanged(), this.configuration.getAggregators().toArray(configs));

		if (this.configuration.getGroupBy() != null) {
			this.groupByFieldNames = ImmutableList.copyOf(this.configuration.getGroupBy().split(","));
		}
	}

	@Override
	protected void setLowestCurrentHeartBeat(Date value) {
		super.setLowestCurrentHeartBeat(value);
		
		this.aggregatorManager.advanceInTime(value.getTime());
	}
	
	@Override
	public void execute(Event event) {
		SimpleVariableBindings bindings = new SimpleVariableBindings(event.getTuple());
		ImmutableList<Object> groupByKey;

		if (groupByFieldNames == null) { // no grouping configured
			groupByKey = this.nonGroupedAggegateKey;
		} else {
			// create the group-by key
			ImmutableList.Builder<Object> builder = ImmutableList.builder();
			for (Object fieldName : this.groupByFieldNames) {
				builder.add(bindings.get(fieldName));
			}
			groupByKey = builder.build();
		}

		this.aggregatorManager.incorporateValue(groupByKey, bindings);
	}

	@Override
	public String getId() {
		return this.configuration.getId();
	}

}

@@ -1795,16 +1795,46 @@
 prepare(
+@SuppressWarnings(%22rawtypes%22) 
 Map stor
@@ -4456,96 +4456,228 @@
 %0A%09%09%09
-this.groupByFieldNames = ImmutableList.copyOf(this.configuration.getGroupBy().split(%22,%22)
+ImmutableList.Builder%3CString%3E builder = ImmutableList.builder();%0A%09%09%09for (String fieldName : this.configuration.getGroupBy().split(%22,%22)) %7B%0A%09%09%09%09builder.add(fieldName.trim());%0A%09%09%09%7D%0A%09%09%09this.groupByFieldNames = builder.build(
 );%0A%09
@@ -4796,11 +4796,9 @@
 e);%0A
-%09%09
 %0A
+
 %09%09th
@@ -4853,17 +4853,16 @@
 ());%0A%09%7D%0A
-%09
 %0A%09@Overr

remove white spaces from fieldnames in groupBy field