package org.gridgain.examples.resources;

import org.gridgain.grid.*;
import org.gridgain.grid.logger.*;
import org.gridgain.grid.resources.*;
import java.util.*;
import java.io.*;

public class GridResourcesTask extends GridTaskSplitAdapter<String, Integer> {
    /** Grid logger. */
    @GridLoggerResource
    private GridLogger log = null;

    /**
     * Splits the passed in phrase into words and creates a job for every
     * word. Every job will print out the word and return number of letters in that
     * word. Job use context to store data through injected context.
     *
     * @param gridSize Number of nodes in the grid.
     * @param phrase Any phrase.
     * @return Created grid jobs for remote execution.
     * @throws GridException If split failed.
     */
    @Override
    public Collection<? extends GridJob> split(int gridSize, String phrase) throws GridException {
        // Split the passed in phrase into multiple words separated by spaces.
        String[] words = phrase.split(" ");

        List<GridJob> jobs = new ArrayList<GridJob>(words.length);

        for (String word : words) {
            // Every job gets its own word as an argument.
            jobs.add(new GridJobAdapter<String>(word) {
                /** */
                @SuppressWarnings("hiding")
                @GridUserResource
                private transient GridResourcesContext ctx = null;

                /*
                 * Simply prints the word passed into the job and
                 * returns number of letters in that word.
                 */
                public Serializable execute() {
                    String word = getArgument();

                    if (log.isInfoEnabled() == true) {
                        log.info(">>>");
                        log.info(">>> Printing '" + word + "' on this node from grid job.");
                        log.info(">>>");
                    }

                    try {
                        ctx.sendData(word);
                    }
                    catch (Exception e) {
                        log.error("Failed to send data: " + word, e);
                    }

                    // Return number of letters in the word.
                    return word.length();
                }
            });
        }

        return jobs;
    }

    /**
     * Sums up all characters returns from all jobs and returns a
     * total number of characters in the phrase.
     *
     * @param results Job results.
     * @return Number of characters for the phrase passed into
     *      <tt>split(gridSize, phrase)</tt> method above.
     * @throws GridException If reduce failed.
     */
    public Integer reduce(List<GridJobResult> results) throws GridException {
        int totalCharCnt = 0;

        for (GridJobResult res : results) {
            // Every job returned a number of letters
            // for the word it was responsible for.
            Integer charCnt = res.getData();

            totalCharCnt += charCnt;
        }

        // Account for spaces. For simplicity we assume one space between words.
        totalCharCnt += results.size() - 1;

        // Total number of characters in the phrase
        // passed into task execution.
        return totalCharCnt;
    }
}