CompositeItemReader

[dwakley]

Hi,

I'm trying to create a compositeItemReader, based on the suggestion in http://forum.springsource.org/showth...983#post191983 which will take the out from reader1 and use it as input for reader2, see the read() method. However when i run the class

package org.atoc.batch.dss.itemreader;

import org.atoc.batch.dss.vo.ExtractFiltersVO;
import org.atoc.batch.dss.vo.ExtractParamsVO;
import org.hibernate.SessionFactory;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.listener.StepExecutionListenerSupport;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.batch.item.database.HibernateCursorItemReader;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;

public class RetrieveExtractParamsExectionContextReader extends
		StepExecutionListenerSupport implements ItemReader, ItemStream,
		InitializingBean {

	private HibernateCursorItemReader<ExtractParamsVO> itemExtractParamsReader;
	private HibernateCursorItemReader<ExtractFiltersVO> itemExtractFilterReader;
	private SessionFactory sessionFactory;
	private String toc;

	public void beforeStep(StepExecution stepExecution) {
		itemExtractParamsReader = new HibernateCursorItemReader();
		itemExtractParamsReader
				.setQueryString("from ExtractParamsVO where tocID='" + toc
						+ "'");
		itemExtractParamsReader.setSessionFactory(sessionFactory);

		itemExtractFilterReader = new HibernateCursorItemReader();
		itemExtractFilterReader.setSessionFactory(sessionFactory);

		callDelegateAfterPropertiesSet(this.itemExtractParamsReader);
	}

	private void callDelegateAfterPropertiesSet(HibernateCursorItemReader d) {
		try {
			d.afterPropertiesSet();
		} catch (Exception e) {
			throw new RuntimeException("Error initializing delegate reader.", e);
		}
	}

	public void setSessionFactory(SessionFactory sessionFactory) {
		this.sessionFactory = sessionFactory;
	}

	public void afterPropertiesSet() throws Exception {
		Assert.notNull(sessionFactory,
				"Property [sessionFactory] is required and must be set.");
	}

	public ExtractFiltersVO read() throws Exception, UnexpectedInputException,
			ParseException {
		Assert.state(itemExtractParamsReader != null,"ItemReader not initialized.  Make sure to register as a step execution listener.");

		ExtractParamsVO epvo = itemExtractParamsReader.read();

		itemExtractFilterReader.setQueryName("from ExtractFiltersVO where tocID = '" + toc + "' and setName = '" + epvo.getSetName() + "'");
		callDelegateAfterPropertiesSet(this.itemExtractFilterReader);

		ExtractFiltersVO efvo = itemExtractFilterReader.read();
		System.out.println("extractfiltersvo " + efvo.toString());
		
		return efvo;
	}

	public void open(ExecutionContext executionContext)
			throws ItemStreamException {
		itemExtractParamsReader.open(executionContext);
	}

	public void update(ExecutionContext executionContext)
			throws ItemStreamException {
		itemExtractParamsReader.update(executionContext);
	}

	public void close() throws ItemStreamException {
		// TODO Auto-generated method stub
		itemExtractParamsReader.close();

	}

	public String getToc() {
		return toc;
	}

	public void setToc(String toc) {
		this.toc = toc;
	}

}

However I get NullPointerException thrown.

Could someone suggest what i'm doing wrong or a better alternative?

Thanks in advance
Dave

[Dave Syer]

You didn't say where the NullPointerException is thrown. Also, I might be wrong, but there seems to be a typo in your read() method since it discards the result of itemExtractFilterReader.read().

[dwakley]

Dave,

Thanks for replying back.

I have edited the code snippet in my original post to include the complete class definition and I have attached completel stacktrace.

thanks in advance.

[Dave Syer]

You didn't open the nested readers. If you implement ItemStream rather than StepExecutionListener that gives you an obvious place to do that.

[dwakley]

I'm already doing that for itemExtractParamsReader in the open() and update() methods at the bottom of the original code snippet. Since your suggestion I have now included the opening of the itemExtractFilterReader.open.

However the only way I can get it to run without any exception's being thrown is to set the queuestring of the itemExtractFilterReader to a temporary value in the beforeStep method and then in the read method reset the querystring with some additional criteria based on returned value from the first reader, but it retains the original/temporary querystring. To further illustrate this i have included the updated class.

package org.atoc.batch.dss.itemreader;

import org.atoc.batch.dss.vo.ExtractFiltersVO;
import org.atoc.batch.dss.vo.ExtractParamsVO;
import org.hibernate.SessionFactory;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.annotation.BeforeStep;
import org.springframework.batch.core.listener.StepExecutionListenerSupport;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.batch.item.database.HibernateCursorItemReader;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;

public class RetrieveExtractParamsExectionContextReader extends StepExecutionListenerSupport implements ItemReader, ItemStream,
		InitializingBean {
	private HibernateCursorItemReader<ExtractParamsVO> itemExtractParamsReader;
	private HibernateCursorItemReader<ExtractFiltersVO> itemExtractFilterReader;
	private SessionFactory sessionFactory;
	private String toc;


	public void beforeStep(StepExecution stepExecution) {
		
		itemExtractParamsReader = new HibernateCursorItemReader();
		itemExtractParamsReader
				.setQueryString("from ExtractParamsVO where tocID='" + toc
						+ "'");
		itemExtractParamsReader.setSessionFactory(sessionFactory);

		itemExtractFilterReader = new HibernateCursorItemReader();
		itemExtractFilterReader.setQueryString("from ExtractFiltersVO where tocID = '" + toc + "'");
		itemExtractFilterReader.setSessionFactory(sessionFactory);
		callDelegateAfterPropertiesSet(this.itemExtractParamsReader);
	}

	private void callDelegateAfterPropertiesSet(HibernateCursorItemReader d) {
		try {
			d.afterPropertiesSet();
		} catch (Exception e) {
			throw new RuntimeException("Error initializing delegate reader.", e);
		}
	}

	public void setSessionFactory(SessionFactory sessionFactory) {
		this.sessionFactory = sessionFactory;
	}

	public void afterPropertiesSet() throws Exception {
		Assert.notNull(sessionFactory,
				"Property [sessionFactory] is required and must be set.");
	}

	public ExtractFiltersVO read() throws Exception, UnexpectedInputException,
			ParseException {
		Assert.state(itemExtractParamsReader != null,"ItemReader not initialized.  Make sure to register as a step execution listener.");

		ExtractParamsVO epvo = itemExtractParamsReader.read();

		if (epvo != null){				
			itemExtractFilterReader.setQueryString("from ExtractFiltersVO where tocID = '" + toc + "' and setName = '" + epvo.getSetName() + "'");
			
			return itemExtractFilterReader.read();
		}
		else {
			return null;
		}
	}
	
	@Override
	public void open(ExecutionContext executionContext)
			throws ItemStreamException {
		itemExtractParamsReader.open(executionContext);
		itemExtractFilterReader.open(executionContext);
	
	}
	
	@Override
	public void update(ExecutionContext executionContext)
			throws ItemStreamException {
		itemExtractParamsReader.update(executionContext);
		itemExtractFilterReader.update(executionContext);
	}
	
	@Override
	public void close() throws ItemStreamException {
		// TODO Auto-generated method stub
		itemExtractParamsReader.close();
		itemExtractFilterReader.close();

	}

	public String getToc() {
		return toc;
	}

	public void setToc(String toc) {
		this.toc = toc;
	}



}

[Dave Syer]

I'm not really clear on what it is you are trying to do (why not just do the join in HQL?), but clearly if you change the query string you have to open the cursor again.

[dwakley]

Hi Dave,

Thank you very much for your assistance.

You are quite right as the above stands I could achieve the same thing using a join. However, since starting this thread I have moved onto a difference solution and have split the above into two different sets of reader, processor and writer tasks as I need to apply some business logic to the results of the first resultset before moving onto the next task. I just used the above as a simple example for writing a compositeItemReader.

Once again thanks for your time.
Dave