public final class AdderComposite extends CompositeOperator implements RecordPipelineOperator {

    //required: declare input ports
    private final RecordPort input = newRecordInput("input");
    //required: declare output ports
    private final RecordPort output = newRecordOutput("output");
    //required: declare properties
    private String inputFieldName1 = "field1";
    private String inputFieldName2 = "field2";
    private String sumFieldName = "sum";
    
    /**
     * Default constructor
     */
    //required: all operators must have a default constructor
    public AdderComposite() {
    }
    
    /**
     * Creates an adder for the specified fields
     * @param inputFieldName1 the first input field
     * @param inputFieldName2 the second input field
     * @param sumFieldName the output field that will contain the sum
     */
    //optional: convenience constructor
    public AdderComposite(String inputFieldName1, String inputFieldName2, String sumFieldName) {
        setInputFieldName1(inputFieldName1);
        setInputFieldName2(inputFieldName2);
        setSumFieldName(sumFieldName);
    }
    
    ///////////////////////////////////////////////////
    //
    // Required: getters and setters for each property
    //
    ///////////////////////////////////////////////////
    /**
     * Returns the name of the first input field to be added
     * @return the name of the first input field
     */
    public String getInputFieldName1() {
        return inputFieldName1;
    }
    
    /**
     * Sets the name of the first input field to be added
     * @param inputFieldName1 the name of the first input field
     */
    public void setInputFieldName1(String inputFieldName1) {
        this.inputFieldName1 = inputFieldName1;
    }
    
    /**
     * Returns the name of the second input field to be added
     * @return the name of the second input field
     */
    public String getInputFieldName2() {
        return inputFieldName2;
    }
    
    /**
     * Sets the name of the second input field to be added
     * @param inputFieldName2 the name of the second input field
     */
    public void setInputFieldName2(String inputFieldName2) {
        this.inputFieldName2 = inputFieldName2;
    }
    
    /**
     * Returns the name of the output field that will contain the sum of the two fields.
     * @return the name of the output field that will contain the sum of the two fields.
     */
    public String getSumFieldName() {
        return sumFieldName;
    }
   
    /**
     * Sets the name of the output field that will contain the sum of the two fields.
     * @param sumFieldName the name of the output field that will contain the sum of the two fields.
     */
    public void setSumFieldName(String sumFieldName) {
        this.sumFieldName = sumFieldName;
    }

    ///////////////////////////////////////////////////
    //
    // Required: getters for each port
    //
    ///////////////////////////////////////////////////
    /**
     * Returns the input port containing the data to add
     */
    @Override
    public RecordPort getInput() {
        return input;
    }

    /**
     * Returns the output port containing the original data from the input
     * plus an additional field containing the sum
     */
    @Override
    public RecordPort getOutput() {
        return output;
    }

    private void validateInput(MetadataContext ctx) {

        //make sure specified properties are not empty
        PropertyUtil.checkNotEmpty("inputFieldName1", inputFieldName1);
        PropertyUtil.checkNotEmpty("inputFieldName2", inputFieldName2);
        PropertyUtil.checkNotEmpty("sumFieldName", sumFieldName);

        //make sure inputFieldName1 and inputFieldName2 are defined in the input
        RecordTokenType inputType = input.getType(ctx);
        inputType.verifyNames(inputFieldName1,inputFieldName2);
    }

    @Override
    protected void compose(CompositionContext ctx) {

        //best practice: perform any input validation: should be done first
        validateInput(ctx);

        //define a FieldDerivation to derive a new field equal to the
        //sum of two other fields
        FieldDerivation sum =
            FieldDerivation.derive(sumFieldName, 
                    Arithmetic.add(inputFieldName1, inputFieldName2));

        //add DeriveFields to the graph
        DeriveFields derive = ctx.add(new DeriveFields(sum));     

        //connect the input of this operator to the input of DeriveFields
        ctx.connect(input, derive.getInput());

        //required: compose must connect sources to all output ports
        //  in this case, we connect the output of DeriveFields. 
        ctx.connect(derive.getOutput(), output);
    }
}