Chapter 50. Consumer Interface
Abstract
Consumer
interface, which is an essential step in the implementation of a Apache Camel component.
50.1. The Consumer Interface
Overview
org.apache.camel.Consumer
type represents a source endpoint in a route. There are several different ways of implementing a consumer (see Section 47.1.3, “Consumer Patterns and Threading”), and this degree of flexibility is reflected in the inheritance hierarchy ( see Figure 50.1, “Consumer Inheritance Hierarchy”), which includes several different base classes for implementing a consumer.
Figure 50.1. Consumer Inheritance Hierarchy

Consumer parameter injection
custom
prefix:
custom:destination?consumer.myConsumerParam
consumer.*
. For the consumer.myConsumerParam
parameter, you need to define corresponding setter and getter methods on the Consumer
implementation class as follows:
public class CustomConsumer extends ScheduledPollConsumer { ... String getMyConsumerParam() { ... } void setMyConsumerParam(String s) { ... } ... }
configureConsumer()
method in the implementation of Endpoint.createConsumer()
. See the section called “Scheduled poll endpoint implementation”). Example 50.1, “FileEndpoint createConsumer() Implementation” shows an example of a createConsumer()
method implementation, taken from the FileEndpoint
class in the file component:
Example 50.1. FileEndpoint createConsumer() Implementation
... public class FileEndpoint extends ScheduledPollEndpoint { ... public Consumer createConsumer(Processor processor) throws Exception { Consumer result = new FileConsumer(this, processor); configureConsumer(result); return result; } ... }
- When the endpoint is created, the default implementation of
DefaultComponent.createEndpoint(String uri)
parses the URI to extract the consumer parameters, and stores them in the endpoint instance by callingScheduledPollEndpoint.configureProperties()
. - When
createConsumer()
is called, the method implementation callsconfigureConsumer()
to inject the consumer parameters (see Example 50.1, “FileEndpoint createConsumer() Implementation”). - The
configureConsumer()
method uses Java reflection to call the setter methods whose names match the relevant options after theconsumer.
prefix has been stripped off.
Scheduled poll parameters
Table 50.1. Scheduled Poll Parameters
Name | Default | Description |
---|---|---|
initialDelay | 1000 | Delay, in milliseconds, before the first poll. |
delay | 500 | Depends on the value of the useFixedDelay flag (time unit is milliseconds). |
useFixedDelay | false |
If
false , the delay parameter is interpreted as the polling period. Polls will occur at initialDelay , initialDelay+delay , initialDelay+2*delay , and so on.
If
true , the delay parameter is interpreted as the time elapsed between the previous execution and the next execution. Polls will occur at initialDelay , initialDelay+[ProcessingTime]+delay , and so on. Where ProcessingTime is the time taken to process an exchange object in the current thread.
|
Converting between event-driven and polling consumers
org.apache.camel.impl.EventDrivenPollingConsumer
—Converts an event-driven consumer into a polling consumer instance.org.apache.camel.impl.DefaultScheduledPollConsumer
—Converts a polling consumer into an event-driven consumer instance.
Endpoint
type. The Endpoint
interface defines the following two methods for creating a consumer instance:
package org.apache.camel; public interface Endpoint { ... Consumer createConsumer(Processor processor) throws Exception; PollingConsumer createPollingConsumer() throws Exception; }
createConsumer()
returns an event-driven consumer and createPollingConsumer()
returns a polling consumer. You would only implement one these methods. For example, if you are following the event-driven pattern for your consumer, you would implement the createConsumer()
method provide a method implementation for createPollingConsumer()
that simply raises an exception. With the help of the conversion classes, however, Apache Camel is able to provide a more useful default implementation.
DefaultEndpoint
and implementing the createConsumer()
method. The implementation of createPollingConsumer()
is inherited from DefaultEndpoint
, where it is defined as follows:
public PollingConsumer<E> createPollingConsumer() throws Exception { return new EventDrivenPollingConsumer<E>(this); }
EventDrivenPollingConsumer
constructor takes a reference to the event-driven consumer, this
, effectively wrapping it and converting it into a polling consumer. To implement the conversion, the EventDrivenPollingConsumer
instance buffers incoming events and makes them available on demand through the receive()
, the receive(long timeout)
, and the receiveNoWait()
methods.
DefaultPollingEndpoint
and implementing the createPollingConsumer()
method. In this case, the implementation of the createConsumer()
method is inherited from DefaultPollingEndpoint
, and the default implementation returns a DefaultScheduledPollConsumer
instance (which converts the polling consumer into an event-driven consumer).
ShutdownPrepared interface
org.apache.camel.spi.ShutdownPrepared
interface, which enables your custom consumer endpoint to receive shutdown notifications.
ShutdownPrepared
interface.
Example 50.2. ShutdownPrepared Interface
package org.apache.camel.spi; public interface ShutdownPrepared { void prepareShutdown(boolean forced); }
ShutdownPrepared
interface defines the following methods:
prepareShutdown
- Receives notifications to shut down the consumer endpoint in one or two phases, as follows:
- Graceful shutdown—where the
forced
argument has the valuefalse
. Attempt to clean up resources gracefully. For example, by stopping threads gracefully. - Forced shutdown—where the
forced
argument has the valuetrue
. This means that the shutdown has timed out, so you must clean up resources more aggressively. This is the last chance to clean up resources before the process exits.
ShutdownAware interface
org.apache.camel.spi.ShutdownAware
interface, which interacts with the graceful shutdown mechanism, enabling a consumer to ask for extra time to shut down. This is typically needed for components such as SEDA, which can have pending exchanges stored in an internal queue. Normally, you would want to process all of the exchanges in the queue before shutting down the SEDA consumer.
ShutdownAware
interface.
Example 50.3. ShutdownAware Interface
// Java package org.apache.camel.spi; import org.apache.camel.ShutdownRunningTask; public interface ShutdownAware extends ShutdownPrepared { boolean deferShutdown(ShutdownRunningTask shutdownRunningTask); int getPendingExchangesSize(); }
ShutdownAware
interface defines the following methods:
deferShutdown
- Return
true
from this method, if you want to delay shutdown of the consumer. TheshutdownRunningTask
argument is anenum
which can take either of the following values:ShutdownRunningTask.CompleteCurrentTaskOnly
—finish processing the exchanges that are currently being processed by the consumer's thread pool, but do not attempt to process any more exchanges than that.ShutdownRunningTask.CompleteAllTasks
—process all of the pending exchanges. For example, in the case of the SEDA component, the consumer would process all of the exchanges from its incoming queue.
getPendingExchangesSize
- Indicates how many exchanges remain to be processed by the consumer. A zero value indicates that processing is finished and the consumer can be shut down.
ShutdownAware
methods, see Example 50.7, “Custom Threading Implementation”.