package org.slc.sli.ingestion.routes.orchestra;

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

import org.apache.camel.Exchange;
import org.apache.camel.processor.aggregate.AggregationStrategy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.slc.sli.ingestion.WorkNote;

/**
 * WorkNote aggregator to be used from camel
 *
 * @author dduran
 *
 */
public class WorkNoteAggregator implements AggregationStrategy {

    private static final Logger LOG = LoggerFactory.getLogger(WorkNoteAggregator.class);

    @Override
    public Exchange aggregate(Exchange hasBeenAggregated, Exchange toBeAggregated) {

        WorkNote workNote = toBeAggregated.getIn().getBody(WorkNote.class);
        if (workNote != null) {

            // on the first message, hasBeenAggregated will be null so "intialize" it
            if (hasBeenAggregated == null) {
                hasBeenAggregated = initializeAggregationExchange(toBeAggregated);
            }

            @SuppressWarnings("unchecked")
            List<WorkNote> workNoteList = hasBeenAggregated.getIn().getBody(List.class);

            workNoteList.add(workNote);

            LOG.info("aggregated WorkNote: {} ", workNote);
        }

        return hasBeenAggregated;
    }

    private Exchange initializeAggregationExchange(Exchange toBeAggregated) {

        Exchange hasBeenAggregated = toBeAggregated;

        hasBeenAggregated.getIn().setBody(new ArrayList<WorkNote>(), List.class);

        return hasBeenAggregated;
    }
}

@@ -289,16 +289,65 @@
 orkNote;
+%0Aimport org.slc.sli.ingestion.queues.MessageType;
 %0A%0A/**%0A *
@@ -1388,53 +1388,628 @@
     
-Exchange hasBeenAggregated = toBeAggregated;%0A
+WorkNote workNote = toBeAggregated.getIn().getBody(WorkNote.class);%0A%0A        LOG.info(%22Setting %7B%7D as the aggregation completion size for %7B%7D%22, workNote.getBatchSize(),%0A                workNote.getIngestionStagedEntity());%0A%0A        Exchange hasBeenAggregated = toBeAggregated;%0A%0A        if (MessageType.DATA_TRANSFORMATION.name().equals(toBeAggregated.getIn().getHeader(%22IngestionMessageType%22))) %7B%0A            hasBeenAggregated.getIn().setHeader(%22workNoteByEntityCount%22, workNote.getBatchSize());%0A            hasBeenAggregated.getIn().setHeader(%22totalWorkNoteCount%22, toBeAggregated.getProperty(%22CamelSplitSize%22));%0A        %7D
 %0A   

set the fields in the exchange message by which we will aggregate.