<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xmlns:os-core="http://www.openspaces.org/schema/core"
      xmlns:os-events="http://www.openspaces.org/schema/events"
      xmlns:os-remoting="http://www.openspaces.org/schema/remoting"
	
	xsi:schemaLocation="http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch-2.1.xsd
	      http://www.openspaces.org/schema/core http://www.openspaces.org/schema/core/openspaces-core.xsd
      http://www.openspaces.org/schema/events http://www.openspaces.org/schema/events/openspaces-events.xsd
      http://www.openspaces.org/schema/remoting http://www.openspaces.org/schema/remoting/openspaces-remoting.xsd	
		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd"
	xmlns:p="http://www.springframework.org/schema/p">
		
	<bean id="itemReader" class="com.gigaspaces.springbatch.SpaceItemReader" >
		<property name="blocking" value="false"/>
		<property name="dedicatedWorker" value="true"/>
	</bean>
	
	<bean id="itemWriter" class="com.gigaspaces.springbatch.SpaceItemWriter" />
	<bean id="itemProcessor" class="com.gigaspaces.springbatch.SpaceItemProcessor" />

    <job id="spaceJob" xmlns="http://www.springframework.org/schema/batch" restartable="false" >
        <step id="step1" >
            <tasklet>
                <chunk reader="itemReader" writer="itemWriter" processor="itemProcessor" commit-interval="2" />
            </tasklet>            
        </step>
    </job>

	<bean id="jobRepository" class="org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean">
		<property name="transactionManager" ref="transactionManager"/>
	</bean>

	<bean id="jobLauncher" class="org.springframework.batch.core.launch.support.SimpleJobLauncher">
		<property name="jobRepository" ref="jobRepository" />
	</bean>

	<bean id="jobRunner" class="com.gigaspaces.springbatch.JobRunner">
	</bean>

	<bean id="jobRegistry" class="org.springframework.batch.core.configuration.support.MapJobRegistry" />
 	
	<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager" lazy-init="true">
		<property name="dataSource" ref="dataSource" />
	</bean>

    <bean id="dataSource" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close" >
        <property name="driverClassName" value="org.hsqldb.jdbcDriver"/>
        <property name="url" value="jdbc:hsqldb:hsql://localhost/xdb"/>		
        <property name="username" value="sa"/>
        <property name="password" value=""/>
    </bean>

	<os-core:space id="space" url="jini://*/*/mySpace"/>
	<os-core:giga-space id="gigaSpace" space="space" />
	<os-core:giga-space-context/>
</beans>

public class ItemBase {
	String id=null;
	Object payload;
	
	Integer jobID;
	Integer routing;

	@SpaceId (autoGenerate = false)
	public String getId() {
		return id;
	}
	public void setId(String id) {
		this.id = id;
	}
	
	@Override
	public String toString() {
		return " " +id;
	}
	public Object getPayload() {
		return payload;
	}
	public void setPayload(Object payload) {
		this.payload = payload;
	}
	public Integer getJobID() {
		return jobID;
	}
	public void setJobID(Integer jobID) {
		this.jobID = jobID;
	}
	
	@SpaceRouting
	public Integer getRouting() {
		return routing;
	}
	public void setRouting(Integer routing) {
		this.routing = routing;
	}
}

@SpaceClass(fifoSupport=FifoSupport.ALL)
public class ItemRequest extends ItemBase{
	public ItemRequest ()
	{
	}
	
	void execute(){};	
}

public class ItemResult  extends ItemBase{
	public ItemResult(){}	
}

public class SpaceItemReader<T> implements ItemReader<T>, InitializingBean  {

	@GigaSpaceContext
	GigaSpace space;
	
	@ClusterInfoContext
	ClusterInfo clusterInfo;
	int instanceID = 0;

	boolean blocking = true;
	boolean dedicatedWorker= true;
	
	@Override
	public void afterPropertiesSet() throws Exception {
		if (clusterInfo!=null) instanceID=clusterInfo.getInstanceId();
		System.out.println("SpaceItemReader been created - Instance ID:"+instanceID + " Blocking mode:" + blocking);
	}

	@Override
	public T read() throws Exception, UnexpectedInputException, ParseException,
			NonTransientResourceException {
		System.out.println("SpaceItemReader called");
		ItemRequest request=null;
		ItemRequest requestTemplate=new ItemRequest();

		if (dedicatedWorker)
		{
			if (clusterInfo == null)
			{
				System.out.println("Running a single Item Reader");
			}
			else
			{
				String total_members = space.getSpace().getURL().getProperty(SpaceURL.CLUSTER_TOTAL_MEMBERS);
				int partitionCount=1;
				if (total_members != null)
					partitionCount = Integer.valueOf(total_members .substring(0,total_members.indexOf(","))).intValue();

		        int routingValue = (clusterInfo.getInstanceId() - 1) % partitionCount;
				requestTemplate.setRouting(routingValue);
				System.out.println("-> "+ space.getSpace().getName() + 
					" Space got " + partitionCount+ " partitions - Worker Routing value:"+routingValue);
			}
		}
		
		while (true)
		{
			if (blocking)
				request =space.take(requestTemplate, 10000); 
			else
			{
				request=space.take(requestTemplate, 0);
				if (request==null)
				{
					Thread.sleep(1000);
				}
			}
			if (request !=null)
				break;
		}
		return (T)request; 
	}

	public boolean isBlocking() {
		return blocking;
	}

	public void setBlocking(boolean blocking) {
		this.blocking = blocking;
	}

	public boolean isDedicatedWorker() {
		return dedicatedWorker;
	}

	public void setDedicatedWorker(boolean dedicatedWorker) {
		this.dedicatedWorker = dedicatedWorker;
	}
}

public class SpaceItemProcessor<I,O>  implements ItemProcessor<I,O>{

	@GigaSpaceContext
	GigaSpace space;

	@Override
	public O process(I request) throws Exception {
		System.out.println("SpaceItemProcessor called with Request ID:" + request);
		Thread.sleep(1000);
		ItemRequest itemRequest = (ItemRequest)request;
		ItemResult result = new ItemResult();
		result.setId(itemRequest.getId());		
		result.setJobID(itemRequest.getJobID());
		result.setRouting(itemRequest.getRouting());
		space.write(result); 
		return (O)result;
	}
}

public class SpaceItemWriter<T> implements ItemWriter<T> , InitializingBean  {

	@GigaSpaceContext
	GigaSpace space;

	@Override
	public void afterPropertiesSet() throws Exception {
		System.out.println("SpaceItemWriter been created");
	}

	@Override
	public void write(List<? extends T> items) throws Exception {
		System.out.println("SpaceItemWriter called with "+  items.size()+" Space Have " + space.count(new ItemResult()) + " results");
	}
}

public class JobRunner implements ApplicationContextAware,InitializingBean ,DisposableBean{

	FlowJob job ;
	SimpleJobLauncher launcher;
	Task launcherTask;
	
	@Override
	public void setApplicationContext(ApplicationContext appContent)
			throws BeansException {
		job = (FlowJob)appContent.getBean("spaceJob");
		launcher = (SimpleJobLauncher)appContent.getBean("jobLauncher");
	}

	@Override
	public void afterPropertiesSet() throws Exception {

		ExecutorService threadExecutor = Executors.newFixedThreadPool(1);
		launcherTask = new Task (launcher,job);
		threadExecutor.execute(launcherTask); 
	}

	static class Task implements Runnable
	{
		SimpleJobLauncher launcher;
		FlowJob job ;
		public JobExecution jobExecution;
		
		public Task(SimpleJobLauncher _launcher, FlowJob _job)
		{
			this.launcher=_launcher;
			this.job=_job;
		}
		
		public void run() {
			org.springframework.batch.core.JobParameters jobParameters = new JobParameters(); 
			try {
				jobExecution = launcher.run(job, jobParameters);
			} catch (Exception e) {
				e.printStackTrace();
			} 		
			
		}
		
	}

	public void destroy() throws Exception {
		launcherTask.jobExecution.stop();
	}
}