Custom Adapter

Pipelite exposes a small, open SPI that allows custom adapters to be created for any external system. An adapter author implements a handful of contracts and is handed a few handles by the framework; everything around them - the queue between a source and its flow, the workers, the backpressure, the durable inbox, the error handling, the threads - belongs to the framework and is not part of the SPI.

Contract What it is for

Contract	What it is for
`ChannelAdapter`	The entry point, one per protocol (`my-adapter://…`): creates an `Endpoint` for every URL.
`Endpoint`	A resolved URL. It creates the `Source` that feeds a flow (`fromSource`) and the `Producer` that writes to it (`toSink` or a route destination). Either side may be left unsupported.
`Source`	What feeds a flow: an `EventSource` that pushes exchanges, or a `PollingSource` the framework polls on a schedule.
`Producer`	What writes one exchange to the external system.
`ChannelConfigurer`, `SourceConfigurer`, `SinkConfigurer`	Optional typed configuration, adapter-wide and per `fromSource` and `toSink`.
`ContextEventListener`	Optional: told when the context starts and stops.

ChannelAdapter

The entry point, one per protocol (my-adapter://…): creates an Endpoint for every URL.

Endpoint

A resolved URL. It creates the Source that feeds a flow (fromSource) and the Producer that writes to it (toSink or a route destination). Either side may be left unsupported.

Source

What feeds a flow: an EventSource that pushes exchanges, or a PollingSource the framework polls on a schedule.

Producer

What writes one exchange to the external system.

ChannelConfigurer, SourceConfigurer, SinkConfigurer

Optional typed configuration, adapter-wide and per fromSource and toSink.

ContextEventListener

Optional: told when the context starts and stops.

SPI dependency

<dependency>
    <groupId>io.pipelite</groupId>
    <artifactId>pipelite-spi</artifactId>
    <version>1.0.0-SNAPSHOT</version>
</dependency>

pipelite-spi depends only on pipelite-dsl, for the Exchange and Headers interfaces an adapter reads and writes. An adapter never needs pipelite-core, and cannot use its classes.

Implementing ChannelAdapter

public class MyCustomChannelAdapter implements ChannelAdapter {

    @Override
    public Endpoint createEndpoint(String url) {
        return new MyCustomEndpoint(EndpointURL.parse(url));
    }
}

Implementing Endpoint

An Endpoint has an EndpointURL, and creates the two sides. Both are optional: the defaults throw an UnsupportedOperationException saying the endpoint "cannot be used as a source" or "as a destination", so an adapter overrides only the side it supports. A flow that reads from an endpoint with no source fails when the context starts, not later.

public class MyCustomEndpoint implements Endpoint {

    private final EndpointURL endpointURL;

    public MyCustomEndpoint(EndpointURL endpointURL) {
        this.endpointURL = endpointURL;
    }

    @Override
    public EndpointURL getEndpointURL() {
        return endpointURL;
    }

    @Override
    public Source createSource() {
        return new MyCustomSource(this);        // for use in fromSource(...)
    }

    @Override
    public Producer createProducer() {
        return new MyCustomProducer(this);      // for use in toSink(...)
    }
}

getProperties() returns the query-string parameters of the URL (my-adapter://orders?batchSize=10).

Implementing a Source

A source only receives. It is handed a SourceContext when it starts, and hands every incoming exchange to the framework. It does not create threads for the flow, queue anything, or call the flow itself.

SourceContext

`SourceContext`
`newExchange(Object payload)`	A new `Exchange` whose input payload is `payload`.
`newExchange(Object payload, Map<String, String> transportHeaders)`	A new `Exchange` whose input payload is `payload`, with the headers of what the source received (the headers of a request or of a record, and what else the protocol says about it that a flow may want: the method, the topic, the offset…). Hand over all of them, with the names the protocol gives them, as text: the source does not choose. The framework puts on the exchange only the ones the flow declared with `propagateHeaders(…)`, so every adapter behaves the same way.
`accept(Exchange)`	Hands an exchange to the framework. It returns once the exchange is accepted: written to the durable inbox and queued, applying the backpressure of the flow. It throws when the exchange cannot be accepted, and the source must then not acknowledge its input.
`flowName()`	The name of the flow the source feeds.
`threadFactory()`	A factory of threads named after the flow, for a source that needs one of its own.
`isRunning()`	`false` once the flow is stopping; a source thread must then end.

newExchange(Object payload)

A new Exchange whose input payload is payload.

newExchange(Object payload, Map<String, String> transportHeaders)

A new Exchange whose input payload is payload, with the headers of what the source received (the headers of a request or of a record, and what else the protocol says about it that a flow may want: the method, the topic, the offset…). Hand over all of them, with the names the protocol gives them, as text: the source does not choose. The framework puts on the exchange only the ones the flow declared with propagateHeaders(…), so every adapter behaves the same way.

accept(Exchange)

Hands an exchange to the framework. It returns once the exchange is accepted: written to the durable inbox and queued, applying the backpressure of the flow. It throws when the exchange cannot be accepted, and the source must then not acknowledge its input.

flowName()

The name of the flow the source feeds.

threadFactory()

A factory of threads named after the flow, for a source that needs one of its own.

isRunning()

false once the flow is stopping; a source thread must then end.

An event source

An EventSource pushes: it receives on its own and calls accept for each exchange. start must return promptly; a source that needs a thread of its own creates it with threadFactory().

public class MyCustomSource implements EventSource {

    private final MyCustomEndpoint endpoint;
    private SourceContext context;
    private MySubscription subscription;

    public MyCustomSource(MyCustomEndpoint endpoint) {
        this.endpoint = endpoint;
    }

    @Override
    public void start(SourceContext context) {
        this.context = context;
        this.subscription = mySystem.subscribe(endpoint.getEndpointURL().getResource(), this::onMessage);
    }

    private void onMessage(MyMessage message) {
        final Exchange exchange = context.newExchange(message.body());
        exchange.putHeader("my-adapter.key", message.key());
        context.accept(exchange);                   (1)
        message.acknowledge();                      (2)
    }

    @Override
    public void stop() {
        subscription.cancel();                      (3)
    }
}

1	Returns once the exchange is durable and queued. If it throws, the exchange was not accepted.
2	Acknowledge the external system (commit the offset, answer the request) only after `accept` returned. The Kafka source commits the offsets of a batch once all of it has been accepted; the HTTP source answers `500` to a request that could not be accepted.
3	Called before the framework stops its own machinery.

A source with a loop of its own (a poll-loop thread, as the Kafka source has) creates the thread with context.threadFactory() and keeps running while context.isRunning().

A polling source

A PollingSource is called by the framework on a schedule: the initialDelay, period, timeUnit and batchSize parameters of the endpoint (see File and Time). poll() returns the next exchange, or null when there is nothing to read now; it is called up to batchSize times per tick, until it returns null.

public class MyCustomPollingSource implements PollingSource {

    private SourceContext context;

    @Override
    public void start(SourceContext context) {      // optional
        this.context = context;
    }

    @Override
    public Exchange poll() {
        final Object data = mySystem.fetchNext();
        return data != null ? context.newExchange(data) : null;
    }

    @Override
    public void stop() {                            // optional
        mySystem.close();
    }
}

A polling source that reads several records at a time, and must not move its own position (a file offset) before they are safe, hands each of them over with accept, which makes it durable, and returns null: the framework then takes the exchanges from its own queue. The file source works this way.

Implementing a Producer

A Producer writes one exchange. It is a functional interface:

public class MyCustomProducer implements Producer {

    private final MyClient client;

    public MyCustomProducer(MyCustomEndpoint endpoint) {
        this.client = MyClient.connect(endpoint.getEndpointURL().getResource());
    }

    @Override
    public void produce(Exchange exchange) {
        client.send(exchange.getInputPayload());        (1)
    }

    @Override
    public void close() {                               (2)
        client.close();
    }
}

1	A `RuntimeException` thrown here is a failure of the step: the framework applies the flow’s error handling around `produce`, so it is retried, dead-lettered or handled like the failure of any processor.
2	Optional (the default does nothing). The framework calls it once the producer is no longer needed: when the context stops, for a producer that ends a flow; right after the exchange is delivered, for one created to deliver a single exchange (for example by `supplyExchange`).

A protocol that has headers (the headers of a record, of a request) overrides produce(Exchange, Map<String, String> headers) too. The framework calls this method, never produce(Exchange), with the headers the flow chose with propagateHeaders(…) on its sink, already filtered and as text: the producer sends them in the form of its protocol and does not choose. The default of the method ignores them and calls produce(Exchange), so a producer of a protocol without headers needs nothing.

Typed configuration

An adapter that needs configuration offers it in two places, both optional:

Adapter-wide, shared by every endpoint of the protocol (the Kafka adapter’s bootstrap servers): implement ChannelConfigurer<T> and return its type from ChannelAdapter#getChannelConfigurerType(); the framework passes the configurers registered with addChannelConfigurer(…) to ChannelAdapter#configure(…).
Per fromSource(…), as the typed alternative to a query string (fromSource("my-adapter://orders", (MySourceConfigurer c) → c.batchSize(10))): extend SourceConfigurer (or PollingSourceConfigurer, which already carries initialDelay, period, timeUnit and batchSize), contribute the query parameters in contributeQueryParameters, and return a fresh instance from ChannelAdapter#newSourceConfigurer(). An adapter may pre-seed defaults there: the time adapter turns the durable inbox off that way.
Per toSink(…): implement SinkConfigurer (a marker interface, a sink has no base class), and return a fresh instance from ChannelAdapter#newSinkConfigurer(). A sink with no settings returns nothing, and a flow that gives it a configurer fails when the context starts.

Capabilities

What a source or a sink can be configured with that is not specific to one adapter is a capability: an interface that a concrete configurer implements, as many as its endpoint supports. The framework does not know any adapter, it asks each configurer which capabilities it has. Today there is one:

PropagatesHeaders: The endpoint has headers, so the flow can choose which of them cross the border with propagateHeaders(…). Implement it in the configurer of a protocol that has headers (the headers of a request, of a record) and keep the value in a field, through the two accessors the interface asks for; a protocol without headers does not implement it, and so a flow cannot declare them. The source passes all the headers to SourceContext#newExchange and the producer overrides produce(Exchange, Map<String, String>), as described above: the filter is the framework’s.

public final class MySinkConfigurer implements SinkConfigurer, PropagatesHeaders {

    private HeaderPropagation headerPropagation = HeaderPropagation.none();

    @Override
    public HeaderPropagation headerPropagation() {
        return headerPropagation;
    }

    @Override
    public void setHeaderPropagation(HeaderPropagation propagation) {
        this.headerPropagation = propagation;
    }
}

Registration via classpath scan

Create the registration file in the resources directory of your module:

src/main/resources/META-INF/pipelite.factories

File content: a standard java.util.Properties file, one <protocol>=<fully-qualified-class-name> entry per adapter (MyCustomChannelAdapter itself declares nothing about its own protocol - the key here is what defines it):

my-adapter=com.example.adapters.MyCustomChannelAdapter

The context scans every META-INF/pipelite.factories on the classpath and instantiates one adapter per entry at startup, failing fast with a DuplicateChannelAdapterProtocolException if two different classes claim the same protocol name.

queue:// is not an adapter you can replace or extend: it is a built-in of pipelite-core (see Queue).

Integration with the context lifecycle

If your adapter needs to be notified of context startup and shutdown, implement ContextEventListener:

public class MyCustomChannelAdapter
    implements ChannelAdapter, ContextEventListener {

    @Override
    public void onContextStarted() {
        // start connection to the external system
    }

    @Override
    public void onContextStopped() {
        // graceful connection shutdown
    }
}

Testing an adapter

An adapter module depends on pipelite-spi and pipelite-dsl, not on pipelite-core, so its unit tests do not have the framework’s Exchange implementation. A source is tested with a hand-written SourceContext whose newExchange and accept record what the source produced, and a producer with a small Exchange stub holding a payload and a map of headers. To test the adapter inside a running flow, add pipelite-core as a test dependency of an integration-test module.