/*
 * LICENSE AGREEMENT
 * 
 * GRIDGAIN 1.0 - GRID COMPUTING FOR JAVA.
 * COPYRIGHT (C) 2005-2007 GRIDGAIN SYSTEMS. ALL RIGHTS RESERVED.
 * 
 * THIS IS FREE SOFTWARE; YOU CAN REDISTRIBUTE IT AND/OR
 * MODIFY IT UNDER THE TERMS OF THE GNU LESSER GENERAL PUBLIC
 * LICENSE AS PUBLISHED BY THE FREE SOFTWARE FOUNDATION; EITHER
 * VERSION 2.1 OF THE LICENSE, OR (AT YOUR OPTION) ANY LATER 
 * VERSION.
 * 
 * THIS LIBRARY IS DISTRIBUTED IN THE HOPE THAT IT WILL BE USEFUL,
 * BUT WITHOUT ANY WARRANTY; WITHOUT EVEN THE IMPLIED WARRANTY OF
 * MERCHANTABILITY OR FITNESS FOR A PARTICULAR PURPOSE.  SEE THE 
 * GNU LESSER GENERAL PUBLIC LICENSE FOR MORE DETAILS.
 * 
 * YOU SHOULD HAVE RECEIVED A COPY OF THE GNU LESSER GENERAL PUBLIC
 * LICENSE ALONG WITH THIS LIBRARY; IF NOT, WRITE TO THE FREE 
 * SOFTWARE FOUNDATION, INC., 51 FRANKLIN ST, FIFTH FLOOR, BOSTON, MA  
 * 02110-1301 USA
 */


package org.gridgain.examples.helloworld.gridify.task;

import java.io.*;
import java.util.*;
import org.gridgain.grid.*;
import org.gridgain.grid.gridify.*;

/**
 * This is a grid job implementing split and aggregate logic for this grid task.
 *
 * @author 2005-2007 Copyright (C) GridGain Systems. All Rights Reserved.
 * @version 1.0.0-07022007
 */
public class GridifyHelloWorldJob extends GridifyJobAdapter {
    /** Injected job context. */
    @GridJobContextResource
    private GridJobContext jobCtx = null;

    /**
     * @see GridJob#split(Serializable, Collection)
     */
    public Collection<GridJobReference> split(Serializable arg, Collection<GridNode> subGrid) {
        GridifyArgument gridifyArg = (GridifyArgument)arg;
        
        // Only split if this is a root job and there are other available nodes.
        if (jobCtx.isRootJob() == true && subGrid.isEmpty() == false) {
            // Split the passed in phrase into multiple words.
            String[] words = ((String)gridifyArg.getMethodParameters()[0]).split(" ");
            
            List<GridJobReference> refs = new ArrayList<GridJobReference>(words.length);
            
            Iterator<GridNode> nodes = subGrid.iterator();

            // Try to assign every word to a different node 
            // for execution.
            for (String word : words) {
                // Execute word on remote node.
                refs.add(new GridJobReferenceAdapter(jobCtx.getJobTypeId(), new GridifyArgumentAdapter(gridifyArg,
                    word), nodes.next()));
                
                // If we run out of available nodes, then recycle the 
                // iterator to reuse nodes.
                if (nodes.hasNext() == false) {
                    nodes = subGrid.iterator();
                }
            }

            // Trigger remote execution.
            return refs;
        }

        // Trigger local execution.
        return null;
    }

    /**
     * <tt>HelloWorld</tt> job has no return value.
     * 
     * @see GridJob#aggregate(List)
     */
    public Serializable aggregate(List<GridJobRemoteResult> results) {
        // Nothing to aggregate or return.
        return null;
    }
}