Implementing the Endpoint Interface

Alternative ways of implementing an endpoint

The following alternative endpoint implementation patterns are supported:

Event-driven endpoint implementation

If your custom endpoint conforms to the event-driven pattern (see Consumer Patterns and Threading), it is implemented by extending the abstract class, org.apache.camel.impl.DefaultEndpoint, as shown in Example 6.2.

Example 6.2. Implementing DefaultEndpoint

import java.util.Map;
import java.util.concurrent.BlockingQueue;

import org.apache.camel.Component;
import org.apache.camel.Consumer;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
import org.apache.camel.impl.DefaultEndpoint;

public class CustomEndpoint extends DefaultEndpoint<CustomExchange> { 

    public CustomEndpoint(String endpointUri, Component component) { 
        super(endpointUri, component);
        // Do any other initialization...
    }

    public Producer createProducer() throws Exception { 
        return new CustomProducer(this);
    }

    public Consumer createConsumer(Processor processor) throws Exception { 
        return new CustomConsumer(this, processor);
    }

    public boolean isSingleton() { 
        return true;
    }

    // Implement the following two methods, only if you need a custom exchange class.
    //
    public CustomExchange createExchange() { 
        return new CustomExchange(getCamelContext(), getExchangePattern());
    }

    public CustomExchange createExchange(ExchangePattern pattern) {
        return new CustomExchange(getCamelContext(), pattern);
    }
}

Implement an event-driven custom endpoint, CustomEndpoint, by extending the DefaultEndpoint class.

You must have at least one constructor that takes the endpoint URI, endpointUri, and the parent component reference, component, as arguments.

Implement the createProducer() factory method to create producer endpoints.

Implement the createConsumer() factory method to create event-driven consumer instances.

	Important
	Do not override the `createPollingConsumer()` method.

If you intend to customize the exchange implementation, you should override the createExchange() and the createExchange(ExchangePattern) methods, to ensure that the correct exchange type is created. If you do not override these methods, the implementations inherited from DefaultEndpoint will create a DefaultExchange instance.

The DefaultEndpoint class provides default implementations of the following methods, which you might find useful when writing your custom endpoint code:

getEndpointUri()—Returns the endpoint URI.
getCamelContext()—Returns a reference to the CamelContext.
getComponent()—Returns a reference to the parent component.
getExecutorService()—Returns a reference to a scheduled executor service. The scheduled executor is a java.util.concurrent.ScheduledExecutorService object.
createPollingConsumer()—Creates a polling consumer. The created polling consumer's functionality is based on the event-driven consumer. If you override the event-driven consumer method, createConsumer(), you get a polling consumer implementation for free.
createExchange(Exchange e)—Converts the given exchange object, e, to the type required for this endpoint. This method creates a new endpoint using the overridden createExchange() endpoints. This ensures that the method also works for custom exchange types.

Scheduled poll endpoint implementation

If your custom endpoint conforms to the scheduled poll pattern (see Consumer Patterns and Threading) it is implemented by inheriting from the abstract class, org.apache.camel.impl.ScheduledPollEndpoint, as shown in Example 6.3.

Example 6.3. ScheduledPollEndpoint Implementation

import org.apache.camel.Consumer;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
import org.apache.camel.ExchangePattern;
import org.apache.camel.Message;
import org.apache.camel.impl.ScheduledPollEndpoint;

public class CustomEndpoint extends ScheduledPollEndpoint<CustomExchange> { 

    protected CustomEndpoint(String endpointUri, CustomComponent component) { 
        super(endpointUri, component);
        // Do any other initialization...
    }

    public Producer<CustomExchange> createProducer() throws Exception { 
        Producer<CustomExchange> result = new CustomProducer(this);
        return result;
    }

    public Consumer<CustomExchange> createConsumer(Processor processor) throws Exception { 
        Consumer<CustomExchange> result = new CustomConsumer(this, processor);
        configureConsumer(result); 
        return result;
    }

    public boolean isSingleton() {
        return true;
    }

    // Implement the following two methods, only if you need a custom exchange class.
    //
    public CustomExchange createExchange() { 
        return new CustomExchange(...);
    }

    public CustomExchange createExchange(ExchangePattern pattern) {
        return new CustomExchange(getCamelContext(), pattern);
    }
}

Implement a scheduled poll custom endpoint, CustomEndpoint, by extending the ScheduledPollEndpoint class.

You must to have at least one constructor that takes the endpoint URI, endpointUri, and the parent component reference, component, as arguments.

Implement the createProducer() factory method to create a producer endpoint.

Implement the createConsumer() factory method to create a scheduled poll consumer instance.

	Important
	Do not override the `createPollingConsumer()` method.

The configureConsumer() method, defined in the ScheduledPollEndpoint base class, is responsible for injecting consumer query options into the consumer. See Consumer parameter injection.

Polling endpoint implementation

If your custom endpoint conforms to the polling consumer pattern (see Consumer Patterns and Threading), it is implemented by inheriting from the abstract class, org.apache.camel.impl.DefaultPollingEndpoint, as shown in Example 6.4.

Example 6.4. DefaultPollingEndpoint Implementation

import org.apache.camel.Consumer;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
import org.apache.camel.ExchangePattern;
import org.apache.camel.Message;
import org.apache.camel.impl.DefaultPollingEndpoint;

public class CustomEndpoint extends DefaultPollingEndpoint<CustomExchange> {
    ...
    public PollingConsumer<CustomExchange> createPollingConsumer() throws Exception {
        PollingConsumer<CustomExchange> result = new CustomConsumer(this);
        configureConsumer(result);
        return result;
    }

    // Do NOT implement createConsumer(). It is already implemented in DefaultPollingEndpoint.
    ...
}

Because this CustomEndpoint class is a polling endpoint, you must implement the createPollingConsumer() method instead of the createConsumer() method. The consumer instance returned from createPollingConsumer() must inherit from the PollingConsumer interface. For details of how to implement a polling consumer, see Polling consumer implementation.

Apart from the implementation of the createPollingConsumer() method, the steps for implementing a DefaultPollingEndpoint are similar to the steps for implementing a ScheduledPollEndpoint. See Example 6.3 for details.

Implementing the BrowsableEndpoint interface

If you want to expose the list of exchange instances that are pending in the current endpoint, you can implement the org.apache.camel.spi.BrowsableEndpoint interface, as shown in Example 6.5. It makes sense to implement this interface if the endpoint performs some sort of buffering of incoming events. For example, the FUSE Mediation Router SEDA endpoint implements the BrowsableEndpoint interface—see Example 6.6.

Example 6.5. BrowsableEndpoint Interface

package org.apache.camel.spi;

import java.util.List;

import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;

public interface BrowsableEndpoint<T extends Exchange> extends Endpoint<T> {
    List<Exchange> getExchanges();
}

Example

Example 6.6 shows the implementation of SedaEndpoint, which is taken from the FUSE Mediation Router SEDA component implementation. The SEDA endpoint is an example of an event-driven endpoint. Incoming events are stored in a FIFO queue (an instance of java.util.concurrent.BlockingQueue) and a SEDA consumer starts up a thread to read and process the events. The events themselves are represented by org.apache.camel.Exchange objects.

Example 6.6. SedaEndpoint Implementation

package org.apache.camel.component.seda;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;

import org.apache.camel.Component;
import org.apache.camel.Consumer;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
import org.apache.camel.impl.DefaultEndpoint;
import org.apache.camel.spi.BrowsableEndpoint;

public class SedaEndpoint extends DefaultEndpoint<Exchange> implements BrowsableEndpoint<Exchange> { 
    private BlockingQueue<Exchange> queue;

    public SedaEndpoint(String endpointUri, Component component, BlockingQueue<Exchange> queue) { 
        super(endpointUri, component);
        this.queue = queue;
    }

    public SedaEndpoint(String uri, SedaComponent component, Map parameters) { 
        this(uri, component, component.createQueue(uri, parameters));
    }

    public Producer createProducer() throws Exception { 
        return new CollectionProducer(this, getQueue());
    }

    public Consumer createConsumer(Processor processor) throws Exception { 
        return new SedaConsumer(this, processor);
    }

    public BlockingQueue<Exchange> getQueue() { 
        return queue;
    }

    public boolean isSingleton() { 
        return true;
    }

    public List<Exchange> getExchanges() { 
        return new ArrayList<Exchange>(getQueue());
    }
}

	The `SedaEndpoint` class follows the pattern for implementing an event-driven endpoint by extending the `DefaultEndpoint` class. The `SedaEndpoint` class also implements the `BrowsableEndpoint` interface, which provides access to the list of exchange objects in the queue.
	Following the usual pattern for an event-driven consumer, `SedaEndpoint` defines a constructor that takes an endpoint argument, `endpointUri`, and a component reference argument, `component`.
	Another constructor is provided, which delegates queue creation to the parent component instance.
	The `createProducer()` factory method creates an instance of `CollectionProducer`, which is a producer implementation that adds events to the queue.
	The `createConsumer()` factory method creates an instance of `SedaConsumer`, which is responsible for pulling events off the queue and processing them.
	The `getQueue()` method returns a reference to the queue.
	The `isSingleton()` method returns `true`, indicating that a single endpoint instance should be created for each unique URI string.
	The `getExchanges()` method implements the corresponding abstract method from `BrowsableEndpoint`.