package com.github.dexecutor.ignite;

import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCompute;
import org.apache.ignite.Ignition;
import org.apache.ignite.configuration.IgniteConfiguration;

import com.github.dexecutor.core.DefaultDependentTasksExecutor;
import com.github.dexecutor.core.DependentTasksExecutorConfig;
import com.github.dexecutor.core.ExecutionConfig;
import com.github.dexecutor.core.task.Task;
import com.github.dexecutor.core.task.TaskProvider;

public class Job {

	public void run(boolean isMaster, final String nodeName) throws Exception {

		IgniteConfiguration cfg = new IgniteConfiguration();
		cfg.setGridName(nodeName); 

		Ignite ignite = Ignition.start(cfg); 

		if (isMaster) {
			DefaultDependentTasksExecutor<Integer, Integer> dexecutor = newTaskExecutor(ignite.compute());

			buildGraph(dexecutor);
			dexecutor.execute(ExecutionConfig.TERMINATING);
		}

		System.out.println("Ctrl+D/Ctrl+Z to stop.");
	}
	
	private DefaultDependentTasksExecutor<Integer, Integer> newTaskExecutor(final IgniteCompute igniteCompute) {
		DependentTasksExecutorConfig<Integer, Integer> config = new DependentTasksExecutorConfig<Integer, Integer>(
				new IgniteExecutionEngine<Integer, Integer>(igniteCompute), new SleepyTaskProvider());
		return new DefaultDependentTasksExecutor<Integer, Integer>(config);
	}

	private void buildGraph(final DefaultDependentTasksExecutor<Integer, Integer> dexecutor) {
		dexecutor.addDependency(1, 2);
		dexecutor.addDependency(1, 2);
		dexecutor.addDependency(1, 3);
		dexecutor.addDependency(3, 4);
		dexecutor.addDependency(3, 5);
		dexecutor.addDependency(3, 6);
		dexecutor.addDependency(2, 7);
		dexecutor.addDependency(2, 9);
		dexecutor.addDependency(2, 8);
		dexecutor.addDependency(9, 10);
		dexecutor.addDependency(12, 13);
		dexecutor.addDependency(13, 4);
		dexecutor.addDependency(13, 14);
		dexecutor.addIndependent(11);
	}
	
	private static class SleepyTaskProvider implements TaskProvider<Integer, Integer> {

		public Task<Integer, Integer> provideTask(final Integer id) {

			return new Task<Integer, Integer>() {

				private static final long serialVersionUID = 1L;

				public Integer execute() {
					try {
						System.out.println("Executing :*****  " +  getId());
						Thread.sleep(time(0, 5000));
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					return id;
				}

				private long time(int min, int max) {
					return min + (int)(Math.random() * ((max - min) + 1));
				}
			};
		}
	}
}

@@ -29,24 +29,65 @@
 .ignite;%0D%0A%0D%0A
+import java.util.concurrent.TimeUnit;%0D%0A%0D%0A
 import org.a
@@ -139,23 +139,26 @@
 e.Ignite
-Compute
+AtomicLong
 ;%0D%0Aimpor
@@ -372,32 +372,76 @@
 xecutorConfig;%0D%0A
+import com.github.dexecutor.core.Duration;%0D%0A
 import com.githu
@@ -917,18 +917,8 @@
 nite
-.compute()
 );%0D%0A
@@ -967,16 +967,20 @@
 execute(
+new 
 Executio
@@ -990,20 +990,66 @@
 nfig
-.TERMINATING
+().scheduledRetrying(4, new Duration(1, TimeUnit.SECONDS))
 );%0D%0A
@@ -1200,27 +1200,13 @@
 nite
-Compute igniteCompu
+ igni
 te) 
@@ -1377,15 +1377,18 @@
 nite
-C
+.c
 ompute
+()
 ), n
@@ -1409,16 +1409,22 @@
 rovider(
+ignite
 ));%0D%0A%09%09r
@@ -2163,16 +2163,157 @@
 eger%3E %7B%0D
+%0A%09%09final IgniteAtomicLong count;%0D%0A%0D%0A%09%09public SleepyTaskProvider(final Ignite ignite) %7B%0D%0A%09%09%09count = ignite.atomicLong(%22count%22, 0, true);%0D%0A%09%09%7D%0D
 %0A%0D%0A%09%09pub
@@ -2514,24 +2514,258 @@
 %09%09%09%09%09try %7B%0D%0A
+%09%09%09%09%09%09if (id == 2) %7B%0D%0A%09%09%09%09%09%09%09count.incrementAndGet();%0D%0A%09%09%09%09%09%09%09System.out.println(%22Count is %22 + count.get());%0D%0A%09%09%09%09%09%09%09if (count.get() %3C 3) %7B%09%09%09%09%09%09%09%0D%0A%09%09%09%09%09%09%09%09throw new IllegalArgumentException(%22Invalid task%22);%0D%0A%09%09%09%09%09%09%09%7D%09%09%09%09%09%09%0D%0A%09%09%09%09%09%09%7D%0D%0A
 %09%09%09%09%09%09System

Test for retry support