public class PrimeNumberJob implements Job {
    private static final long serialVersionUID = 1L;
    private Identity id;
    private JobDescriptor jobDescriptor;
    private int primeCandidate = -1;
    private int numOfTasks;
    /**
     * Create a PrimNumberJob
     */
    public PrimeNumberJob() {
        id = Identity.newInstance();
    }
    /**
     * @see com.gigaspaces.grid.master.Job#setArguments(Object[])
     */
    public void setArguments(Object[] args) {
        if(args != null && args.length > 0) {
            if(args[0] instanceof String)
                primeCandidate = Integer.valueOf((String)args[0]).intValue();
            else
                throw new IllegalArgumentException("The prime candidate must "+
                                                   "be a String");
        } else {
            throw new IllegalArgumentException("The prime candidate is not " +
                                               "specified");
        }
    }
    /**
     * @see com.gigaspaces.grid.master.Job#split()
     */
    public Task[] split() {
        numOfTasks = getTotalTasks();
        if(numOfTasks==0)
            return(new Task[0]);
        int numbersPerTask = (primeCandidate - 2) / numOfTasks;
        System.out.println("Creating "
                           + numOfTasks
                           + " tasks, "
                           + numbersPerTask
                           + " numbers per task.");
        Task[] result = new Task[numOfTasks];
        for(int i = 0; i < numOfTasks; i++) {
            int startDivisor = i * numbersPerTask + 2;
            int endDivisor = i * numbersPerTask + 2 + numbersPerTask - 1;
            result[i] = new PrimeTask(getIdentity(),
                                             primeCandidate,
                                             startDivisor,
                                             endDivisor);
        }
        return result;
    }
    /**
     * @see com.gigaspaces.grid.master.Job#getTotalTasks()
     */
    public int getTotalTasks() {
        if(primeCandidate == -1)
            throw new IllegalStateException("The job arguments have not been "+
                                            "specified");
        return (int)Math.sqrt((double)primeCandidate - 2);
    }
    /**
     * @see com.gigaspaces.grid.master.Job#getJobDescriptor()
     */
    public JobDescriptor getJobDescriptor() {
        if(jobDescriptor == null) {
            jobDescriptor = new JobDescriptor();
            jobDescriptor.setTitle("Prime Number Cruncher");
            jobDescriptor.setDescription("This job computes if a given integer "+
                                          "is a prime number by dividing the "+
                                          "check process into sub tasks "+
                                          "and utilizing a JavaSpace to coordinate "+
                                          "threads that perform the sub tasks.");
            jobDescriptor.setResultDescription("Result is TRUE if a given integer "
                                                + "is a prime number, otherwise "+
                                                "FALSE");
            jobDescriptor.setArgumentsDescription("The candidate integer number");
            jobDescriptor.setVendor("GigaSpaces");
            jobDescriptor.setVersion("0.1");
        }
        return(jobDescriptor);
    }    
    /**
     * @see com.gigaspaces.grid.master.Job#getIdentity()
     */
    public Identity getIdentity() {
        return(id);
    }
    /**
     * @throws JobException
     * @see com.gigaspaces.grid.master.Job#process(JavaSpace, Transaction, boolean)
     */
    public JobResult process(JavaSpace space,
                             Transaction txn,
                             boolean batch) throws JobException {
        if(space==null)
            throw new NullPointerException("space is null");
        long start = System.currentTimeMillis();
        if(primeCandidate <= 1) {
            long now = System.currentTimeMillis();
            return(new JobResult(new Boolean(false), (now - start)));
        }
        if(primeCandidate == 2) {
            long now = System.currentTimeMillis();
            return(new JobResult(new Boolean(true), (now - start)));
        }
        
        boolean result = true;
        long execTime = -1;
        int numOfResults = 0;
        PrimeResult divisorResult = null;
        PrimeResult template = new PrimeResult(getIdentity());
        System.out.println("Waiting for results, "+
                           "total tasks="+getTotalTasks()+", "+
                           "transaction="+(txn==null?"null":txn.toString()));
        while (numOfResults < getTotalTasks()) {
            try {
                PrimeResult taskResult = (PrimeResult)space.take(template,
                                                                 txn,
                                                                 Long.MAX_VALUE);                         
                numOfResults++;                                    
                if(taskResult.isDivisable()) {
                    divisorResult = taskResult;                
                    result = false;
                    execTime = (long)taskResult.getExecutionTime();
                }
            } catch(Exception e) {
                Throwable cause = (e.getCause()==null?e:e.getCause());
                throw new JobException("Taking PrimeResult entries", cause);
            }            
        }
        if(divisorResult!=null)
            System.out.println("["+numOfResults+"] results tallied, divisor "+
                               divisorResult.m_Divisor.intValue()+" "+
                               "found for "+ primeCandidate);
        else
            System.out.println("["+numOfResults+"] results tallied, no divisor "+
                               "found for "+ primeCandidate);
        if(txn != null) {
            try {
                txn.commit();
            } catch(Exception e) {
                e.printStackTrace();
            }
        }
        long now = System.currentTimeMillis();
        return(new JobResult(new Boolean(result),
                             (execTime==0?(now - start):(execTime))));
    }
}

public class PrimeTask extends Task {
    private static final long serialVersionUID = 1L;
    public Integer m_ValueToCheck;
    public Integer m_StartDivisor;
    public Integer m_EndDivisor;
    public PrimeTask() {
    }
    /**
     * Create a PrimeTask
     *
     * @param id A non-null {@link com.gigaspaces.grid.master.Identity}
     */
    public PrimeTask(Identity id) {
        super(id);
    }
    /**
     * Create a PrimeTask
     *
     * @param id
     * @param valueToCheck
     * @param startDivisor
     * @param endDivisor
     */
    public PrimeTask(Identity id,
                     int valueToCheck,
                     int startDivisor,
                     int endDivisor) {
        super(id);
        m_ValueToCheck = new Integer(valueToCheck);
        m_StartDivisor = new Integer(startDivisor);
        m_EndDivisor = new Integer(endDivisor);
    }
    /**
     * @see com.gigaspaces.grid.master.Task#execute()
     */
    public TaskResult execute(Map map)  {
        for(int i = m_StartDivisor.intValue();
            i <= m_EndDivisor.intValue();
            i++) {
            /* divisor found */
            if(m_ValueToCheck.intValue() % i == 0) {
                return new PrimeResult(getIdentity(),
                                       m_ValueToCheck.intValue(),
                                       m_StartDivisor.intValue(),
                                       m_EndDivisor.intValue(),
                                       i);
            }
        }
        /* divisor not found */
        return new PrimeResult(getIdentity(),
                               m_ValueToCheck.intValue(),
                               m_StartDivisor.intValue(),
                               m_EndDivisor.intValue(),
                               0);
    }
}

public class PrimeResult extends TaskResult {
    private static final long serialVersionUID = 1L;
    public Integer m_CheckedValue;
    public Integer m_StartDivisor;
    public Integer m_EndDivisor;
    public Integer m_Divisor;
    /**
     * Create a PrimeResult with no properties
     */
    public PrimeResult() {
    }
    /**
     * Create a PrimeResult
     *
     * @param id The Identity of the PrimeResult
     */
    public PrimeResult(Identity id) {
        super(id);
    }
    /**
     * Create a PrimeResult, setting all properties
     *
     * @param id The Identity of the PrimeResult
     * @param checkedValue
     * @param startDivisor
     * @param endDivisor
     * @param divisor
     */
    public PrimeResult(Identity id,
                       int checkedValue,
                       int startDivisor,
                       int endDivisor,
                       int divisor) {
        super(id);
        m_CheckedValue = new Integer(checkedValue);
        m_StartDivisor = new Integer(startDivisor);
        m_EndDivisor = new Integer(endDivisor);
        m_Divisor = new Integer(divisor);
    }
    /**
     * Get whether the result is prime
     *
     * @return Whether the result is prime
     */
    public boolean isDivisable() {
        return m_Divisor.intValue() != 0;
    }
}