com.neurosys.etl.readers

Class ChainIterativeReader<TUPLE_IN,RECORD_OUT,STATE>

java.lang.Object

com.neurosys.etl.process.EventAwareProcess

com.neurosys.etl.readers.BaseIterativeProcessor<TUPLE_IN,RECORD_OUT,STATE>

com.neurosys.etl.readers.AbstractReader<TUPLE_IN,RECORD_OUT,STATE>

com.neurosys.etl.readers.ChainIterativeReader<TUPLE_IN,RECORD_OUT,STATE>

All Implemented Interfaces:

FailFastAware, Identifiable, ParamsAware, BatchProcess, ControllableProcess<STATE>, Process, RealTimeLineReader<RECORD_OUT>, ResourceAwareRealTimeLineReader<RECORD_OUT>, TransformAwareReader, TupleContextAwareReader<TUPLE_IN,RECORD_OUT>, Serializable, Iterator<RECORD_OUT>

public class ChainIterativeReader<TUPLE_IN,RECORD_OUT,STATE>
extends AbstractReader<TUPLE_IN,RECORD_OUT,STATE>

Chain multiple BaseIterativeProcessors

Author:

Arjun Dhar TODO: Add unit test case

See Also:

Serialized Form

Nested Class Summary

Nested classes/interfaces inherited from interface com.neurosys.etl.process.Process

Process.Event, Process.EventPoint

Field Summary

Fields inherited from class com.neurosys.etl.readers.BaseIterativeProcessor

exceptions, failFast, filter, hasNext, index, messages, params, paused, resource, transformer

Fields inherited from class com.neurosys.etl.process.EventAwareProcess

listeners, processState

Constructor Summary

Constructors

Constructor and Description
ChainIterativeReader() Default Constructor to allow lazy initialization If so, init() must be called by the calling API explicitly after calling BaseIterativeProcessor.setResource(Resource)
ChainIterativeReader(org.springframework.core.io.Resource resource)
ChainIterativeReader(org.springframework.core.io.Resource resource, List<BaseIterativeProcessor> processes)

Method Summary

Modifier and Type	Method and Description
void	dispose() Un-initialize reader, so that it maybe re consumed Except for the main Resource & configuration set by Spring context all other instance variables used will be re-set/disposed/cleared.
Long	getCurrentRawIndex(TUPLE_IN row) Get the raw index/position on the source; BaseIterativeProcessor.getCurrentIndex() only tracks the relevant position.
Collection<Throwable>	getExceptions() Used if failFast is off (false) to get the exceptions that occurred in the run/load
Collection<String>	getMessages() Messages generated during the process
Object[]	getParams() Additional Params for the ETL Process
List<BaseIterativeProcessor>	getProcesses()
STATE	getState() Get current state of process
boolean	hasNext() True if there is another record that can be read by the parser, false implies end of parse/read
void	init() Sets all BaseIterativeProcessor processes with a BaseIterativeProcessor.getResource() if not defined Additionally it checks if the process has been RealTimeLineReader.isInitialized().
boolean	isInitialized() Checks if all processes are RealTimeLineReader.isInitialized()
boolean	isPaused()
void	kill() Kill a process
protected TUPLE_IN	nextRecord() Locates the position of the next Record based on BaseIterativeProcessor.nextRecord() of the active process.
void	pause() Pause till resume
void	resume()
void	resumeFromState(STATE state) Resume process from the given State provided
void	setExceptions(Collection<Throwable> exceptions) Used if failFast is off (false) to get the exceptions that occurred in the run/load
void	setFailFast(boolean failFast) If true will exit on first exception else will collect exceptions
void	setMessages(Collection<String> messages) Messages generated during the process
void	setParams(Object[] params) Additional Params for the ETL Process
void	setProcesses(List<BaseIterativeProcessor> processes)
Long	size() Get number of tuples/records in resource
void	stop() Request graceful Shutdown

Methods inherited from class com.neurosys.etl.readers.BaseIterativeProcessor

close, filter, getCurrentIndex, getCurrentTupleIdentifier, getFilter, getId, getMeta, getResource, getSourceLocation, getTransformer, handleMiscException, isFailFast, next, nextTupleResult, remove, setFilter, setId, setMeta, setResource, setTransformer, transform

Methods inherited from class com.neurosys.etl.process.EventAwareProcess

addListener, getListeners, onChangeHandler, setListeners

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface java.util.Iterator

forEachRemaining

Constructor Detail

ChainIterativeReader

public ChainIterativeReader()

Default Constructor to allow lazy initialization
If so, init() must be called by the calling API explicitly after calling BaseIterativeProcessor.setResource(Resource)

ChainIterativeReader

public ChainIterativeReader(org.springframework.core.io.Resource resource)

ChainIterativeReader

public ChainIterativeReader(org.springframework.core.io.Resource resource,
                            List<BaseIterativeProcessor> processes)

Method Detail

init

@PostConstruct
public void init()
                         throws IOException,
                                Exception

Sets all BaseIterativeProcessor processes with a BaseIterativeProcessor.getResource() if not defined
Additionally it checks if the process has been RealTimeLineReader.isInitialized(). If NOT then it will call BaseIterativeProcessor.init()

Specified by:

init in interface RealTimeLineReader<RECORD_OUT>

Overrides:

init in class BaseIterativeProcessor<TUPLE_IN,RECORD_OUT,STATE>

Throws:

Exception

IOException

isInitialized

public boolean isInitialized()
                      throws Exception

Checks if all processes are RealTimeLineReader.isInitialized()

Throws:

Exception

dispose

public void dispose()
             throws BreakChainException

Description copied from interface: RealTimeLineReader

Un-initialize reader, so that it maybe re consumed
Except for the main Resource & configuration set by Spring context all other instance variables used will be re-set/disposed/cleared.

Throws:

BreakChainException

size

public Long size()

Description copied from interface: ResourceAwareRealTimeLineReader

Get number of tuples/records in resource

Returns:

Long

getState

public STATE getState()
               throws Exception

Description copied from interface: ControllableProcess

Get current state of process

Throws:

Exception

resumeFromState

public void resumeFromState(STATE state)
                     throws Exception

Description copied from interface: ControllableProcess

Resume process from the given State provided

Throws:

Exception

nextRecord

protected TUPLE_IN nextRecord()

Locates the position of the next Record based on BaseIterativeProcessor.nextRecord() of the active process. The active process is the first process that has its hasNext() is true.

Specified by:

nextRecord in class BaseIterativeProcessor<TUPLE_IN,RECORD_OUT,STATE>

setFailFast

public void setFailFast(boolean failFast)

If true will exit on first exception else will collect exceptions

Specified by:

setFailFast in interface FailFastAware

Overrides:

setFailFast in class BaseIterativeProcessor<TUPLE_IN,RECORD_OUT,STATE>

getExceptions

public Collection<Throwable> getExceptions()

Used if failFast is off (false) to get the exceptions that occurred in the run/load

Specified by:

getExceptions in interface FailFastAware

Overrides:

getExceptions in class BaseIterativeProcessor<TUPLE_IN,RECORD_OUT,STATE>

Returns:

setExceptions

public void setExceptions(Collection<Throwable> exceptions)

Used if failFast is off (false) to get the exceptions that occurred in the run/load

Specified by:

setExceptions in interface FailFastAware

Overrides:

setExceptions in class BaseIterativeProcessor<TUPLE_IN,RECORD_OUT,STATE>

getMessages

public Collection<String> getMessages()

Messages generated during the process

Specified by:

getMessages in interface FailFastAware

Overrides:

getMessages in class BaseIterativeProcessor<TUPLE_IN,RECORD_OUT,STATE>

Returns:

setMessages

public void setMessages(Collection<String> messages)

Messages generated during the process

Specified by:

setMessages in interface FailFastAware

Overrides:

setMessages in class BaseIterativeProcessor<TUPLE_IN,RECORD_OUT,STATE>

getParams

public Object[] getParams()

Additional Params for the ETL Process

Specified by:

getParams in interface ParamsAware

Overrides:

getParams in class BaseIterativeProcessor<TUPLE_IN,RECORD_OUT,STATE>

setParams

public void setParams(Object[] params)

Additional Params for the ETL Process

Specified by:

setParams in interface ParamsAware

Overrides:

setParams in class BaseIterativeProcessor<TUPLE_IN,RECORD_OUT,STATE>

hasNext

public boolean hasNext()

Description copied from interface: RealTimeLineReader

True if there is another record that can be read by the parser, false implies end of parse/read

Specified by:

hasNext in interface RealTimeLineReader<RECORD_OUT>

Specified by:

hasNext in interface Iterator<RECORD_OUT>

Overrides:

hasNext in class BaseIterativeProcessor<TUPLE_IN,RECORD_OUT,STATE>

Returns:

boolean

stop

public void stop()
          throws Exception

Description copied from interface: ControllableProcess

Request graceful Shutdown

Specified by:

stop in interface ControllableProcess<STATE>

Overrides:

stop in class BaseIterativeProcessor<TUPLE_IN,RECORD_OUT,STATE>

Throws:

Exception

pause

public void pause()
           throws Exception

Description copied from interface: ControllableProcess

Pause till resume

Specified by:

pause in interface ControllableProcess<STATE>

Overrides:

pause in class BaseIterativeProcessor<TUPLE_IN,RECORD_OUT,STATE>

Throws:

Exception

isPaused

public boolean isPaused()

Specified by:

isPaused in interface ControllableProcess<STATE>

Overrides:

isPaused in class BaseIterativeProcessor<TUPLE_IN,RECORD_OUT,STATE>

resume

public void resume()
            throws Exception

Specified by:

resume in interface ControllableProcess<STATE>

Overrides:

resume in class BaseIterativeProcessor<TUPLE_IN,RECORD_OUT,STATE>

Throws:

Exception

kill

public void kill()
          throws Exception

Description copied from interface: ControllableProcess

Kill a process

Specified by:

kill in interface ControllableProcess<STATE>

Overrides:

kill in class BaseIterativeProcessor<TUPLE_IN,RECORD_OUT,STATE>

Throws:

Exception

getCurrentRawIndex

public Long getCurrentRawIndex(TUPLE_IN row)

Description copied from class: BaseIterativeProcessor

Get the raw index/position on the source; BaseIterativeProcessor.getCurrentIndex() only tracks the relevant position. This method allows us to get the position on the source.
In many cases/implementations this maybe used internally to derive the BaseIterativeProcessor.getCurrentIndex()

Specified by:

getCurrentRawIndex in class BaseIterativeProcessor<TUPLE_IN,RECORD_OUT,STATE>

Returns:

Long

getProcesses

public List<BaseIterativeProcessor> getProcesses()

setProcesses

public void setProcesses(List<BaseIterativeProcessor> processes)