Sorting rows

The sample program sorts all input rows based on two properties that are fetched from the Java Transformer stage using the Stage.getUserProperties() method. The User's Properties provide the sort column and order:

• descending = false
• nKeyIndex = 5

Looping on the readRow() method is demonstrated, which might exhaust the available rows on an input link and generate a LinkNotReadyException. Returning from the process() method between two calls of readRow() allows the system to fetch the next row.

package com.ascentialsoftware.jds.test;

import com.ascentialsoftware.jds.Column;
import com.ascentialsoftware.jds.Row;
import com.ascentialsoftware.jds.Stage;

import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Properties;

/**
 * Example of transformer that sorts all incoming rows by the value
 * of a column whose index is stored in the stage's user properties.
 */
public class Sorter extends Stage
{
    /**
     *  Reads the user properties of the stage, returned as a single string.
     *  In this case, the string contains a set of actual properties, in the
     *  form of a list of key-value pairs, easy to process using Java's
     *  Properties class.
     *
     *  Example of such a string: "descending = false\nkeyIndex = 5"
     */
    public void initialize(){
        byte[]      userProperties = getUserProperties().getBytes();
        InputStream propertyStream = new ByteArrayInputStream(userProperties);
        Properties  properties     = new Properties();
        String      propertyValue;

        try {
            properties.load(propertyStream);
        } catch (IOException eIO) {
            //  Should never happen.
        }

        //  This property tells whether the rows must be sorted
        //  in an ascending or a descending order.
        propertyValue = properties.getProperty("descending");
        _descending   = (propertyValue == null)
                         ? false
                         : propertyValue.equals("true");
						
        //  This property gives the index of the column that contains
        //  the key by which each row will be sorted.
        propertyValue = properties.getProperty("keyIndex");
        _keyIndex     = (propertyValue == null)
                         ? 0
                         : Integer.parseInt(propertyValue);

        _keys = new ArrayList();
        _rows = new HashMap();
    }

    /**
     *  Loops on reading input rows as long as new rows are available.
     *  Rows are stored in an hash map. When no more rows are available,
     *  the stored rows are sorted and written on the output link,
     *  in the right order.
     *
     *  Note: the same thing could be achieved by storing only one row
     *  every time this method is called. Looping on all rows within
     *  one call is more efficient, but it is recommended to give
     *  the hand back to the system from time to time.
     */
    public int process()    {
        Row inputRow=null;
        
        do {
			if (inputRow != null) 
			{
				String key = inputRow.getValueAsString(_keyIndex);
				_keys.add(key);
				_rows.put(key, inputRow);
			} 
				inputRow = readRow();
        }
        while (inputRow != null);

        Collections.sort(_keys);

        int keyCount = _keys.size();
        if (_descending) {
            for (int keyNumber = keyCount - 1; keyNumber >= 0; keyNumber--) {
                Row outputRow = (Row)_rows.get(_keys.get(keyNumber));
                writeRow(outputRow);
            }
        } else {
            for (int keyNumber = 0; keyNumber < keyCount; keyNumber++) {
                Row outputRow = (Row)_rows.get(_keys.get(keyNumber));
                writeRow(outputRow);
            }
        }
			 
		_keys.clear();
        return OUTPUT_STATUS_READY;
    }
   
    private boolean   _descending;
    private int       _keyIndex;
    private ArrayList _keys;
    private HashMap   _rows;
}