Scalable Job

package tutorial;

import java.util.Arrays;
import java.util.List;

class MergeSort {

	private static void merge(List<Integer> arr, int left, int middle, int right) {

		int low = middle - left + 1;
		int high = right - middle;

		int[] L = new int[low];
		int[] R = new int[high];

		int i = 0, j = 0;

		for (i = 0; i < low; i++) {
			L[i] = arr.get(left + i);
		}

		for (j = 0; j < high; j++) {
			R[j] = arr.get(middle + 1 + j);
		}

		int k = left;
		i = 0;
		j = 0;

		while (i < low && j < high) {
			if (L[i] <= R[j]) {
				arr.set(k, L[i]);
				i++;
			} else {
				arr.set(k, R[j]);
				j++;
			}
			k++;
		}

		while (i < low) {
			arr.set(k, L[i]);
			i++;
			k++;
		}

		while (j < high) {
			arr.set(k, R[j]);
			j++;
			k++;
		}
	}

	private static void mergeSort(List<Integer> arr, int left, int right) {
		int middle;
		if (left < right) {

			middle = (left + right) / 2;

			mergeSort(arr, left, middle); // left subarray
			mergeSort(arr, middle + 1, right); // right subarray

			merge(arr, left, middle, right); // merge the two subarrays
		}
	}

	public static void sort(List<Integer> arr) {
		mergeSort(arr, 0, arr.size() - 1);
	}

}

def queue = grid.objQueue("MSORT_QUEUE"); //create or get the queue

def lines = msg.lines(); //split the incoming data into list of lines
int count = 0;

for(line in lines){
    
    def csv = line.trim();
    
    if("".equals(csv)) continue; //if blank skip it
    
    def vals = csv.split(",");
    
    def iarray = [];
    
    for(val in vals){
        iarray.add(Integer.valueOf(val))
    }
    
    queue.offer(iarray);
    
    ++count;
}

log.info("%d arrays feeded", count);

import tutorial.MergeSort;

log.info("MergeSorter running...");

def queue = grid.objQueue("MSORT_QUEUE");
def al = grid.atomicLong("MSORT_QUEUE_JOB");
def JOBID = al.incrementAndGet();

try{

    while(THIS.isRunning()){
        
        def narray = queue.take(); //wait for data
        
        try{
            
            MergeSort.sort(narray); //do the sorting
        
            log.info("Node[%d][%s] Sorted: %s", JOBID, THIS.getNodeId(), narray);
        
        }catch(Exception ex){
            log.error(ex);
        }
        
    }
    
}finally{
    al.decrementAndGet();
}

log.info("MergeSorter stopped.")