org.apache.camel.component.file
Class GenericFileConsumer<T>

java.lang.Object
  extended by org.apache.camel.impl.ServiceSupport
      extended by org.apache.camel.impl.DefaultConsumer
          extended by org.apache.camel.impl.ScheduledPollConsumer
              extended by org.apache.camel.component.file.GenericFileConsumer<T>
All Implemented Interfaces:
Runnable, BatchConsumer, Consumer, Service, ShutdownAware, SuspendableService
Direct Known Subclasses:
FileConsumer, RemoteFileConsumer

public abstract class GenericFileConsumer<T>
extends ScheduledPollConsumer
implements BatchConsumer, ShutdownAware

Base class for remote file consumers.


Field Summary
protected  GenericFileEndpoint<T> endpoint
           
protected  String fileExpressionResult
           
protected  org.apache.commons.logging.Log log
           
protected  boolean loggedIn
           
protected  int maxMessagesPerPoll
           
protected  GenericFileOperations<T> operations
           
protected  int pendingExchanges
           
protected  ShutdownRunningTask shutdownRunningTask
           
 
Constructor Summary
GenericFileConsumer(GenericFileEndpoint<T> endpoint, Processor processor, GenericFileOperations<T> operations)
           
 
Method Summary
 boolean deferShutdown(ShutdownRunningTask shutdownRunningTask)
          To defer shutdown during first phase of shutdown.
protected  void doStart()
           
 int getPendingExchangesSize()
          Gets the number of pending exchanges.
 boolean isBatchAllowed()
          Whether processing the batch is still allowed.
protected  boolean isInProgress(GenericFile<T> file)
          Is the given file already in progress.
protected  boolean isMatched(GenericFile<T> file, boolean isDirectory)
          Strategy to perform file matching based on endpoint configuration.
protected  boolean isValidFile(GenericFile<T> file, boolean isDirectory)
          Strategy for validating if the given remote file should be included or not
protected  void poll()
          Poll for files
protected abstract  void pollDirectory(String fileName, List<GenericFile<T>> fileList)
          Polls the given directory for files to process
protected  void postPollCheck()
          Override if required.
protected  boolean prePollCheck()
          Override if required.
 void processBatch(Queue<Object> exchanges)
          Processes the list of Exchange in a batch.
protected  void processExchange(Exchange exchange)
          Processes the exchange
 void setMaxMessagesPerPoll(int maxMessagesPerPoll)
          Sets a maximum number of messages as a limit to poll at each polling.
 
Methods inherited from class org.apache.camel.impl.ScheduledPollConsumer
doStop, getDelay, getInitialDelay, getPollStrategy, getTimeUnit, isPollAllowed, isSuspended, isUseFixedDelay, resume, run, setDelay, setInitialDelay, setPollStrategy, setTimeUnit, setUseFixedDelay, suspend
 
Methods inherited from class org.apache.camel.impl.DefaultConsumer
getEndpoint, getExceptionHandler, getProcessor, handleException, setExceptionHandler, toString
 
Methods inherited from class org.apache.camel.impl.ServiceSupport
addChildService, getStatus, getVersion, isRunAllowed, isStarted, isStarting, isStopped, isStopping, removeChildService, start, stop
 
Methods inherited from class java.lang.Object
clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait
 
Methods inherited from interface org.apache.camel.Consumer
getEndpoint
 
Methods inherited from interface org.apache.camel.Service
start, stop
 

Field Detail

log

protected final transient org.apache.commons.logging.Log log

endpoint

protected GenericFileEndpoint<T> endpoint

operations

protected GenericFileOperations<T> operations

loggedIn

protected boolean loggedIn

fileExpressionResult

protected String fileExpressionResult

maxMessagesPerPoll

protected int maxMessagesPerPoll

shutdownRunningTask

protected volatile ShutdownRunningTask shutdownRunningTask

pendingExchanges

protected volatile int pendingExchanges
Constructor Detail

GenericFileConsumer

public GenericFileConsumer(GenericFileEndpoint<T> endpoint,
                           Processor processor,
                           GenericFileOperations<T> operations)
Method Detail

poll

protected void poll()
             throws Exception
Poll for files

Specified by:
poll in class ScheduledPollConsumer
Throws:
Exception - can be thrown if an exception occurred during polling

setMaxMessagesPerPoll

public void setMaxMessagesPerPoll(int maxMessagesPerPoll)
Description copied from interface: BatchConsumer
Sets a maximum number of messages as a limit to poll at each polling.

Can be used to limit eg to 100 to avoid when starting and there are millions of messages for you in the first poll.

Is default unlimited, but use 0 or negative number to disable it as unlimited.

Specified by:
setMaxMessagesPerPoll in interface BatchConsumer
Parameters:
maxMessagesPerPoll - maximum messages to poll.

processBatch

public void processBatch(Queue<Object> exchanges)
Description copied from interface: BatchConsumer
Processes the list of Exchange in a batch.

Each message exchange will be processed individually but the batch consumer will add properties with the current index and total in the batch. The items in the Queue may actually be Holder objects that store other data alongside the Exchange.

Specified by:
processBatch in interface BatchConsumer
Parameters:
exchanges - list of items in this batch

deferShutdown

public boolean deferShutdown(ShutdownRunningTask shutdownRunningTask)
Description copied from interface: ShutdownAware
To defer shutdown during first phase of shutdown. This allows any pending exchanges to be completed and therefore ensure a graceful shutdown without loosing messages. At the very end when there are no more inflight and pending messages the consumer could then safely be shutdown.

This is needed by SedaConsumer.

Specified by:
deferShutdown in interface ShutdownAware
Parameters:
shutdownRunningTask - the configured option for how to act when shutting down running tasks.
Returns:
true to defer shutdown to very last.

getPendingExchangesSize

public int getPendingExchangesSize()
Description copied from interface: ShutdownAware
Gets the number of pending exchanges.

Some consumers has internal queues with Exchange which are pending. For example the SedaConsumer.

Return zero to indicate no pending exchanges and therefore ready to shutdown.

Specified by:
getPendingExchangesSize in interface ShutdownAware
Returns:
number of pending exchanges

isBatchAllowed

public boolean isBatchAllowed()
Description copied from interface: BatchConsumer
Whether processing the batch is still allowed.

This is used during shutdown to help indicate whether to complete the pending exchanges or stop after current exchange has been processed.

Specified by:
isBatchAllowed in interface BatchConsumer
Returns:
true to continue processing from the batch, or false to stop.
See Also:
ShutdownRunningTask

prePollCheck

protected boolean prePollCheck()
                        throws Exception
Override if required. Perform some checks (and perhaps actions) before we poll.

Returns:
true to poll, false to skip this poll.
Throws:
Exception

postPollCheck

protected void postPollCheck()
Override if required. Perform some checks (and perhaps actions) after we have polled.


pollDirectory

protected abstract void pollDirectory(String fileName,
                                      List<GenericFile<T>> fileList)
Polls the given directory for files to process

Parameters:
fileName - current directory or file
fileList - current list of files gathered

processExchange

protected void processExchange(Exchange exchange)
Processes the exchange

Parameters:
exchange - the exchange

isValidFile

protected boolean isValidFile(GenericFile<T> file,
                              boolean isDirectory)
Strategy for validating if the given remote file should be included or not

Parameters:
file - the remote file
isDirectory - whether the file is a directory or a file
Returns:
true to include the file, false to skip it

isMatched

protected boolean isMatched(GenericFile<T> file,
                            boolean isDirectory)
Strategy to perform file matching based on endpoint configuration.

Will always return false for certain files/folders:

And then true for directories.

Parameters:
file - the file
isDirectory - whether the file is a directory or a file
Returns:
true if the remote file is matched, false if not

isInProgress

protected boolean isInProgress(GenericFile<T> file)
Is the given file already in progress.

Parameters:
file - the file
Returns:
true if the file is already in progress

doStart

protected void doStart()
                throws Exception
Overrides:
doStart in class ScheduledPollConsumer
Throws:
Exception


Copyright © 2007-2010 IONA Open Source Community. All Rights Reserved.