/*
 * (c) Copyright 2005-2008 Pervasive Software Inc. All rights reserved.
 */
package example.join;

import com.pervasive.datarush.config.CliConfig;
import com.pervasive.datarush.config.CliHelper;
import com.pervasive.datarush.config.annotations.Param;
import com.pervasive.datarush.executor.DataflowGraphBase;
import com.pervasive.datarush.operators.io.textfile.DelimitedTextProperties;
import com.pervasive.datarush.operators.io.textfile.ReadDelimitedText;
import com.pervasive.datarush.operators.io.textfile.WriteDelimitedText;
import com.pervasive.datarush.operators.join.JoinMode;
import com.pervasive.datarush.operators.join.JoinSortedRows;

/**
 * Join sample application
 * 

 * This sample illustrates basic DataRush development, including
 * 

 * 
 Reusing operator provided in the standard operator library.
 * 
 Composing a dataflow graph by adding operators and connecting flows
 * 
 Using the CLI and configuration convenience classes
 * 
 * 

 * See also {@link JoinSampleHardcoded} for a "simpler", non-configurable
 * example that does not use a properties file and does not take advantage of
 * the CLI and configuration convenience classes provided with the SDK.
 *
 * @author Pervasive Software
 */
public class JoinSample extends DataflowGraphBase
{
    private static final String[] JSON_PROPERTY_KEYS = {
        "leftInputProperties", "rightInputProperties", "joinKeys", 
        "outputProperties"
    };

    public static void main(String[] args) {
        // Parse command-line flags, and read and parse properties file
        CliConfig cli = CliHelper.processArgs(args, JSON_PROPERTY_KEYS);
        // Reflectively construct a configured JoinSample, and run it
        CliHelper.run(cli, JoinSample.class);
    }

    public JoinSample (
            @Param("leftInputProperties") DelimitedTextProperties 
                    leftInputProperties,
            @Param("rightInputProperties") DelimitedTextProperties 
                    rightInputProperties,
            @Param("joinMode") JoinMode joinMode,
            @Param("joinKeys") String[] keyNames,
            @Param("outputProperties") DelimitedTextProperties 
                    outputProperties) {

        // Add reader for left hand side of join
        ReadDelimitedText readLeft = add(
                new ReadDelimitedText(leftInputProperties), "readLeft");

        // Add reader for right hand side of join
        ReadDelimitedText readRight = add(
                new ReadDelimitedText(rightInputProperties), "readRight");

        // Add join
        JoinSortedRows join = add(
                new JoinSortedRows(readLeft.getOutput(), readRight.getOutput(), 
                keyNames, keyNames, joinMode), "join");

        // Add writer for output of join
        add(new WriteDelimitedText(join.getOutput(), outputProperties), 
        "write");
    }
}