<?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-proxy id="space" name="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();
    }
}