package weatherhandler.processor;

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

import weatherhandler.data.Measurement;

/**
 * Processor that batches updates together in larger chunks for possibly quicker
 * further processing. It does _NOT_ guarantee that exactly `batchSize` updates
 * are passed through--just that _AT LEAST_ `batchSize` updates are passed
 * through to the next processor.
 */
public class BatchUpdatesProcessor implements Processor {
    // Default amount of measurements per batch
    final private static int DEFAULT_BATCH_SIZE = 1000;
    // Added to initial list size to avoid resizing the allocated list when only
    // a few too many items come in at once
    final private int BATCH_BUFFER = 50;

    // Amount of Measurements per batch
    private int batchSize;
    // Current batch
    private List<Measurement> batch;
    // Processor to pass batched data to
    private Processor next;

    public BatchUpdatesProcessor(Processor next) {
        this(DEFAULT_BATCH_SIZE, next);
    }
    public BatchUpdatesProcessor(int batchSize, Processor next) {
        this.next = next;
        this.batchSize = batchSize;
        this.batch = new ArrayList<Measurement>(batchSize + BATCH_BUFFER);
    }

    /**
     * Batch & push new updates.
     */
    public void processMeasurements(List<Measurement> measurements) throws ProcessorException {
        this.batch.addAll(measurements);
        if (this.batch.size() >= this.batchSize) {
            // Replace batch by a new list immediately, so threads can read from
            // this batch and other threads can write to the next batch
            List<Measurement> current = this.batch;
            this.batch = new ArrayList<Measurement>(this.batchSize + BATCH_BUFFER);
            this.next.processMeasurements(current);
        }
    }
}

@@ -1249,32 +1249,117 @@
   * 
-Batch & push new updates
+Concurrently append new items to the current batch. Replace the batch if%0A     * the current one is %22complete%22
 .%0A  
@@ -1373,26 +1373,33 @@
    p
-ublic void process
+rivate synchronized List%3C
 Meas
@@ -1405,17 +1405,24 @@
 surement
-s
+%3E append
 (List%3CMe
@@ -1449,34 +1449,8 @@
 nts)
- throws ProcessorException
  %7B%0A 
@@ -1845,45 +1845,356 @@
     
-this.next.processMeasurements(current
+return current;%0A        %7D%0A        return null;%0A    %7D%0A%0A    /**%0A     * Batch & push new updates.%0A     */%0A    public void processMeasurements(List%3CMeasurement%3E measurements) throws ProcessorException %7B%0A        List%3CMeasurement%3E nextBatch = this.append(measurements);%0A        if (nextBatch != null) %7B%0A            this.next.processMeasurements(nextBatch
 );%0A 

move actual batching work to a synchronized method...