Set itemReaders dynamically?

[joseps]

Is it possible to set itemReaders dynamically? Something like below is what I am looking to do.

<job id="myJob" restartable="true">
<step id="step1" >
<tasklet task-executor="taskExecutor">
<chunk reader="#{jobParameters[customReader]}"
processor="#{jobParameters[customProcessor]}"
writer="#{jobParameters[customWriter]}"
commit-interval="1000"/>
<listeners>
<listener ref=" myJobStepListener" />
</listeners>
</tasklet>
</step>
<listeners>
<listener ref="myJobListener"/>
</listeners>
</job>

As shown above, I would like to ‘wire-in’ readers, processors and writers for a step, by passing spring bean references as job parameters. We require this (or we think we require this!) as we deal with different types of file formats (multiline, nested and other custom formats over which we have no control) received on different (Mule) file inbound endpoints, which are dynamically setup with configuration details coming from a database. We will need to keep adding these custom readers on an on-going basis, and we are essentially trying to minimize the overall code changes.

If the above is not possible, what other alternatives are there? I came close to an alternative, where a reader is ‘wired-in’ with one level of abstraction (or via a wrapper), but it gave me a ‘ReaderNotOpenException’. I didn’t want to go down that path where I will have to explicitly code open, close etc. on the reader.

Any help is appreciated. Thank you.

[joseps]

Never mind, I figured out a way to resolve this via spring factory-bean/factory-method. Thanks.

[sansun2111]

hey hi
just can you guide me how did you achieve it ,
can you paste a sample code !!
i am very much in need of that functionality .

[joseps]

In our implementation, Job is described as:

<job id="myJob" restartable="true">
	<step id="step1">
		<tasklet task-executor="taskExecutor">
			<chunk	reader="readerFactory" 
				processor="processorFactory"
				writer="writerFactory"
				commit-interval="1000"/>
			<listeners>
				<listener ref="stepListener" />
			</listeners>
		</tasklet>
	</step>
	<listeners>
		<listener ref="jobListener"/>
	</listeners>
</job>

and the 'readerFactory' above is described as:

<bean id="ReaderFactory" class="com.mycompany.batch.ReaderFactory" scope="step">
	<!--	
		pass in parameters required to construct and initialize the 
		required reader. In my code I passed in the id of a spring 
		bean defined elsewhere, and parameters required to initialize 
		that bean.
	-->
	<property name="reader" value="#{jobParameters[input.itemReader]}"/>
	<property name="resource" value="#{jobParameters[input.file.name]}"/>
	<property name="format" value="#{jobParameters[job.input.format]}" />	
</bean>

and 'ReaderFactory' is a Spring FactoryBean class, defined as:

public class ReaderFactory implements ApplicationContextAware, FactoryBean {
	protected ApplicationContext applicationContext;
	protected String reader;
	protected String resource;
	protected Format format; //provides format (fixedWidth, delimited etc.,
			//and other detailed mapping information such as
			//column names and ranges, also if header/footer is 
			//present, wether multi-line etc.

	public Class getObjectType() {
		return AbstractItemCountingItemStreamItemReader.class; 
	}

	public boolean isSingleton() {
		return false;
	}

	public Object getObject() {
		// Construct, initialize and return the reader
	}
}

In our implementation, 'reader' above is the id of a spring bean defined elsewhere. Using 'applicationContext', the reader bean is obtained and is initialized using the parameters provided.

I hope this is helpful. Thanks.

[sansun2111]

Thanks a ton mate ,
I was running short of ideas will implement it and then will let you know whether I was successfull in achivieng it.

[sansun2111]

<bean id="EDIJob" parent="simpleJob" >
<property name="steps" >
<bean id="step1" parent="simpleStep" >
<property name="streams" >
<list>

<ref bean="i9FileItemReader" />
<ref bean="ansiFileWriter" />
<ref bean="neutralizeWriter" />
<ref bean="reportsWriter" />
</list>

</property>

<property name="itemReader" ref="itemReader" />
<property name="itemWriter" ref="itemWriter" />


</bean>
</property>
</bean>



<bean id="itemReader" class="com.abcbs.EDI837.reader.EDI837ItemReader" >

<property name="itemReader" ref="someOtherItemReader" />
<property name="resource" ref="inputResource" />
<property name="i9FileReader" ref="i9FileItemReader" />
<property name="additionalSegmentFlag" value="#{jobParameters['addSeg']}"></property>
<property name="replaceHISegment" value="jobParameters['hiSeg']}"></property>

</bean>



There so many other readers !!!
So I need to read the file on the basis of I/P file Format ..


I am not getting any idea how to intialize my own reader in the ReaderFactory class you have mentioned and apart form that how am i supposed to instatiate the application Context .

Sorry for the trouble,i am newbee to the spring batch... so Detailed explanation would surely help me .

Thanks
Sandeep

[joseps]

Below is how you can use the ApplicationContext:

public class ReaderFactory implements ApplicationContextAware, FactoryBean {
	protected ApplicationContext applicationContext;
	protected String reader;
	protected String resource;
	protected Format format; //provides format (fixedWidth, delimited etc.,
			//and other detailed mapping information such as
			//column names and ranges, also if header/footer is 
			//present, wether multi-line etc.

	public Class getObjectType() {
		return AbstractItemCountingItemStreamItemReader.class; 
	}

	public boolean isSingleton() {
		return false;
	}

	public Object getObject() {
		if (reader != null) {
			Object reader = applicationContext.getBean(reader);
			// Now, you can determine the type of reader and 
			// initialize it with whatever, as needed
			return reader;
		}
		else return null;	
	}
}

In the above code, you need 'setters' defined for all properties for the 'wiring' to take place.

ApplicationContext is fundamental to the the Spring Framework. Learning the framework is essential in what you are trying to do.

[sansun2111]

Sorry Joseps

Reagrding applicationContext ...
I guess i was outta of mind when i typed that , i myself dint realize it :|
What i was typing then .

<bean id="EDIJob" parent="simpleJob" >
		<property name="steps"  >
			<bean id="step1" parent="simpleStep" >
				<property name="streams" >
					<list>
					
						<ref bean="i9FileItemReader" />
    					<ref bean="ansiFileWriter" />
						<ref bean="neutralizeWriter" />
						<ref bean="reportsWriter" />						
			        	<ref bean="multiFileItemReader" />
					</list>

				</property>
				
				<property name="itemReader" ref="reader" />
				<property name="itemWriter" ref="itemWriter" />
				
				
			</bean>
		</property>
	</bean>





 <bean id="reader" class="com.abcbs.EDI837.common.ReaderFactory"  >
	
	<property name="reader" value="#{jobParameters['reader.name']}"/>		
	</bean>

ReaderFactory class

public class ReaderFactory implements ApplicationContextAware, FactoryBean  {

	protected ApplicationContext applicationContext;
	protected String reader;

	public void setReader( String reader)
	{
		this.reader = reader;
	}

	public Class getObjectType() {
		return AbstractItemCountingItemStreamItemReader.class; 
	}

	public boolean isSingleton() {
		return false;
	}

	public Object getObject() {

		
		if (reader != null) {
			System.out.println("READER"+reader);
			Object reader1 = applicationContext.getBean(reader);
			
			return reader1;
		}
		else return null;	
	}

	public void setApplicationContext(ApplicationContext applicationContext)
			throws BeansException {
		this.applicationContext = applicationContext;
		
	}

}

I am getting this Exception
[code]
Exception

org.springframework.beans.factory.NoSuchBeanDefini tionException: No bean named '#{jobParameters['reader.name']}' is defined
at org.springframework.beans.factory.support.DefaultL istableBeanFactory.getBeanDefinition(DefaultListab leBeanFactory.java:387)
at org.springframework.beans.factory.support.Abstract BeanFactory.getMergedLocalBeanDefinition(AbstractB eanFactory.java:968)
at org.springframework.beans.factory.support.Abstract BeanFactory.doGetBean(AbstractBeanFactory.java:246 )
at org.springframework.beans.factory.support.Abstract BeanFactory.getBean(AbstractBeanFactory.java:185)
at org.springframework.beans.factory.support.Abstract BeanFactory.getBean(AbstractBeanFactory.java:164)
at org.springframework.context.support.AbstractApplic ationContext.getBean(AbstractApplicationContext.ja va:881)
at com.abcbs.EDI837.common.ReaderFactory.getObject(Re aderFactory.java:39)
at org.springframework.beans.factory.support.FactoryB eanRegistrySupport$1.run(FactoryBeanRegistrySuppor t.java:121)
... 35 more


[code]


If i am not wrong this should be converted into the Value which i mentioned in the JobParameters .... and the value is itemReader ...


but the value is not being replaced i guess ...

[joseps]

scope="step" must be set for receiving jobParameters, like below:

<bean id="reader" class="com.abcbs.EDI837.common.ReaderFactory"  scope="step>
.
.
.
</bean>

[sansun2111]

In setDelegate
Hibernate: select icdftpdtl0_.FTP_ID_NBR as FTP1_6_, icdftpdtl0_.FTP_TRANSFER_TYPE as FTP2_6_, icdftpdtl0_.CO_ID_CD as CO3_6_, icdftpdtl0_.ADDRESS_TYPE_CD as ADDRESS4_6_, icdftpdtl0_.IP_URL_ADDRESS as IP5_6_, icdftpdtl0_.LOGON_ID as LOGON6_6_, icdftpdtl0_.PASSWORD_ID as PASSWORD7_6_, icdftpdtl0_.FTP_FILES_PATH as FTP8_6_, icdftpdtl0_.PROCESS_FILES_PATH as PROCESS9_6_, icdftpdtl0_.FTP_TYPE_CD as FTP10_6_, icdftpdtl0_.PROCESS_FILES as PROCESS11_6_, icdftpdtl0_.TRNS_FREQUENCY as TRNS12_6_ from ICD10_web.ICD_FTP_DTL icdftpdtl0_, ICD10_web.ICD_NEUTRALIZER_DTL icdneutral1_ where icdftpdtl0_.FTP_ID_NBR=icdneutral1_.FTP_ID_NBR and trim(upper(icdneutral1_.NEUTRALIZATION_STATUS))='ACTIVE' order by icdftpdtl0_.FTP_ID_NBR
Hibernate: select icdneutral0_.FTP_ID_NBR as FTP1_5_0_, icdneutral0_.CO_ID_CD as CO2_5_0_, icdneutral0_.TRANS_TYPE_CD as TRANS3_5_0_, icdneutral0_.DATA_FORMAT_CD as DATA4_5_0_, icdneutral0_.ADDITIONAL_SEGMENT as ADDITIONAL5_5_0_, icdneutral0_.ADDITIONAL_FILE as ADDITIONAL6_5_0_, icdneutral0_.ADDITIONAL_FILE_TYPE as ADDITIONAL7_5_0_, icdneutral0_.REPLACE_HI_SEG as REPLACE8_5_0_, icdneutral0_.NEUTRALIZATION_STATUS as NEUTRALI9_5_0_, icdneutral0_.START_DT as START10_5_0_ from ICD10_web.ICD_NEUTRALIZER_DTL icdneutral0_ where icdneutral0_.FTP_ID_NBR=?
READER {jobParameters['reader']}
ERROR - Encountered an error executing the step
org.springframework.aop.AopInvocationException: AOP configuration seems to be invalid: tried calling method [public abstract void org.springframework.batch.item.ItemStream.open(org.springframework.batch.item.ExecutionContext) throws org.springframework.batch.item.ItemStreamException] on target [com.abcbs.EDI837.reader.EDI837ItemReader@56fc16]; nested exception is java.lang.IllegalArgumentException: object is not an instance of declaring class
	at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:318)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:182)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:149)
	at org.springframework.aop.support.DelegatingIntroductionInterceptor.doProceed(DelegatingIntroductionInterceptor.java:131)
	at org.springframework.aop.support.DelegatingIntroductionInterceptor.invoke(DelegatingIntroductionInterceptor.java:119)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:171)
	at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:204)
	at $Proxy8.open(Unknown Source)
	at org.springframework.batch.item.support.CompositeItemStream.open(CompositeItemStream.java:98)
	at org.springframework.batch.core.step.tasklet.TaskletStep.open(TaskletStep.java:379)
	at org.springframework.batch.core.step.AbstractStep.execute(AbstractStep.java:195)
	at org.springframework.batch.core.job.AbstractJob.handleStep(AbstractJob.java:348)
	at org.springframework.batch.core.job.SimpleJob.doExecute(SimpleJob.java:121)
	at org.springframework.batch.core.job.AbstractJob.execute(AbstractJob.java:250)
	at com.abcbs.EDI837.common.MySimpleJobLauncher$1.run(MySimpleJobLauncher.java:96)
	at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:49)
	at com.abcbs.EDI837.common.MySimpleJobLauncher.run(MySimpleJobLauncher.java:91)
	at com.abcbs.EDI837.common.MyCommandLine.start(MyCommandLine.java:343)
	at com.abcbs.EDI837.common.MyCommandLine.main(MyCommandLine.java:402)
Caused by: java.lang.IllegalArgumentException: object is not an instance of declaring class
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
	at java.lang.reflect.Method.invoke(Method.java:597)
	at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:310)
	... 18 more
ERROR - Exception while closing step execution resources
org.springframework.aop.AopInvocationException: AOP configuration seems to be invalid: tried calling method [public abstract void org.springframework.batch.item.ItemStream.close() throws org.springframework.batch.item.ItemStreamException] on target [com.abcbs.EDI837.reader.EDI837ItemReader@56fc16]; nested exception is java.lang.IllegalArgumentException: object is not an instance of declaring class
	at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:318)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:182)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:149)
	at org.springframework.aop.support.DelegatingIntroductionInterceptor.doProceed(DelegatingIntroductionInterceptor.java:131)
	at org.springframework.aop.support.DelegatingIntroductionInterceptor.invoke(DelegatingIntroductionInterceptor.java:119)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:171)
	at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:204)
	at $Proxy8.close(Unknown Source)
	at org.springframework.batch.item.support.CompositeItemStream.close(CompositeItemStream.java:86)
	at org.springframework.batch.core.step.tasklet.TaskletStep.close(TaskletStep.java:375)
	at org.springframework.batch.core.step.AbstractStep.execute(AbstractStep.java:258)
	at org.springframework.batch.core.job.AbstractJob.handleStep(AbstractJob.java:348)
	at org.springframework.batch.core.job.SimpleJob.doExecute(SimpleJob.java:121)
	at org.springframework.batch.core.job.AbstractJob.execute(AbstractJob.java:250)
	at com.abcbs.EDI837.common.MySimpleJobLauncher$1.run(MySimpleJobLauncher.java:96)
	at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:49)
	at com.abcbs.EDI837.common.MySimpleJobLauncher.run(MySimpleJobLauncher.java:91)
	at com.abcbs.EDI837.common.MyCommandLine.start(MyCommandLine.java:343)
	at com.abcbs.EDI837.common.MyCommandLine.main(MyCommandLine.java:402)
Caused by: java.lang.IllegalArgumentException: object is not an instance of declaring class
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
	at java.lang.reflect.Method.invoke(Method.java:597)
	at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:310)
	... 18 more
1154 [main] INFO org.hibernate.impl.SessionFactoryImpl - building session factory
1466 [main] INFO org.hibernate.impl.SessionFactoryObjectFactory - Not binding factory to JNDI, no JNDI name configured
2636 [main] INFO org.hibernate.impl.SessionFactoryImpl - closing